Spaces:
Build error
Build error
File size: 8,870 Bytes
3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae 6e94c23 3a7b7ae | 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 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | 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]
|