Update app.py

app.py

@@ -1,265 +1,71 @@
+```python
 import os
+import sqlite3
+import requests
+import openai
 import gradio as gr
-import pandas as pd
-import json
-from dotenv import load_dotenv
-import httpx
 
-# Load environment variables
-load_dotenv()
+# Load API keys from environment
+openai.api_key = os.getenv("OPENAI_API_KEY")
 
-# Get OpenAI API key
-api_key = os.getenv("OPENAI_API_KEY")
+# --- Agents ---
+def db_agent(query: str) -> str:
+    conn = sqlite3.connect("shop.db")
+    cur = conn.cursor()
+    if "max revenue" in query.lower():
+        cur.execute(
+            """
+            SELECT product, SUM(amount) AS revenue
+            FROM transactions
+            WHERE date = date('now')
+            GROUP BY product
+            ORDER BY revenue DESC
+            LIMIT 1
+            """
+        )
+        row = cur.fetchone()
+        return f"Top product today: {row[0]} with ₹{row[1]:,.2f}" if row else "No transactions found for today."
+    return None
 
-# Helper function for OpenAI API calls
-def openai_chat_completion(messages, temperature=0, response_format=None, max_tokens=None):
-    """Make a direct API call to OpenAI without using the client library"""
-    url = "https://api.openai.com/v1/chat/completions"
-    
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {api_key}"
-    }
-    
-    payload = {
-        "model": "gpt-3.5-turbo",
-        "messages": messages,
-        "temperature": temperature
-    }
-    
-    if response_format:
-        payload["response_format"] = response_format
-        
-    if max_tokens:
-        payload["max_tokens"] = max_tokens
-    
-    response = httpx.post(url, json=payload, headers=headers)
-    
-    if response.status_code != 200:
-        raise Exception(f"OpenAI API error: {response.status_code} - {response.text}")
-        
-    return response.json()
 
-# Simple database using pandas DataFrames
-class SimpleDatabase:
-    def __init__(self):
-        # Sample product data
-        self.products = pd.DataFrame({
-            'product_id': [1, 2, 3, 4, 5],
-            'name': ['Laptop', 'Smartphone', 'Headphones', 'Monitor', 'Keyboard'],
-            'category': ['Electronics', 'Electronics', 'Audio', 'Electronics', 'Accessories'],
-            'price': [1200, 800, 150, 300, 80],
-            'stock': [10, 25, 50, 15, 30]
-        })
-        
-        # Sample transactions data
-        self.transactions = pd.DataFrame({
-            'transaction_id': [101, 102, 103, 104, 105, 106, 107],
-            'product_id': [1, 2, 3, 1, 5, 2, 4],
-            'quantity': [1, 2, 3, 1, 2, 1, 2],
-            'date': ['2025-04-29', '2025-04-29', '2025-04-28', '2025-04-28', '2025-04-27', '2025-04-29', '2025-04-29'],
-            'revenue': [1200, 1600, 450, 1200, 160, 800, 600]
-        })
-    
-    def query_database(self, query_type, **kwargs):
-        """Execute queries on the database based on query type"""
-        if query_type == "product_info":
-            if 'product_name' in kwargs:
-                return self.products[self.products['name'].str.lower() == kwargs['product_name'].lower()]
-            elif 'product_id' in kwargs:
-                return self.products[self.products['product_id'] == kwargs['product_id']]
-            else:
-                return self.products
-                
-        elif query_type == "max_revenue_product":
-            date_filter = kwargs.get('date', '2025-04-29')  # Default to today
-            
-            # Group by product_id and calculate total revenue for the specified date
-            daily_revenue = self.transactions[self.transactions['date'] == date_filter].groupby(
-                'product_id')['revenue'].sum().reset_index()
-            
-            if daily_revenue.empty:
-                return "No sales data found for that date."
-                
-            # Find the product with max revenue
-            max_revenue_product_id = daily_revenue.loc[daily_revenue['revenue'].idxmax()]['product_id']
-            max_revenue = daily_revenue.loc[daily_revenue['revenue'].idxmax()]['revenue']
-            
-            # Get product details
-            product_details = self.products[self.products['product_id'] == max_revenue_product_id].iloc[0]
-            
-            return {
-                'product_name': product_details['name'],
-                'revenue': max_revenue,
-                'date': date_filter
-            }
-            
-        elif query_type == "inventory_check":
-            product_name = kwargs.get('product_name')
-            if product_name:
-                product = self.products[self.products['name'].str.lower() == product_name.lower()]
-                if not product.empty:
-                    return {'product': product_name, 'stock': product.iloc[0]['stock']}
-                return f"Product '{product_name}' not found."
-            return self.products[['name', 'stock']]
-            
-        return "Query type not supported"
+def web_search_agent(query: str) -> str:
+    # Example using SerpAPI
+    resp = requests.get(
+        "https://serpapi.com/search",
+        params={"q": query, "api_key": os.getenv("SERPAPI_KEY")}  
+    ).json()
+    snippet = resp.get("organic_results", [{}])[0].get("snippet", "")
+    return llm_agent(f"Summarize: {snippet}")
 
-class QueryRouter:
-    def __init__(self):
-        """Initialize the query router"""
-        pass
-        
-    def _classify_query(self, query):
-        """Classify the query to determine which agent should handle it"""
-        # Use OpenAI to classify the query
-        messages = [
-            {"role": "system", "content": """
-            You are a query classifier for a shop assistant system. 
-            Classify customer queries into one of these categories:
-            - max_revenue_product: Questions about which product generated the most revenue (today or on a specific date)
-            - inventory_check: Questions about product availability or stock levels
-            - product_info: Questions about product details, pricing, etc.
-            - general_knowledge: Questions that require general knowledge not related to specific shop data
-            
-            Return ONLY the category as a single word without any explanation.
-            """},
-            {"role": "user", "content": query}
-        ]
-        
-        response = openai_chat_completion(messages, temperature=0)
-        
-        # Extract the query type from the response
-        query_type = response["choices"][0]["message"]["content"].strip().lower()
-        return query_type
-        
-    def _extract_parameters(self, query, query_type):
-        """Extract relevant parameters from the query based on query type"""
-        # Use OpenAI to extract parameters
-        prompt_content = f"""
-        Extract parameters from this customer query: "{query}"
-        Query type: {query_type}
-        
-        For max_revenue_product:
-        - date (in YYYY-MM-DD format, extract "today" as today's date which is 2025-04-29)
-        
-        For inventory_check or product_info:
-        - product_name (the name of the product being asked about)
-        
-        Return ONLY a valid JSON object with the extracted parameters, nothing else.
-        Example: {{"product_name": "laptop"}} or {{"date": "2025-04-29"}}
-        """
-        
-        messages = [
-            {"role": "system", "content": "You extract parameters from customer queries for a shop assistant."},
-            {"role": "user", "content": prompt_content}
-        ]
-        
-        response = openai_chat_completion(messages, temperature=0, response_format={"type": "json_object"})
-        
-        # Parse the JSON response
-        try:
-            parameters = json.loads(response["choices"][0]["message"]["content"])
-            return parameters
-        except json.JSONDecodeError:
-            return {}
-        
-        # Parse the JSON response
-        import json
-        try:
-            parameters = json.loads(response.choices[0].message.content)
-            return parameters
-        except json.JSONDecodeError:
-            return {}
-            
-    def _handle_general_knowledge(self, query):
-        """Handle general knowledge queries using OpenAI"""
-        messages = [
-            {"role": "system", "content": """
-            You are a helpful assistant for a shop. Answer the customer's question 
-            using your general knowledge. Keep answers brief and focused.
-            """},
-            {"role": "user", "content": query}
-        ]
-        
-        response = openai_chat_completion(messages, temperature=0.7, max_tokens=150)
-        
-        return response["choices"][0]["message"]["content"]
-        
-    def _format_response(self, query_type, data):
-        """Format the response based on query type and data"""
-        if query_type == "max_revenue_product":
-            if isinstance(data, str):
-                return data
-            return f"The product with the highest revenue on {data['date']} is {data['product_name']} with ${data['revenue']} in sales."
-            
-        elif query_type == "inventory_check":
-            if isinstance(data, str):
-                return data
-            if isinstance(data, dict) and 'product' in data:
-                return f"We currently have {data['stock']} units of {data['product']} in stock."
-            return "Here's our current inventory: " + ", ".join([f"{row['name']}: {row['stock']} units" for _, row in data.iterrows()])
-            
-        elif query_type == "product_info":
-            if data.empty:
-                return "Product not found."
-            if len(data) == 1:
-                product = data.iloc[0]
-                return f"{product['name']} ({product['category']}): ${product['price']}. We have {product['stock']} units in stock."
-            return "Here are our products: " + ", ".join([f"{row['name']}: ${row['price']}" for _, row in data.iterrows()])
-            
-        return str(data)
-        
-    def process(self, query, db):
-        """Process the query and return a response"""
-        # Classify the query
-        query_type = self._classify_query(query)
-        
-        # If it's a general knowledge query, handle it differently
-        if query_type == "general_knowledge":
-            return self._handle_general_knowledge(query)
-            
-        # Extract parameters from the query
-        parameters = self._extract_parameters(query, query_type)
-        
-        # Query the database
-        result = db.query_database(query_type, **parameters)
-        
-        # Format the response
-        response = self._format_response(query_type, result)
-        
-        return response
 
-# Initialize database and router
-db = SimpleDatabase()
-router = QueryRouter()
+def llm_agent(prompt: str) -> str:
+    resp = openai.ChatCompletion.create(
+        model="gpt-4o-mini",
+        messages=[
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "user", "content": prompt}
+        ],
+        temperature=0.2
+    )
+    return resp.choices[0].message.content.strip()
 
-def process_query(query):
-    """Process the user query and return a response"""
-    if not query.strip():
-        return "Please ask a question about our shop products or services."
-    
-    response = router.process(query, db)
-    return response
 
-# Create Gradio interface
-demo = gr.Interface(
-    fn=process_query,
-    inputs=gr.Textbox(
-        placeholder="Ask about product pricing, inventory, sales, or any other question...",
-        label="Customer Query"
-    ),
-    outputs=gr.Textbox(label="Shop Assistant Response"),
-    title="Shop Voice Box Assistant",
-    description="Ask questions about products, inventory, sales, or general questions.",
-    examples=[
-        ["What's the maximum revenue product today?"],
-        ["How many laptops do we have in stock?"],
-        ["Tell me about the smartphone."],
-        ["What's the weather like today?"]
-    ]
-)
+def handle_query(query: str) -> str:
+    q = query.lower()
+    if any(k in q for k in ["max", "revenue", "today", "product"]):
+        return db_agent(query)
+    elif any(k in q for k in ["who", "what", "when", "where"]):
+        return web_search_agent(query)
+    else:
+        return llm_agent(query)
+
+# --- Gradio UI ---
+with gr.Blocks() as demo:
+    gr.Markdown("## Shop Voice-Box Assistant")
+    user_input = gr.Textbox(placeholder="Type your question here...", lines=2)
+    submit_btn = gr.Button("Submit")
+    response_box = gr.Textbox(label="Answer")
+    submit_btn.click(fn=handle_query, inputs=user_input, outputs=response_box)
 
-# Launch the app
 if __name__ == "__main__":
     demo.launch()