From a429bcc7944931f5fdb40bb94e886175dafd91e5 Mon Sep 17 00:00:00 2001 From: Russell Ballestrini Date: Fri, 25 Apr 2025 23:01:23 -0400 Subject: [PATCH] modified: duck_duck_go_hermes_unturf.py --- duck_duck_go_hermes_unturf.py | 435 +++++++++++++++++++++++++++------- 1 file changed, 349 insertions(+), 86 deletions(-) diff --git a/duck_duck_go_hermes_unturf.py b/duck_duck_go_hermes_unturf.py index c400737..5d43858 100644 --- a/duck_duck_go_hermes_unturf.py +++ b/duck_duck_go_hermes_unturf.py @@ -3,9 +3,11 @@ import requests import random import time import re -from urllib.parse import quote_plus +from urllib.parse import quote_plus, urlparse, parse_qs, unquote from datetime import datetime, timezone from concurrent.futures import ThreadPoolExecutor, as_completed +import urllib.robotparser +import threading from sqlalchemy import ( create_engine, @@ -22,24 +24,37 @@ from bs4 import BeautifulSoup # Default Hermes endpoints DEFAULT_HERMES_ENDPOINTS = [ - "https://hermes.ai.unturf.com/v1", - "https://hermes2.ai.unturf.com/v1", + "https://hermes.ai.unturf.com/v1", # 80k context window + "https://hermes2.ai.unturf.com/v1", # Smaller context window ] # Max characters to send to extraction to avoid context overflow MAX_HTML_INPUT_CHARS = 50000 +# Default crawl delay if robots.txt is missing or doesn't specify +DEFAULT_CRAWL_DELAY = 2.0 # seconds +# Token limits for Hermes endpoints +HERMES_TOKEN_LIMIT = 80000 # hermes.ai.unturf.com +HERMES2_TOKEN_LIMIT = 48000 # hermes2.ai.unturf.com +HERMES_INPUT_TOKENS = 72000 +HERMES_COMPLETION_TOKENS = 8000 +HERMES2_INPUT_TOKENS = 40000 +HERMES2_COMPLETION_TOKENS = 8000 Base = declarative_base() + class Article(Base): - __tablename__ = 'articles' + __tablename__ = "articles" id = Column(Integer, primary_key=True) url = Column(String, unique=True, nullable=False) title = Column(String, nullable=False) raw_html = Column(Text, nullable=False) extracted_content = Column(Text, nullable=False) - summary = Column(Text, nullable=False) - fetched_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) - __table_args__ = (UniqueConstraint('url', name='_url_uc'),) + summary = Column(Text, nullable=False) # Stores extracted_content unless summarized + fetched_at = Column( + DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + __table_args__ = (UniqueConstraint("url", name="_url_uc"),) + class SQLAlchemyDuckDuckGoCrawler: def __init__( @@ -48,160 +63,408 @@ class SQLAlchemyDuckDuckGoCrawler: model, db_path="data/articles.db", hermes_endpoints=None, + user_agent_append="", ): # Setup Hermes clients self.hermes_endpoints = hermes_endpoints or DEFAULT_HERMES_ENDPOINTS - self.clients = [OpenAI(base_url=ep, api_key=api_key) for ep in self.hermes_endpoints] + self.clients = [ + OpenAI(base_url=ep, api_key=api_key) for ep in self.hermes_endpoints + ] self.api_key = api_key self.model = model self.db_path = db_path - # HTTP session for DuckDuckGo and page fetches + # Configure user agent + default_requests_ua = f"python-requests/{requests.__version__}" + self.user_agent = ( + f"{default_requests_ua} unturf-deep-research {user_agent_append}".strip() + ) self.session = requests.Session() - self.session.headers.update({ - "User-Agent": ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/91.0.4472.124 Safari/537.36" - ) - }) + self.session.headers.update({"User-Agent": self.user_agent}) # Initialize DB - self.engine = create_engine(f"sqlite:///{self.db_path}", echo=False, future=True) + self.engine = create_engine( + f"sqlite:///{self.db_path}", + echo=False, + future=True, + connect_args={"check_same_thread": False}, + ) Base.metadata.create_all(self.engine) - SessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) - self.db = SessionLocal() + self.SessionLocal = sessionmaker( + bind=self.engine, autoflush=False, autocommit=False + ) + + # Robots.txt parser and domain tracking + self.robot_parsers = {} + self.domain_last_fetched = {} + self.domain_crawl_delays = {} + self.robots_txt_content = {} + + def _get_domain(self, url): + """Extract the domain from a URL.""" + parsed = urlparse(url) + return parsed.netloc + + def _fetch_robots_txt(self, domain): + """Fetch and parse robots.txt for a domain.""" + if domain in self.robot_parsers: + return self.robot_parsers[domain] + + robots_url = f"https://{domain}/robots.txt" + parser = urllib.robotparser.RobotFileParser() + parser.set_url(robots_url) + try: + resp = self.session.get(robots_url, timeout=5) + resp.raise_for_status() + self.robots_txt_content[domain] = resp.text + parser.parse(resp.text.splitlines()) + self.robot_parsers[domain] = parser + delay = parser.crawl_delay(self.user_agent) + self.domain_crawl_delays[domain] = ( + delay if delay is not None else DEFAULT_CRAWL_DELAY + ) + except Exception as e: + print(f"Could not fetch robots.txt for {domain}: {e}") + self.robot_parsers[domain] = None + self.robots_txt_content[domain] = None + self.domain_crawl_delays[domain] = DEFAULT_CRAWL_DELAY + return self.robot_parsers[domain] + + def _can_fetch(self, url): + """Check if crawling the URL is allowed per robots.txt.""" + domain = self._get_domain(url) + parser = self._fetch_robots_txt(domain) + if parser is None: + return True + + can_fetch = parser.can_fetch(self.user_agent, url) + if not can_fetch: + print(f"Blocked by robots.txt: {url}") + robots_content = self.robots_txt_content.get(domain, "") + if robots_content: + print(f"Relevant robots.txt rules for {domain}:") + current_user_agent = None + relevant_rules = [] + for line in robots_content.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + if line.lower().startswith("user-agent:"): + current_user_agent = line[11:].strip() + elif line.lower().startswith("disallow:") and current_user_agent: + rule = line[9:].strip() + if current_user_agent.lower() in (self.user_agent.lower(), "*"): + parsed_url = urlparse(url) + path = parsed_url.path + if rule and ( + path.startswith(rule) + or (rule.endswith("*") and path.startswith(rule[:-1])) + ): + relevant_rules.append( + f" User-agent: {current_user_agent}\n Disallow: {rule}" + ) + if relevant_rules: + print("\n".join(relevant_rules)) + else: + print( + " No specific Disallow rules found; may be blocked by a broad rule." + ) + else: + print(" No robots.txt content available to display rules.") + return can_fetch + + def _enforce_crawl_delay(self, domain): + """Enforce crawl delay for the domain.""" + delay = self.domain_crawl_delays.get(domain, DEFAULT_CRAWL_DELAY) + last_fetched = self.domain_last_fetched.get(domain, 0) + elapsed = time.time() - last_fetched + if elapsed < delay: + sleep_time = delay - elapsed + print( + f"Delaying crawl for {domain} by {sleep_time:.2f} seconds due to crawl delay" + ) + time.sleep(sleep_time) + self.domain_last_fetched[domain] = time.time() def search_duckduckgo(self, query, max_results=10): + """Search DuckDuckGo and extract target URLs from redirect links.""" encoded = quote_plus(query) url = f"https://html.duckduckgo.com/html/?q={encoded}" resp = self.session.get(url, timeout=10) resp.raise_for_status() - soup = BeautifulSoup(resp.text, 'html.parser') + soup = BeautifulSoup(resp.text, "html.parser") results = [] - for link in soup.select('.result__title a'): + for link in soup.select(".result__title a"): if len(results) >= max_results: break - results.append((link.get_text(strip=True), link.get('href'))) + href = link.get("href") + title = link.get_text(strip=True) + if href and href.startswith("//duckduckgo.com/l/?uddg="): + parsed = urlparse(href) + query_params = parse_qs(parsed.query) + target_url = query_params.get("uddg", [None])[0] + if target_url: + target_url = unquote(target_url) + if target_url.startswith("//"): + target_url = "https:" + target_url + elif not target_url.startswith(("http://", "https://")): + target_url = "https://" + target_url + results.append((title, target_url)) + else: + print(f"Skipping invalid redirect URL: {href}") + else: + if href and not href.startswith(("javascript:", "#")): + if href.startswith("//"): + href = "https:" + href + elif not href.startswith(("http://", "https://")): + href = "https://" + href + results.append((title, href)) + else: + print(f"Skipping invalid URL: {href}") return results def fetch_webpage(self, url): + """Fetch a webpage, validating the URL first.""" + if not url.startswith(("http://", "https://")): + print(f"Invalid URL scheme: {url}") + return None try: - time.sleep(random.uniform(1, 3)) + if not self._can_fetch(url): + return None + domain = self._get_domain(url) + self._enforce_crawl_delay(domain) r = self.session.get(url, timeout=15) r.raise_for_status() return r.text - except Exception as e: + except requests.exceptions.RequestException as e: print(f"Error fetching {url}: {e}") return None - def _fanout_call(self, messages, max_tokens): - def call_client(client): + def _fanout_call(self, messages, max_tokens, prefer_hermes=False): + """Call Hermes endpoints, preferring hermes.ai.unturf.com for aggregate_and_answer.""" + + def call_client(client, endpoint): try: + adjusted_max_tokens = max_tokens + if endpoint == "https://hermes2.ai.unturf.com/v1": + input_tokens = sum(len(m["content"]) // 4 + 1 for m in messages) + if input_tokens + max_tokens > HERMES2_TOKEN_LIMIT: + adjusted_max_tokens = max( + 1000, HERMES2_TOKEN_LIMIT - input_tokens + ) + print( + f"Adjusted max_tokens to {adjusted_max_tokens} for {endpoint} due to context limit" + ) response = client.chat.completions.create( model=self.model, messages=messages, temperature=0, - max_tokens=max_tokens + max_tokens=adjusted_max_tokens, ) - return response.choices[0].message.content - except Exception: - return None + return response.choices[0].message.content, endpoint + except Exception as e: + print(f"Error calling endpoint {endpoint}: {e}") + return None, endpoint - with ThreadPoolExecutor(max_workers=len(self.clients)) as executor: - futures = [executor.submit(call_client, c) for c in self.clients] - for future in as_completed(futures): - content = future.result() - if content is not None: - for f in futures: - if not f.done(): - f.cancel() - return content + if prefer_hermes: + hermes_endpoint = "https://hermes.ai.unturf.com/v1" + hermes_client = next( + ( + c + for c, e in zip(self.clients, self.hermes_endpoints) + if e == hermes_endpoint + ), + None, + ) + if hermes_client: + print(f"Attempting {hermes_endpoint} for final answer") + result, used_endpoint = call_client(hermes_client, hermes_endpoint) + if result is not None: + print(f"Success using {used_endpoint}") + return result + print(f"Fallback: {hermes_endpoint} failed, trying other endpoints") + other_clients = [ + (c, e) + for c, e in zip(self.clients, self.hermes_endpoints) + if e != hermes_endpoint + ] + with ThreadPoolExecutor(max_workers=len(other_clients)) as executor: + futures = [ + executor.submit(call_client, c, e) for c, e in other_clients + ] + for future in as_completed(futures): + result, used_endpoint = future.result() + if result is not None: + print(f"Success using {used_endpoint}") + return result + else: + print( + f"Warning: {hermes_endpoint} not in endpoints, using parallel execution" + ) + else: + with ThreadPoolExecutor(max_workers=len(self.clients)) as executor: + futures = [ + executor.submit(call_client, c, e) + for c, e in zip(self.clients, self.hermes_endpoints) + ] + for future in as_completed(futures): + result, used_endpoint = future.result() + if result is not None: + print(f"Success using {used_endpoint}") + return result raise RuntimeError("All Hermes endpoints failed.") def extract_with_hermes(self, html): - text = BeautifulSoup(html, 'html.parser').get_text(separator='\n') + """Extract main article content using Hermes.""" + text = BeautifulSoup(html, "html.parser").get_text(separator="\n") if len(text) > MAX_HTML_INPUT_CHARS: text = text[:MAX_HTML_INPUT_CHARS] messages = [ - {"role": "system", "content": ( - "Extract the main article content, preserving formatting and structure, " - "excluding ads and navigation. Return only the full text." - )}, - {"role": "user", "content": text} + { + "role": "system", + "content": ( + "Extract the main article content, preserving formatting and structure, " + "excluding ads and navigation. Return only the full text." + ), + }, + {"role": "user", "content": text}, ] try: - return self._fanout_call(messages, max_tokens=10000) + return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS) except RuntimeError: - shortened = text[:MAX_HTML_INPUT_CHARS//2] - messages[1]['content'] = shortened - return self._fanout_call(messages, max_tokens=5000) + shortened = text[: MAX_HTML_INPUT_CHARS // 2] + messages[1]["content"] = shortened + return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS // 2) def summarize_with_hermes(self, content): + """Summarize content as bullet points, used only when token limit is exceeded.""" messages = [ - {"role": "system", "content": ( - "Summarize the following article concisely as bullet points, keeping all factual details and structure. Avoid hallucination." - )}, - {"role": "user", "content": content} + { + "role": "system", + "content": ( + "Summarize the following article concisely as bullet points, keeping all factual details and structure. Avoid hallucination." + ), + }, + {"role": "user", "content": content}, ] - return self._fanout_call(messages, max_tokens=10000) + return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS) def cache_article(self, url, title, html, extracted, summary): - art = Article( - url=url, - title=title, - raw_html=html, - extracted_content=extracted, - summary=summary, - fetched_at=datetime.now(timezone.utc) - ) - self.db.add(art) - self.db.commit() + """Cache article in database, using extracted_content as summary.""" + with self.SessionLocal() as db: + art = Article( + url=url, + title=title, + raw_html=html, + extracted_content=extracted, + summary=summary, # Stores extracted_content unless summarized + fetched_at=datetime.now(timezone.utc), + ) + db.add(art) + try: + db.commit() + except Exception as e: + db.rollback() + print(f"Error caching article {url}: {e}") + raise def process_url(self, title_url): + """Process a URL, extracting content but not summarizing.""" title, url = title_url - if self.db.query(Article).filter_by(url=url).first(): - print(f"Already cached: {url}") - return - print(f"Fetching: {url}") - html = self.fetch_webpage(url) - if not html: - return - extracted = self.extract_with_hermes(html) - summary = self.summarize_with_hermes(extracted) - self.cache_article(url, title, html, extracted, summary) + with self.SessionLocal() as db: + if db.query(Article).filter_by(url=url).first(): + print(f"Already cached: {url}") + return + print(f"Fetching: {url}") + html = self.fetch_webpage(url) + if not html: + return + extracted = self.extract_with_hermes(html) + # Skip summarization; use extracted content as summary + self.cache_article(url, title, html, extracted, extracted) def aggregate_and_answer(self, query): - summaries = [art.summary for art in self.db.query(Article).all()] + """Aggregate article content and generate an answer, summarizing if needed.""" + with self.SessionLocal() as db: + articles = db.query(Article).all() + content_list = [ + (art.summary, art.extracted_content) for art in articles + ] # summary is extracted_content + + def estimate_tokens(text): + return len(text) // 4 + 1 + + # Try with full extracted content + combined = "\n\n".join(content for content, _ in content_list) + prompt_template = f"Based on these article contents:\n{{}}\n\nProvide a comprehensive answer to: {query}" + template_tokens = estimate_tokens(prompt_template.format("")) + combined_tokens = estimate_tokens(combined) + + if combined_tokens + template_tokens <= HERMES_INPUT_TOKENS: + print(f"Using full content ({combined_tokens} tokens)") + messages = [{"role": "user", "content": prompt_template.format(combined)}] + return self._fanout_call( + messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True + ) + + # Summarize if token limit exceeded + print( + f"Warning: Input exceeds token limit ({combined_tokens + template_tokens} tokens). Summarizing content." + ) + summaries = [] + for _, extracted_content in content_list: + summary = self.summarize_with_hermes(extracted_content) + summaries.append(summary) + combined = "\n\n".join(summaries) - messages = [{"role": "user", "content": ( - f"Based on these article summaries:\n{combined}\n\n" - f"Provide a comprehensive answer to: {query}" - )}] - return self._fanout_call(messages, max_tokens=30000) + combined_tokens = estimate_tokens(combined) + if combined_tokens + template_tokens > HERMES_INPUT_TOKENS: + print( + f"Warning: Summarized input still exceeds token limit ({combined_tokens + template_tokens} tokens). Truncating." + ) + max_chars = (HERMES_INPUT_TOKENS - template_tokens) * 4 + combined = combined[:max_chars] + last_newline = combined.rfind("\n\n", 0, max_chars) + if last_newline != -1: + combined = combined[:last_newline] + combined_tokens = estimate_tokens(combined) + print(f"Truncated to {combined_tokens} tokens.") + + messages = [{"role": "user", "content": prompt_template.format(combined)}] + return self._fanout_call( + messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True + ) def run(self, query, max_results=10): hits = self.search_duckduckgo(query, max_results) - # Parallel fetch/extract/summarize using endpoints count - with ThreadPoolExecutor(max_workers=len(self.clients)) as executor: + with ThreadPoolExecutor(max_workers=4) as executor: futures = [executor.submit(self.process_url, hit) for hit in hits] for future in as_completed(futures): future.result() print("Generating comprehensive answer...") print(self.aggregate_and_answer(query)) + if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("query", help="Search and deep-query prompt") parser.add_argument("--api-key", default="dummy-api-key", help="OpenAI API key") - parser.add_argument("--model", default="adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic", help="Model ID") - parser.add_argument("--max-results", type=int, default=10, help="Max search results to process") + parser.add_argument( + "--model", + default="adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic", + help="Model ID", + ) + parser.add_argument( + "--max-results", type=int, default=10, help="Max search results to process" + ) + parser.add_argument( + "--user-agent-append", default="", help="String to append to default user agent" + ) args = parser.parse_args() crawler = SQLAlchemyDuckDuckGoCrawler( - api_key=args.api_key, - model=args.model + api_key=args.api_key, model=args.model, user_agent_append=args.user_agent_append ) crawler.run(args.query, args.max_results)