Create app.py

app.py

@@ -0,0 +1,220 @@
+import gradio as gr
+import json
+import requests
+from typing import List, Optional
+from pydantic import BaseModel, Field
+from together import Together
+
+# Initialize Together API
+together = Together(api_key=os.getenv("togetherai"))
+
+# Define the schemas for different API operations
+class OrderItem(BaseModel):
+    item_id: str
+    quantity: int
+
+class CreateOrderExtract(BaseModel):
+    user_id: Optional[str] = Field(None, description="The user's ID if provided it can be a name as well")
+    items: List[OrderItem] = Field(..., description="List of items ordered")
+
+class CancelOrderExtract(BaseModel):
+    order_id: str = Field(..., description="The ID of the order to cancel can be referenced as order id or order or id or order number")
+
+class CheckOrderStatusExtract(BaseModel):
+    order_id: str = Field(..., description="The ID of the order to check can be referenced as order id or order or id or order number")
+
+class CreateInvoiceExtract(BaseModel):
+    order_id: str = Field(..., description="The ID of the order for which to create an invoice")
+    amount: float = Field(..., description="The amount of the invoice")
+
+class GetInvoiceDetailsExtract(BaseModel):
+    invoice_id: str = Field(..., description="The ID of the invoice to get details for")
+
+class CreatePaymentExtract(BaseModel):
+    invoice_id: str = Field(..., description="The ID of the invoice to pay")
+    order_id: str = Field(..., description="The ID of the order associated with the payment")
+    amount: float = Field(..., description="The amount of the payment")
+
+# Function to classify user message
+def classify_message(message: str) -> str:
+    classify = together.chat.completions.create(
+        messages=[
+            {
+                "role": "system",
+                "content": "Classify the following message into one of these categories: create_order, cancel_order, check_order_status, create_invoice, get_invoice_details, create_payment. Respond only with the category name.",
+            },
+            {
+                "role": "user",
+                "content": message,
+            },
+        ],
+        model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
+    )
+    return classify.choices[0].message.content.strip()
+
+# Function to extract information based on classification
+def extract_info(message: str, classification: str) -> dict:
+    schema_map = {
+        "create_order": CreateOrderExtract,
+        "cancel_order": CancelOrderExtract,
+        "check_order_status": CheckOrderStatusExtract,
+        "create_invoice": CreateInvoiceExtract,
+        "get_invoice_details": GetInvoiceDetailsExtract,
+        "create_payment": CreatePaymentExtract,
+    }
+    
+    schema = schema_map[classification]
+    
+    extract = together.chat.completions.create(
+        messages=[
+            {
+                "role": "system",
+                "content": f"Extract {classification} information from the following message. Respond only in JSON.",
+            },
+            {
+                "role": "user",
+                "content": message,
+            },
+        ],
+        model="mistralai/Mixtral-8x7B-Instruct-v0.1",
+        response_format={
+            "type": "json_object",
+            "schema": schema.model_json_schema(),
+        },
+    )
+    return json.loads(extract.choices[0].message.content)
+
+# Function to make API call
+def make_api_call(classification: str, info: dict) -> dict:
+    base_url = os.getenv("baseurl")
+    
+    api_map = {
+        "create_order": ("POST", f"{base_url}/orders"),
+        "cancel_order": ("POST", f"{base_url}/orders/{{order_id}}/cancel"),
+        "check_order_status": ("GET", f"{base_url}/orders/{{order_id}}/status"),
+        "create_invoice": ("POST", f"{base_url}/invoices"),
+        "get_invoice_details": ("GET", f"{base_url}/invoices/{{invoice_id}}"),
+        "create_payment": ("POST", f"{base_url}/payments"),
+    }
+    
+    method, url_template = api_map[classification]
+    
+    # Replace placeholders in the URL if necessary
+    try:
+        url = url_template.format(**info)
+    except KeyError as e:
+        return {"error": f"Missing required information: {str(e)}"}
+    
+    try:
+        if method == "GET":
+            response = requests.get(url)
+        elif method == "POST":
+            response = requests.post(url, json=info)
+        response.raise_for_status()
+        return response.json()
+    except requests.RequestException as e:
+        return {"error": f"API request failed: {str(e)}"}
+
+# Function to interpret API response
+def interpret_response(user_message: str, classification: str, api_response: dict) -> str:
+    interpret = together.chat.completions.create(
+        messages=[
+            {
+                "role": "system",
+                "content": "Process user query and automated system API response to form a coherent natural reply that will help the user affirm the situation of their request. Your response will be streamed directly to the user. Do not include the fact that you are reading the API response. It should be natural and helpful. Make sure to give the user the relevant IDs and information, and do not talk more than that. Just give information and that's it.",
+            },
+            {
+                "role": "user",
+                "content": f"User message: {user_message}\nRequest type: {classification}\nAPI response: {json.dumps(api_response)}",
+            },
+        ],
+        model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
+    )
+    return interpret.choices[0].message.content
+
+# Main function to process the user request
+def process_request(user_message: str) -> str:
+    # Classify the message
+    classification = classify_message(user_message)
+    
+    # Extract information based on classification
+    info = extract_info(user_message, classification)
+    
+    # Make API call
+    api_response = make_api_call(classification, info)
+    
+    # Interpret the response
+    interpretation = interpret_response(user_message, classification, api_response)
+    
+    return interpretation
+
+# Define the API documentation with the repository link
+api_documentation = """
+## Available API Methods
+
+### 1. Create Order
+- **Endpoint:** POST /orders
+- **Inputs:**
+  - user_id: (Optional) The user's ID or name.
+  - items: List of ordered items with `item_id` and `quantity`.
+
+### 2. Cancel Order
+- **Endpoint:** POST /orders/{order_id}/cancel
+- **Inputs:**
+  - order_id: The ID of the order to be canceled.
+
+### 3. Check Order Status
+- **Endpoint:** GET /orders/{order_id}/status
+- **Inputs:**
+  - order_id: The ID of the order to check status.
+
+### 4. Create Invoice
+- **Endpoint:** POST /invoices
+- **Inputs:**
+  - order_id: The ID of the order.
+  - amount: The invoice amount.
+
+### 5. Get Invoice Details
+- **Endpoint:** GET /invoices/{invoice_id}
+- **Inputs:**
+  - invoice_id: The ID of the invoice to retrieve.
+
+### 6. Create Payment
+- **Endpoint:** POST /payments
+- **Inputs:**
+  - invoice_id: The ID of the invoice to pay.
+  - order_id: The ID of the associated order.
+  - amount: The payment amount.
+
+You can use these examples when interacting with the system.
+
+For more details on the API, including the routes and implementation, visit the GitHub repository:
+[Order Management API Repository](https://github.com/akash-mondal/order-management-api)
+"""
+
+# Build the Gradio interface
+with gr.Blocks() as demo:
+    gr.Markdown("# Function Calling Demo")
+    gr.Markdown("""
+    This is a demo for performing function calls using custom API endpoints. The demo uses Together AI for language models (LLMs), where:
+    
+    - **Mixtral 8x7B** is used to extract API information from natural language.
+    - **Llama 3.1 70B** is used to classify which API to route to.
+    - **Llama 3.1 8B** is used to interpret API responses into natural language for the end user.
+    """)
+
+    with gr.Tab("User Input"):
+        user_input = gr.Textbox(label="Enter your message")
+        output = gr.Textbox(label="Response")
+        submit_button = gr.Button("Submit")
+        
+        def handle_submit(user_message):
+            return process_request(user_message)
+        
+        submit_button.click(handle_submit, inputs=user_input, outputs=output)
+    
+    with gr.Tab("API Documentation"):
+        gr.Markdown(api_documentation)
+
+# Run the demo
+demo.launch()