Spaces:
Build error
Build error
| from playwright.async_api import async_playwright | |
| from langchain_community.agent_toolkits import PlayWrightBrowserToolkit | |
| from dotenv import load_dotenv | |
| import os | |
| import requests | |
| from langchain.agents import Tool | |
| from langchain.tools import StructuredTool | |
| from langchain_community.agent_toolkits import FileManagementToolkit | |
| from langchain_community.tools.wikipedia.tool import WikipediaQueryRun | |
| from langchain_experimental.tools import PythonREPLTool | |
| from langchain_community.utilities import GoogleSerperAPIWrapper | |
| from langchain_community.utilities.wikipedia import WikipediaAPIWrapper | |
| from langchain.agents.agent_toolkits import SQLDatabaseToolkit | |
| from langchain.sql_database import SQLDatabase | |
| import uuid | |
| import sqlite3 | |
| from langchain_openai import ChatOpenAI | |
| from pydantic import BaseModel, Field | |
| load_dotenv(override=True) | |
| pushover_token = os.getenv("PUSHOVER_TOKEN") | |
| pushover_user = os.getenv("PUSHOVER_USER") | |
| pushover_url = "https://api.pushover.net/1/messages.json" | |
| serper = GoogleSerperAPIWrapper() | |
| class GetUserHistoryInput(BaseModel): | |
| username: str = Field(..., description="The username to fetch history for") | |
| query_text: str = Field(None, description="Natural language query about history to retrieve") | |
| async def playwright_tools(): | |
| playwright = await async_playwright().start() | |
| browser = await playwright.chromium.launch(headless=False) | |
| toolkit = PlayWrightBrowserToolkit.from_browser(async_browser=browser) | |
| return toolkit.get_tools(), browser, playwright | |
| def push(text: str): | |
| """Send a push notification to the user""" | |
| requests.post(pushover_url, data = {"token": pushover_token, "user": pushover_user, "message": text}) | |
| return "success" | |
| def get_file_tools(): | |
| toolkit = FileManagementToolkit(root_dir="sandbox") | |
| return toolkit.get_tools() | |
| def format_search_history(formatted_results, success_criteria=None): | |
| if not formatted_results: | |
| return "No search history found." | |
| prompt = f""" | |
| You are a formatting expert. Format the following search history data: | |
| {formatted_results} | |
| Format this data according to these requirements: | |
| {success_criteria if success_criteria else "Make it readable and well-organized."} | |
| Use plain text only (no HTML or Markdown). Return ONLY the formatted output without any explanation. | |
| You can use Unicode symbols, ASCII art, or emojis if appropriate. | |
| """ | |
| llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) | |
| response = llm.invoke(prompt) | |
| return response.content | |
| def get_user_history(query_text: str = None, username: str = None): | |
| """ | |
| Retrieves search history from the database based on natural language query. | |
| Converts natural language to SQL, executes the query, and formats results. | |
| Args: | |
| username: The username to fetch history for (required) | |
| query_text: Natural language description of what history to retrieve (e.g. "show my searches about AI") | |
| """ | |
| try: | |
| conn = sqlite3.connect("query_log.db") | |
| cursor = conn.cursor() | |
| cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='search_history'") | |
| if not cursor.fetchone(): | |
| return "Error: The search history table does not exist yet. Try making some searches first." | |
| cursor.execute("PRAGMA table_info(search_history)") | |
| schema_info = cursor.fetchall() | |
| schema_description = "\n".join([f"- {col[1]} ({col[2]})" for col in schema_info]) | |
| default_query = "SELECT * FROM search_history WHERE username = ? ORDER BY timestamp DESC" | |
| if query_text: | |
| llm = ChatOpenAI(model="gpt-4o-mini",temperature=0) | |
| prompt = f""" | |
| Table name: search_history | |
| Table schema: | |
| {schema_description} | |
| User question: {query_text} | |
| Username filter: {username} (Always filter by this username for security) | |
| Generate a SQL query that answers the user's question while ALWAYS including 'WHERE username = ?' | |
| in the query for security. Return ONLY the SQL query without explanation or backticks. | |
| """ | |
| max_tries = 3 | |
| sql_query = None | |
| for attempt in range(max_tries): | |
| try: | |
| #print(f"Generating SQL (attempt {attempt+1}/{max_tries})...") | |
| sql_response = llm.invoke(prompt) | |
| generated_query = sql_response.content.strip() | |
| if "WHERE username = ?" not in generated_query and "where username = ?" not in generated_query: | |
| generated_query = f"{generated_query} WHERE username = ?" | |
| #print(f"Generated query: {generated_query}") | |
| if generated_query.count("WHERE username = ?") > 1 or generated_query.count("where username = ?") > 1: | |
| generated_query = generated_query.replace("WHERE username = ?", "").replace("where username = ?", "") | |
| if "WHERE" not in generated_query and "where" not in generated_query: | |
| generated_query = f"{generated_query} WHERE username = ?" | |
| else: | |
| generated_query = generated_query.replace("WHERE", "WHERE username = ? AND").replace("where", "where username = ? AND") | |
| conn.execute("EXPLAIN " + generated_query, (username,)) | |
| sql_query = generated_query | |
| break | |
| except Exception as e: | |
| error_message = str(e) | |
| print(f"SQL error on attempt {attempt+1}: {error_message}") | |
| prompt += f"\nPrevious attempt failed with error: {error_message}\nPlease fix and try again." | |
| if attempt == max_tries - 1: | |
| print("Falling back to default query") | |
| sql_query = default_query | |
| else: | |
| sql_query = default_query | |
| print(f"Executing query: {sql_query}") | |
| cursor.execute(sql_query, (username,)) | |
| results = cursor.fetchall() | |
| formatted_results = [] | |
| for row in results: | |
| record = {} | |
| for i, col in enumerate(cursor.description): | |
| record[col[0]] = row[i] | |
| formatted_results.append(record) | |
| conn.close() | |
| if not formatted_results: | |
| return "You don't have any search history records matching your query." | |
| output = "Here's your search history:\n\n" | |
| output += format_search_history(formatted_results) | |
| return output | |
| except Exception as e: | |
| print(f"[ERROR] Database error: {str(e)}") | |
| return f"Sorry, I encountered an error while retrieving your search history: {str(e)}" | |
| async def other_tools(): | |
| push_tool = Tool(name="send_push_notification", func=push, description="Use this tool when you want to send a push notification") | |
| file_tools = get_file_tools() | |
| tool_search =Tool( | |
| name="search", | |
| func=serper.run, | |
| description="Use this tool when you want to get the results of an online web search" | |
| ) | |
| get_history_tool = StructuredTool( | |
| name="get_user_history", | |
| func=get_user_history, | |
| description="""Use this tool when the user wants to view or query their search history. | |
| Examples of when to use this tool: | |
| - User asks about their search history | |
| - User wants to see previous searches | |
| - User asks for specific searches (like "show my searches about AI") | |
| Required parameters: | |
| - query_text: The user's natural language question about their history | |
| - username: The username of the current user | |
| """, | |
| args_schema=GetUserHistoryInput | |
| ) | |
| wikipedia = WikipediaAPIWrapper() | |
| wiki_tool = WikipediaQueryRun(api_wrapper=wikipedia) | |
| python_repl = PythonREPLTool() | |
| return file_tools + [push_tool, tool_search, python_repl, wiki_tool, get_history_tool] | |