vikarti.anatra/Automated-AI-Web-Researcher-Ollama
1
0
Fork 0
mirror of https://github.com/TheBlewish/Automated-AI-Web-Researcher-Ollama.git synced 2025-09-01 10:10:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update Self_Improving_Search.py for windows

Browse source 
Reconfigured a few things
This commit is contained in:
Hafeez 2024-11-20 22:34:32 +05:30 • committed by GitHub
parent c3c29a5da5
commit abcd32130c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 58 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
Self_Improving_Search.py
View file

@ -11,6 +11,7 @@ from llm_config import get_llm_config
from llm_response_parser import UltimateLLMResponseParser from llm_response_parser import UltimateLLMResponseParser
from llm_wrapper import LLMWrapper from llm_wrapper import LLMWrapper
from urllib.parse import urlparse from urllib.parse import urlparse
import duckduckgo_search
# Set up logging # Set up logging
log_directory = 'logs' log_directory = 'logs'
@ -35,6 +36,7 @@ for name in ['root', 'duckduckgo_search', 'requests', 'urllib3']:
logging.getLogger(name).propagate = False logging.getLogger(name).propagate = False
class OutputRedirector: class OutputRedirector:
"""Windows-compatible output redirection"""
def __init__(self, stream=None): def __init__(self, stream=None):
self.stream = stream or StringIO() self.stream = stream or StringIO()
self.original_stdout = sys.stdout self.original_stdout = sys.stdout
@ -55,6 +57,8 @@ class EnhancedSelfImprovingSearch:
self.parser = parser self.parser = parser
self.max_attempts = max_attempts self.max_attempts = max_attempts
self.llm_config = get_llm_config() self.llm_config = get_llm_config()
self.last_query = "" # Add this
self.last_time_range = "" # Add this
@staticmethod @staticmethod
def initialize_llm(): def initialize_llm():
@ -238,23 +242,54 @@ Do not provide any additional information or explanation.
return " ".join(words[:5]) return " ".join(words[:5])
def perform_search(self, query: str, time_range: str) -> List[Dict]: def perform_search(self, query: str, time_range: str) -> List[Dict]:
"""Performs the actual web search with enhanced error handling"""
if not query: if not query:
return [] return []
from duckduckgo_search import DDGS from duckduckgo_search import DDGS
# Store the last query and time range for display purposes
self.last_query = query
self.last_time_range = time_range
max_retries = 3
for attempt in range(max_retries):
try:
with DDGS() as ddgs: with DDGS() as ddgs:
try: try:
with OutputRedirector() as output:
if time_range and time_range != 'none': if time_range and time_range != 'none':
results = list(ddgs.text(query, timelimit=time_range, max_results=10)) results = list(ddgs.text(
query,
timelimit=time_range,
max_results=10,
backend='api' # Use API backend for more reliable results
))
else: else:
results = list(ddgs.text(query, max_results=10)) results = list(ddgs.text(
ddg_output = output.getvalue() query,
logger.info(f"DDG Output in perform_search:\n{ddg_output}") max_results=10,
backend='api' # Use API backend for more reliable results
))
if results:
return [{'number': i+1, **result} for i, result in enumerate(results)] return [{'number': i+1, **result} for i, result in enumerate(results)]
else:
print(f"{Fore.YELLOW}No results found on attempt {attempt + 1}. Retrying...{Style.RESET_ALL}")
time.sleep(1) # Add delay between retries
except Exception as e: except Exception as e:
print(f"{Fore.RED}Search error: {str(e)}{Style.RESET_ALL}") logger.error(f"DuckDuckGo search error on attempt {attempt + 1}: {str(e)}")
print(f"{Fore.RED}Search error on attempt {attempt + 1}: {str(e)}{Style.RESET_ALL}")
time.sleep(1) # Add delay between retries
continue
except Exception as e:
logger.error(f"DDGS initialization error on attempt {attempt + 1}: {str(e)}")
print(f"{Fore.RED}Search initialization error on attempt {attempt + 1}: {str(e)}{Style.RESET_ALL}")
time.sleep(1) # Add delay between retries
continue
print(f"{Fore.RED}Search failed after {max_retries} attempts{Style.RESET_ALL}")
return [] return []
def display_search_results(self, results: List[Dict]) -> None: def display_search_results(self, results: List[Dict]) -> None: