stephane.david/agenticSeek
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'Fosowl:main' into main

Browse Source
This commit is contained in:
ganesh nikhil
2025-03-18 00:41:19 +05:30
committed by GitHub
parent 2f912b0b95 aad71179a5
commit 477a145712
16 changed files with 5786 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sources/agents/browser_agent.py
+27 -20
View File
@@ -3,8 +3,9 @@ import time
from sources.utility import pretty_print, animate_thinking
from sources.agents.agent import Agent
from sources.tools.webSearch import webSearch
from sources.tools.searxSearch import searxSearch
from sources.browser import Browser
class BrowserAgent(Agent):
def __init__(self, model, name, prompt_path, provider):
"""
@@ -12,7 +13,7 @@ class BrowserAgent(Agent):
"""
super().__init__(model, name, prompt_path, provider)
self.tools = {
"web_search": webSearch(),
"web_search": searxSearch(),
}
self.role = "deep research and web search"
self.browser = Browser()
@@ -65,23 +66,25 @@ class BrowserAgent(Agent):
You can navigate to these links:
{remaining_links}
You must choose a link (write it down) to navigate to, or go back.
For exemple you can say: i want to go to www.wikipedia.org/cats
Follow up with a summary of the page content (of the current page, not of the link), for example:
Summary: According to https://karpathy.github.io/ LeCun net is the earliest real-world application of a neural net"
The summary should include any useful finding that are useful in answering user query.
If a website does not have usefull information say Error, for exemple:
Error: This forum does not discus anything that can answer the user query
Be short, concise, direct.
If no link seem appropriate, please say "GO_BACK".
Remember, you seek the information the user want.
The user query was : {user_prompt}
You must choose a link (write it down) to navigate to, or go back.
For exemple you can say: i want to go to www.wikipedia.org/cats
Always end with a sentence that summarize when useful information is found for exemple:
Summary: According to https://karpathy.github.io/ LeCun net is the earliest real-world application of a neural net"
Do not say "according to this page", always write down the whole link.
If a website does not have usefull information say Error, for exemple:
Error: This forum does not discus anything that can answer the user query
Do not explain your choice, be short, concise.
"""
def llm_decide(self, prompt):
animate_thinking("Thinking...", color="status")
self.memory.push('user', prompt)
answer, reasoning = self.llm_request(prompt)
answer, reasoning = self.llm_request()
pretty_print("-"*100)
pretty_print(answer, color="output")
pretty_print("-"*100)
@@ -119,16 +122,20 @@ class BrowserAgent(Agent):
def save_notes(self, text):
lines = text.split('\n')
for line in lines:
if "summary:" in line.lower():
if "summary" in line.lower():
self.notes.append(line)
def conclude_prompt(self, user_query):
search_note = '\n -'.join(self.notes)
annotated_notes = [f"{i+1}: {note.lower().replace('summary:', '')}" for i, note in enumerate(self.notes)]
search_note = '\n'.join(annotated_notes)
print("AI research notes:\n", search_note)
return f"""
Following a web search about:
Following a human request:
{user_query}
Write a conclusion based on these notes:
A web AI made the following finding across different pages:
{search_note}
Summarize the finding, and provide a conclusion that answer the request.
"""
def process(self, user_prompt, speech_module) -> str:
@@ -136,8 +143,7 @@ class BrowserAgent(Agent):
animate_thinking(f"Searching...", color="status")
search_result_raw = self.tools["web_search"].execute([user_prompt], False)
search_result = self.jsonify_search_results(search_result_raw)
search_result = search_result[:10] # until futher improvement
search_result = self.jsonify_search_results(search_result_raw)[:5] # until futher improvement
prompt = self.make_newsearch_prompt(user_prompt, search_result)
unvisited = [None]
while not complete:
@@ -147,14 +153,14 @@ class BrowserAgent(Agent):
complete = True
break
links = self.extract_links(answer)
if len(unvisited) == 0:
break
if len(links) == 0 or "GO_BACK" in answer:
unvisited = self.select_unvisited(search_result)
prompt = self.make_newsearch_prompt(user_prompt, unvisited)
pretty_print(f"Going back to results. Still {len(unvisited)}", color="warning")
links = []
continue
if len(unvisited) == 0:
break
animate_thinking(f"Navigating to {links[0]}", color="status")
speech_module.speak(f"Navigating to {links[0]}")
self.browser.go_to(links[0])
@@ -163,11 +169,12 @@ class BrowserAgent(Agent):
self.navigable_links = self.browser.get_navigable()
prompt = self.make_navigation_prompt(user_prompt, page_text)
speech_module.speak(answer)
self.browser.close()
prompt = self.conclude_prompt(user_prompt)
self.memory.push('user', prompt)
answer, reasoning = self.llm_request(prompt)
pretty_print(answer, color="output")
speech_module.speak(answer)
return answer, reasoning
if __name__ == "__main__":
sources/agents/casual_agent.py
+2 -2
View File
@@ -1,7 +1,7 @@
from sources.utility import pretty_print, animate_thinking
from sources.agents.agent import Agent
from sources.tools.webSearch import webSearch
from sources.tools.searxSearch import searxSearch
from sources.tools.flightSearch import FlightSearch
from sources.tools.fileFinder import FileFinder
from sources.tools.BashInterpreter import BashInterpreter
@@ -13,7 +13,7 @@ class CasualAgent(Agent):
"""
super().__init__(model, name, prompt_path, provider)
self.tools = {
"web_search": webSearch(),
"web_search": searxSearch(),
"flight_search": FlightSearch(),
"file_finder": FileFinder(),
"bash": BashInterpreter()
sources/interaction.py
+1 -1
View File
@@ -27,7 +27,7 @@ class Interaction:
self.transcriber = AudioTranscriber(self.ai_name, verbose=False)
self.recorder = AudioRecorder()
if tts_enabled:
self.speech.speak("Hello Sir, we are online and ready. What can I do for you ?")
self.speech.speak("Hello, we are online and ready. What can I do for you ?")
if recover_last_session:
self.recover_last_session()
sources/tools/searxSearch.py
+116
View File
@@ -0,0 +1,116 @@
import requests
from bs4 import BeautifulSoup
import os
if __name__ == "__main__":
from tools import Tools
else:
from sources.tools.tools import Tools
class searxSearch(Tools):
def __init__(self, base_url: str = None):
"""
A tool for searching a SearxNG instance and extracting URLs and titles.
"""
super().__init__()
self.tag = "web_search"
self.base_url = base_url or os.getenv("SEARXNG_BASE_URL") # Requires a SearxNG base URL
self.user_agent = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36"
self.paywall_keywords = [
"Member-only", "access denied", "restricted content", "404", "this page is not working"
]
if not self.base_url:
raise ValueError("SearxNG base URL must be provided either as an argument or via the SEARXNG_BASE_URL environment variable.")
def link_valid(self, link):
"""check if a link is valid."""
# TODO find a better way
if not link.startswith("http"):
return "Status: Invalid URL"
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
try:
response = requests.get(link, headers=headers, timeout=5)
status = response.status_code
if status == 200:
content = response.text.lower()
if any(keyword in content for keyword in self.paywall_keywords):
return "Status: Possible Paywall"
return "Status: OK"
elif status == 404:
return "Status: 404 Not Found"
elif status == 403:
return "Status: 403 Forbidden"
else:
return f"Status: {status} {response.reason}"
except requests.exceptions.RequestException as e:
return f"Error: {str(e)}"
def check_all_links(self, links):
"""Check all links, one by one."""
# TODO Make it asyncromous or smth
statuses = []
print("Web scrawl to verify links accessibilty...")
for i, link in enumerate(links):
status = self.link_valid(link)
statuses.append(status)
return statuses
def execute(self, blocks: list, safety: bool = False) -> str:
"""Executes a search query against a SearxNG instance using POST and extracts URLs and titles."""
if not blocks:
return "Error: No search query provided."
query = blocks[0].strip()
if not query:
return "Error: Empty search query provided."
search_url = f"{self.base_url}/search"
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Language': 'en-US,en;q=0.9',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded',
'Pragma': 'no-cache',
'Upgrade-Insecure-Requests': '1',
'User-Agent': self.user_agent
}
data = f"q={query}&categories=general&language=auto&time_range=&safesearch=0&theme=simple"
try:
response = requests.post(search_url, headers=headers, data=data, verify=False)
response.raise_for_status()
html_content = response.text
soup = BeautifulSoup(html_content, 'html.parser')
results = []
for article in soup.find_all('article', class_='result'):
url_header = article.find('a', class_='url_header')
if url_header:
url = url_header['href']
title = article.find('h3').text.strip() if article.find('h3') else "No Title"
description = article.find('p', class_='content').text.strip() if article.find('p', class_='content') else "No Description"
results.append(f"Title:{title}\nSnippet:{description}\nLink:{url}")
if len(results) == 0:
raise Exception("Searx search failed. did you run start_services.sh? Did docker die?")
return "\n\n".join(results) # Return results as a single string, separated by newlines
except requests.exceptions.RequestException as e:
return f"Error during search: {str(e)}"
def execution_failure_check(self, output: str) -> bool:
"""
Checks if the execution failed based on the output.
"""
return "Error" in output
def interpreter_feedback(self, output: str) -> str:
"""
Feedback of web search to agent.
"""
if self.execution_failure_check(output):
return f"Web search failed: {output}"
return f"Web search result:\n{output}"
if __name__ == "__main__":
search_tool = searxSearch(base_url="http://127.0.0.1:8080")
result = search_tool.execute(["are dog better than cat?"])
print(result)
sources/tools/webSearch.py
+5
View File
@@ -14,6 +14,11 @@ else:
from sources.tools.tools import Tools
from sources.utility import animate_thinking, pretty_print
"""
WARNING
webSearch is fully deprecated and is being replaced by searxSearch for web search.
"""
class webSearch(Tools):
def __init__(self, api_key: str = None):
"""