File size: 2,773 Bytes
b12e5fe 0729c66 b12e5fe 0729c66 b12e5fe 0729c66 dd5aa4f 9fd7af7 b12e5fe 0729c66 dd5aa4f 0729c66 b12e5fe 0729c66 dd5aa4f 0729c66 dd5aa4f 0729c66 dd5aa4f 0729c66 dd5aa4f b12e5fe 0729c66 243aa91 0729c66 b12e5fe dd5aa4f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
import os
import sqlite3
import requests
import openai
import gradio as gr
# Load API keys from environment
openai.api_key = os.getenv("OPENAI_API_KEY")
# --- Agents ---
def db_agent(query: str) -> str:
try:
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()
if row:
return f"Top product today: {row[0]} with ₹{row[1]:,.2f}"
return "No transactions found for today."
return None
except sqlite3.OperationalError as e:
# Handle missing table or other DB errors
return f"Database error: {e}. Please initialize 'transactions' table in shop.db."
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")}
)
data = resp.json()
snippet = data.get("organic_results", [{}])[0].get("snippet", "")
return llm_agent(f"Summarize: {snippet}")
def llm_agent(prompt: str) -> str:
# Updated for openai>=1.0.0 interface
response = openai.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": prompt},
],
temperature=0.2,
)
return response.choices[0].message.content.strip()
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")
# Sample questions
gr.Examples(
examples=[
["What is the max revenue product today?"],
["Who invented the light bulb?"],
["Tell me a joke about cats."],
],
inputs=user_input,
outputs=response_box,
)
submit_btn.click(fn=handle_query, inputs=user_input, outputs=response_box)
if __name__ == "__main__":
demo.launch(share=False, server_name="0.0.0.0", server_port=7860) |