679 lines
28 KiB
Python
679 lines
28 KiB
Python
import os
|
|
import requests
|
|
import random
|
|
import time
|
|
import re
|
|
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 logging
|
|
from sqlalchemy import (
|
|
create_engine,
|
|
Column,
|
|
Integer,
|
|
String,
|
|
Text,
|
|
DateTime,
|
|
UniqueConstraint,
|
|
)
|
|
from sqlalchemy.orm import declarative_base, sessionmaker
|
|
from openai import OpenAI
|
|
from bs4 import BeautifulSoup
|
|
from sklearn.feature_extraction.text import TfidfVectorizer
|
|
from sklearn.metrics.pairwise import cosine_similarity
|
|
import numpy as np
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s",
|
|
handlers=[logging.StreamHandler(), logging.FileHandler("crawler.log")],
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Default Hermes endpoints
|
|
DEFAULT_HERMES_ENDPOINTS = [
|
|
"https://hermes.ai.unturf.com/v1", # 80K tokens
|
|
"https://hermes2.ai.unturf.com/v1", # 48K tokens
|
|
]
|
|
# Max characters to send to extraction
|
|
MAX_HTML_INPUT_CHARS = 50000
|
|
# Default crawl delay
|
|
DEFAULT_CRAWL_DELAY = 2.0
|
|
# Token limits for Hermes
|
|
HERMES_TOKEN_LIMIT = 80000
|
|
HERMES_INPUT_TOKENS = 72000
|
|
HERMES_COMPLETION_TOKENS = 8000
|
|
HERMES2_TOKEN_LIMIT = 48000
|
|
HERMES2_INPUT_TOKENS = 40000
|
|
HERMES2_COMPLETION_TOKENS = 8000
|
|
# Max words for chunking (when needed)
|
|
MAX_WORDS_LONG = 10000
|
|
MAX_WORDS_SHORT = 5000
|
|
# Context window limit (words, ~4 chars per token)
|
|
CONTEXT_WINDOW_WORDS = 20000 # ~80K tokens
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
class Article(Base):
|
|
__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"),)
|
|
|
|
|
|
class SQLAlchemyDuckDuckGoCrawler:
|
|
def __init__(
|
|
self,
|
|
api_key,
|
|
model,
|
|
db_path="data/articles.db",
|
|
hermes_endpoints=None,
|
|
user_agent_append="",
|
|
):
|
|
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.api_key = api_key
|
|
self.model = model
|
|
self.db_path = db_path
|
|
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": self.user_agent})
|
|
os.makedirs(os.path.dirname(db_path), exist_ok=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)
|
|
self.SessionLocal = sessionmaker(
|
|
bind=self.engine, autoflush=False, autocommit=False
|
|
)
|
|
self.robot_parsers = {}
|
|
self.domain_last_fetched = {}
|
|
self.domain_crawl_delays = {}
|
|
self.robots_txt_content = {}
|
|
|
|
def _get_domain(self, url):
|
|
parsed = urlparse(url)
|
|
return parsed.netloc
|
|
|
|
def _fetch_robots_txt(self, domain):
|
|
if domain in self.robot_parsers:
|
|
logger.info(f"Using cached robots.txt for {domain}")
|
|
return self.robot_parsers[domain]
|
|
logger.info(f"Fetching robots.txt for {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:
|
|
logger.error(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):
|
|
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:
|
|
logger.warning(f"Blocked by robots.txt: {url}")
|
|
robots_content = self.robots_txt_content.get(domain, "")
|
|
if robots_content:
|
|
logger.info(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:
|
|
logger.info("\n".join(relevant_rules))
|
|
else:
|
|
logger.info(
|
|
"No specific Disallow rules found; may be blocked by a broad rule."
|
|
)
|
|
else:
|
|
logger.info("No robots.txt content available to display rules.")
|
|
return can_fetch
|
|
|
|
def _enforce_crawl_delay(self, 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
|
|
logger.info(
|
|
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):
|
|
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")
|
|
results = []
|
|
for link in soup.select(".result__title a"):
|
|
if len(results) >= max_results:
|
|
break
|
|
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:
|
|
logger.warning(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:
|
|
logger.warning(f"Skipping invalid URL: {href}")
|
|
return results
|
|
|
|
def fetch_webpage(self, url):
|
|
if not url.startswith(("http://", "https://")):
|
|
logger.error(f"Invalid URL scheme: {url}")
|
|
return None
|
|
try:
|
|
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 requests.exceptions.RequestException as e:
|
|
logger.error(f"Error fetching {url}: {e}")
|
|
return None
|
|
|
|
def _fanout_call(self, messages, max_tokens, prefer_hermes=False):
|
|
def estimate_input_tokens(messages):
|
|
# More accurate: ~5 chars per word, 4 chars per token
|
|
total_chars = sum(len(m["content"]) for m in messages)
|
|
return (total_chars // 5 + 1) // 4 + 1
|
|
|
|
def call_client(client, endpoint, messages, max_tokens):
|
|
try:
|
|
input_tokens = estimate_input_tokens(messages)
|
|
total_tokens = input_tokens + max_tokens
|
|
if (
|
|
endpoint == "https://hermes2.ai.unturf.com/v1"
|
|
and total_tokens > HERMES2_TOKEN_LIMIT
|
|
):
|
|
logger.warning(
|
|
f"Request exceeds {HERMES2_TOKEN_LIMIT} tokens for {endpoint}: {total_tokens}"
|
|
)
|
|
return None, endpoint
|
|
if (
|
|
endpoint == "https://hermes.ai.unturf.com/v1"
|
|
and total_tokens > HERMES_TOKEN_LIMIT
|
|
):
|
|
logger.warning(
|
|
f"Request exceeds {HERMES_TOKEN_LIMIT} tokens for {endpoint}: {total_tokens}"
|
|
)
|
|
return None, endpoint
|
|
response = client.chat.completions.create(
|
|
model=self.model,
|
|
messages=messages,
|
|
temperature=0,
|
|
max_tokens=max_tokens,
|
|
)
|
|
return response.choices[0].message.content, endpoint
|
|
except Exception as e:
|
|
logger.error(f"Error calling endpoint {endpoint}: {e}")
|
|
return None, endpoint
|
|
|
|
# Prefer hermes.ai.unturf.com/v1 (80K limit)
|
|
hermes_endpoint = "https://hermes.ai.unturf.com/v1"
|
|
hermes2_endpoint = "https://hermes2.ai.unturf.com/v1"
|
|
hermes_client = next(
|
|
(
|
|
c
|
|
for c, e in zip(self.clients, self.hermes_endpoints)
|
|
if e == hermes_endpoint
|
|
),
|
|
None,
|
|
)
|
|
hermes2_client = next(
|
|
(
|
|
c
|
|
for c, e in zip(self.clients, self.hermes_endpoints)
|
|
if e == hermes2_endpoint
|
|
),
|
|
None,
|
|
)
|
|
|
|
if prefer_hermes and hermes_client:
|
|
logger.info(f"Attempting {hermes_endpoint} for final answer")
|
|
result, used_endpoint = call_client(
|
|
hermes_client, hermes_endpoint, messages, max_tokens
|
|
)
|
|
if result is not None:
|
|
logger.info(f"Success using {used_endpoint}")
|
|
return result
|
|
logger.warning(
|
|
f"Fallback: {hermes_endpoint} failed, trying {hermes2_endpoint}"
|
|
)
|
|
if hermes2_client:
|
|
result, used_endpoint = call_client(
|
|
hermes2_client, hermes2_endpoint, messages, max_tokens
|
|
)
|
|
if result is not None:
|
|
logger.info(f"Success using {used_endpoint}")
|
|
return result
|
|
else:
|
|
# Try hermes.ai.unturf.com/v1 first
|
|
if hermes_client:
|
|
logger.info(f"Attempting {hermes_endpoint}")
|
|
result, used_endpoint = call_client(
|
|
hermes_client, hermes_endpoint, messages, max_tokens
|
|
)
|
|
if result is not None:
|
|
logger.info(f"Success using {used_endpoint}")
|
|
return result
|
|
# Retry with hermes2 if hermes fails or if hermes2 is preferred
|
|
if hermes2_client:
|
|
logger.info(f"Attempting {hermes2_endpoint}")
|
|
result, used_endpoint = call_client(
|
|
hermes2_client, hermes2_endpoint, messages, max_tokens
|
|
)
|
|
if result is not None:
|
|
logger.info(f"Success using {used_endpoint}")
|
|
return result
|
|
# If hermes2 fails due to context length, retry with hermes
|
|
if hermes_client and "maximum context length" in str(result):
|
|
logger.info(
|
|
f"Retrying with {hermes_endpoint} due to context length error"
|
|
)
|
|
result, used_endpoint = call_client(
|
|
hermes_client, hermes_endpoint, messages, max_tokens
|
|
)
|
|
if result is not None:
|
|
logger.info(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")
|
|
if len(text) > MAX_HTML_INPUT_CHARS:
|
|
text = text[:MAX_HTML_INPUT_CHARS]
|
|
messages = [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"Extract the main article content, preserving all formatting, structure, and relevant details, "
|
|
"including headers, paragraphs, lists, and key text. Exclude only ads, navigation, and unrelated boilerplate. "
|
|
"Maximize content retention to capture comprehensive information, ensuring no relevant text is omitted."
|
|
),
|
|
},
|
|
{"role": "user", "content": text},
|
|
]
|
|
try:
|
|
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=HERMES_COMPLETION_TOKENS // 2)
|
|
|
|
def summarize_with_hermes(self, content):
|
|
messages = [
|
|
{
|
|
"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=HERMES_COMPLETION_TOKENS)
|
|
|
|
def _split_into_chunks(self, text, content_length):
|
|
max_words = MAX_WORDS_LONG if content_length > 5000 else MAX_WORDS_SHORT
|
|
logger.info(f"Using max_words={max_words} for content length={content_length}")
|
|
words = text.split()
|
|
chunks = []
|
|
current_chunk = []
|
|
current_word_count = 0
|
|
for word in words:
|
|
current_chunk.append(word)
|
|
current_word_count += 1
|
|
if current_word_count >= max_words:
|
|
chunks.append(" ".join(current_chunk))
|
|
current_chunk = []
|
|
current_word_count = 0
|
|
if current_chunk:
|
|
chunks.append(" ".join(current_chunk))
|
|
return chunks if len(chunks) > 1 else [text]
|
|
|
|
def cache_article(self, url, title, html, extracted, summary):
|
|
with self.SessionLocal() as db:
|
|
art = Article(
|
|
url=url,
|
|
title=title,
|
|
raw_html=html,
|
|
extracted_content=extracted,
|
|
summary=summary,
|
|
fetched_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(art)
|
|
try:
|
|
db.commit()
|
|
logger.info(f"Cached article: {url}")
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error caching article {url}: {e}")
|
|
raise
|
|
|
|
def process_url(self, title_url):
|
|
title, url = title_url
|
|
with self.SessionLocal() as db:
|
|
if db.query(Article).filter_by(url=url).first():
|
|
logger.info(f"Already cached: {url}")
|
|
return
|
|
logger.info(f"Fetching: {url}")
|
|
html = self.fetch_webpage(url)
|
|
if not html:
|
|
return
|
|
extracted = self.extract_with_hermes(html)
|
|
self.cache_article(url, title, html, extracted, extracted)
|
|
|
|
def _extract_keywords_with_hermes(self, query):
|
|
messages = [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"Analyze the following query and extract a list of up to 5 relevant keywords or phrases "
|
|
"that capture the main topics or entities. Focus on nouns, proper nouns, and key concepts. "
|
|
"Avoid generic terms like 'what', 'is', or redundant variations. "
|
|
"Return the keywords as a comma-separated string."
|
|
),
|
|
},
|
|
{"role": "user", "content": query},
|
|
]
|
|
try:
|
|
result = self._fanout_call(messages, max_tokens=100)
|
|
keywords = [k.strip() for k in result.split(",") if k.strip()]
|
|
if len(keywords) > 5:
|
|
keywords = keywords[:5]
|
|
logger.info(f"Extracted keywords for query '{query}': {keywords}")
|
|
return keywords
|
|
except RuntimeError:
|
|
logger.warning(
|
|
"Failed to extract keywords with Hermes, falling back to query split"
|
|
)
|
|
return query.lower().split()
|
|
|
|
def _search_sqlite(self, query, keywords, max_results=10):
|
|
with self.SessionLocal() as db:
|
|
articles = db.query(Article).all()
|
|
if not articles:
|
|
logger.warning("No articles found in SQLite cache.")
|
|
return []
|
|
|
|
# Keyword-based filtering
|
|
core_keywords = keywords + query.lower().split()
|
|
relevant_articles = []
|
|
for article in articles:
|
|
content_lower = article.extracted_content.lower()
|
|
keyword_score = sum(
|
|
1 for kw in core_keywords if kw.lower() in content_lower
|
|
)
|
|
if keyword_score > 0:
|
|
relevant_articles.append((article, keyword_score))
|
|
|
|
# Sort by keyword score
|
|
relevant_articles.sort(key=lambda x: x[1], reverse=True)
|
|
top_articles = relevant_articles[:max_results]
|
|
|
|
# TF-IDF similarity for refined ranking
|
|
if top_articles:
|
|
documents = [article.extracted_content for article, _ in top_articles]
|
|
vectorizer = TfidfVectorizer(stop_words="english")
|
|
try:
|
|
tfidf_matrix = vectorizer.fit_transform(documents + [query])
|
|
similarities = cosine_similarity(
|
|
tfidf_matrix[-1], tfidf_matrix[:-1]
|
|
)[0]
|
|
scored_articles = [
|
|
(article, score, keyword_score)
|
|
for (article, keyword_score), score in zip(
|
|
top_articles, similarities
|
|
)
|
|
]
|
|
scored_articles.sort(
|
|
key=lambda x: 0.5 * x[1]
|
|
+ 0.5 * (x[2] / max(1, max(s[2] for s in scored_articles))),
|
|
reverse=True,
|
|
)
|
|
return [
|
|
{
|
|
"url": article.url,
|
|
"title": article.title,
|
|
"content": article.extracted_content,
|
|
}
|
|
for article, _, _ in scored_articles
|
|
]
|
|
except ValueError as e:
|
|
logger.warning(
|
|
f"TF-IDF failed: {e}, falling back to keyword ranking"
|
|
)
|
|
return [
|
|
{
|
|
"url": article.url,
|
|
"title": article.title,
|
|
"content": article.extracted_content,
|
|
}
|
|
for article, _ in top_articles
|
|
]
|
|
return []
|
|
|
|
def aggregate_and_answer(self, query, search_results=None, query_keywords=None):
|
|
# Search SQLite cache
|
|
cache_results = self._search_sqlite(query, query_keywords)
|
|
logger.info(f"Retrieved {len(cache_results)} articles from SQLite cache")
|
|
|
|
def estimate_tokens(text):
|
|
# ~5 chars per word, 4 chars per token
|
|
return (len(text) // 5 + 1) // 4 + 1
|
|
|
|
combined_content = []
|
|
total_words = 0
|
|
total_tokens = 0
|
|
prompt_template = f"Based on these article excerpts:\n{{}}\n\nProvide a comprehensive answer to: {query}"
|
|
template_tokens = estimate_tokens(prompt_template.format(""))
|
|
|
|
# Add cached content
|
|
for result in cache_results:
|
|
content = result["content"]
|
|
word_count = len(content.split())
|
|
content_tokens = estimate_tokens(content)
|
|
endpoint_limit = (
|
|
HERMES2_INPUT_TOKENS
|
|
if total_tokens + content_tokens + template_tokens
|
|
> HERMES2_INPUT_TOKENS
|
|
else HERMES_INPUT_TOKENS
|
|
)
|
|
if (
|
|
total_words + word_count <= CONTEXT_WINDOW_WORDS
|
|
and total_tokens + content_tokens + template_tokens <= endpoint_limit
|
|
):
|
|
combined_content.append(
|
|
f"From {result['url']} (Title: {result['title']}):\n{content}"
|
|
)
|
|
total_words += word_count
|
|
total_tokens += content_tokens
|
|
logger.info(
|
|
f"Included article {result['url']} (words: {word_count}, tokens: {content_tokens})"
|
|
)
|
|
else:
|
|
# Trim or chunk content
|
|
if total_tokens + template_tokens < endpoint_limit:
|
|
remaining_tokens = endpoint_limit - total_tokens - template_tokens
|
|
remaining_words = min(word_count, remaining_tokens * 4 // 5)
|
|
trimmed_content = " ".join(content.split()[:remaining_words])
|
|
trimmed_tokens = estimate_tokens(trimmed_content)
|
|
if trimmed_tokens <= remaining_tokens:
|
|
combined_content.append(
|
|
f"From {result['url']} (Title: {result['title']}):\n{trimmed_content}"
|
|
)
|
|
total_words += remaining_words
|
|
total_tokens += trimmed_tokens
|
|
logger.info(
|
|
f"Included trimmed article {result['url']} (words: {remaining_words}, tokens: {trimmed_tokens})"
|
|
)
|
|
else:
|
|
# Chunk if trimming still exceeds
|
|
chunks = self._split_into_chunks(content, len(content))
|
|
for chunk in chunks:
|
|
chunk_words = len(chunk.split())
|
|
chunk_tokens = estimate_tokens(chunk)
|
|
if (
|
|
total_words + chunk_words <= CONTEXT_WINDOW_WORDS
|
|
and total_tokens + chunk_tokens + template_tokens
|
|
<= endpoint_limit
|
|
):
|
|
combined_content.append(
|
|
f"From {result['url']} (Title: {result['title']}):\n{chunk}"
|
|
)
|
|
total_words += chunk_words
|
|
total_tokens += chunk_tokens
|
|
logger.info(
|
|
f"Included chunk from {result['url']} (words: {chunk_words}, tokens: {chunk_tokens})"
|
|
)
|
|
else:
|
|
logger.info(
|
|
f"Skipping chunk from {result['url']} (exceeds limit: {total_tokens + chunk_tokens + template_tokens})"
|
|
)
|
|
break
|
|
else:
|
|
logger.info(
|
|
f"Skipping article {result['url']} (exceeds limit: {total_tokens + content_tokens + template_tokens})"
|
|
)
|
|
|
|
if not combined_content and search_results:
|
|
logger.warning(
|
|
"No relevant content found in cache, using search results as fallback"
|
|
)
|
|
fallback_content = "\n".join(
|
|
f"- {title}: {url}" for title, url in search_results
|
|
)
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": f"Based on these search results:\n{fallback_content}\n\nAnswer: {query}",
|
|
}
|
|
]
|
|
try:
|
|
return self._fanout_call(
|
|
messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True
|
|
)
|
|
except RuntimeError:
|
|
return "No relevant information found, and fallback answer generation failed."
|
|
|
|
if not combined_content:
|
|
return "No relevant information found to answer the query."
|
|
|
|
combined = "\n\n".join(combined_content)
|
|
total_tokens = estimate_tokens(combined) + template_tokens
|
|
logger.info(
|
|
f"Using {total_tokens} tokens from {len(combined_content)} articles/chunks"
|
|
)
|
|
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):
|
|
query_keywords = self._extract_keywords_with_hermes(query)
|
|
hits = self.search_duckduckgo(query, max_results)
|
|
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()
|
|
logger.info("Generating comprehensive answer...")
|
|
result = self.aggregate_and_answer(
|
|
query, search_results=hits, query_keywords=query_keywords
|
|
)
|
|
print(result)
|
|
|
|
|
|
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(
|
|
"--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, user_agent_append=args.user_agent_append
|
|
)
|
|
crawler.run(args.query, args.max_results)
|