diff --git a/.gitignore b/.gitignore index 8fce603..e181ae3 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ data/ +chroma_db/ +crawler.log diff --git a/duck_duck_go_hermes_unturf.py b/duck_duck_go_hermes_unturf.py index e26322f..6abfad4 100644 --- a/duck_duck_go_hermes_unturf.py +++ b/duck_duck_go_hermes_unturf.py @@ -8,7 +8,9 @@ from datetime import datetime, timezone from concurrent.futures import ThreadPoolExecutor, as_completed import urllib.robotparser import threading - +import logging +import chromadb +from sentence_transformers import SentenceTransformer from sqlalchemy import ( create_engine, Column, @@ -22,6 +24,14 @@ from sqlalchemy.orm import declarative_base, sessionmaker from openai import OpenAI from bs4 import BeautifulSoup +# 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 context window @@ -32,8 +42,8 @@ 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_TOKEN_LIMIT = 80000 +HERMES2_TOKEN_LIMIT = 48000 HERMES_INPUT_TOKENS = 72000 HERMES_COMPLETION_TOKENS = 8000 HERMES2_INPUT_TOKENS = 40000 @@ -83,6 +93,7 @@ class SQLAlchemyDuckDuckGoCrawler: self.session.headers.update({"User-Agent": self.user_agent}) # Initialize DB + os.makedirs(os.path.dirname(db_path), exist_ok=True) self.engine = create_engine( f"sqlite:///{self.db_path}", echo=False, @@ -94,6 +105,11 @@ class SQLAlchemyDuckDuckGoCrawler: bind=self.engine, autoflush=False, autocommit=False ) + # Initialize ChromaDB + self.chroma_client = chromadb.PersistentClient(path="./chroma_db") + self.collection = self.chroma_client.get_or_create_collection(name="articles") + self.embedder = SentenceTransformer("all-MiniLM-L6-v2") + # Robots.txt parser and domain tracking self.robot_parsers = {} self.domain_last_fetched = {} @@ -108,10 +124,10 @@ class SQLAlchemyDuckDuckGoCrawler: def _fetch_robots_txt(self, domain): """Fetch and parse robots.txt for a domain.""" if domain in self.robot_parsers: - print(f"Using cached robots.txt for {domain}") + logger.info(f"Using cached robots.txt for {domain}") return self.robot_parsers[domain] - print(f"Fetching robots.txt for {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) @@ -126,7 +142,7 @@ class SQLAlchemyDuckDuckGoCrawler: delay if delay is not None else DEFAULT_CRAWL_DELAY ) except Exception as e: - print(f"Could not fetch robots.txt for {domain}: {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 @@ -141,10 +157,10 @@ class SQLAlchemyDuckDuckGoCrawler: can_fetch = parser.can_fetch(self.user_agent, url) if not can_fetch: - print(f"Blocked by robots.txt: {url}") + logger.warning(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}:") + logger.info(f"Relevant robots.txt rules for {domain}:") current_user_agent = None relevant_rules = [] for line in robots_content.splitlines(): @@ -166,13 +182,13 @@ class SQLAlchemyDuckDuckGoCrawler: f" User-agent: {current_user_agent}\n Disallow: {rule}" ) if relevant_rules: - print("\n".join(relevant_rules)) + logger.info("\n".join(relevant_rules)) else: - print( + logger.info( " No specific Disallow rules found; may be blocked by a broad rule." ) else: - print(" No robots.txt content available to display rules.") + logger.info(" No robots.txt content available to display rules.") return can_fetch def _enforce_crawl_delay(self, domain): @@ -182,7 +198,7 @@ class SQLAlchemyDuckDuckGoCrawler: elapsed = time.time() - last_fetched if elapsed < delay: sleep_time = delay - elapsed - print( + logger.info( f"Delaying crawl for {domain} by {sleep_time:.2f} seconds due to crawl delay" ) time.sleep(sleep_time) @@ -213,7 +229,7 @@ class SQLAlchemyDuckDuckGoCrawler: target_url = "https://" + target_url results.append((title, target_url)) else: - print(f"Skipping invalid redirect URL: {href}") + logger.warning(f"Skipping invalid redirect URL: {href}") else: if href and not href.startswith(("javascript:", "#")): if href.startswith("//"): @@ -222,13 +238,13 @@ class SQLAlchemyDuckDuckGoCrawler: href = "https://" + href results.append((title, href)) else: - print(f"Skipping invalid URL: {href}") + logger.warning(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}") + logger.error(f"Invalid URL scheme: {url}") return None try: if not self._can_fetch(url): @@ -239,7 +255,7 @@ class SQLAlchemyDuckDuckGoCrawler: r.raise_for_status() return r.text except requests.exceptions.RequestException as e: - print(f"Error fetching {url}: {e}") + logger.error(f"Error fetching {url}: {e}") return None def _fanout_call(self, messages, max_tokens, prefer_hermes=False): @@ -254,7 +270,7 @@ class SQLAlchemyDuckDuckGoCrawler: adjusted_max_tokens = max( 1000, HERMES2_TOKEN_LIMIT - input_tokens ) - print( + logger.info( f"Adjusted max_tokens to {adjusted_max_tokens} for {endpoint} due to context limit" ) response = client.chat.completions.create( @@ -265,7 +281,7 @@ class SQLAlchemyDuckDuckGoCrawler: ) return response.choices[0].message.content, endpoint except Exception as e: - print(f"Error calling endpoint {endpoint}: {e}") + logger.error(f"Error calling endpoint {endpoint}: {e}") return None, endpoint if prefer_hermes: @@ -279,12 +295,12 @@ class SQLAlchemyDuckDuckGoCrawler: None, ) if hermes_client: - print(f"Attempting {hermes_endpoint} for final answer") + logger.info(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}") + logger.info(f"Success using {used_endpoint}") return result - print(f"Fallback: {hermes_endpoint} failed, trying other endpoints") + logger.warning(f"Fallback: {hermes_endpoint} failed, trying other endpoints") other_clients = [ (c, e) for c, e in zip(self.clients, self.hermes_endpoints) @@ -297,10 +313,10 @@ class SQLAlchemyDuckDuckGoCrawler: for future in as_completed(futures): result, used_endpoint = future.result() if result is not None: - print(f"Success using {used_endpoint}") + logger.info(f"Success using {used_endpoint}") return result else: - print( + logger.warning( f"Warning: {hermes_endpoint} not in endpoints, using parallel execution" ) else: @@ -312,7 +328,7 @@ class SQLAlchemyDuckDuckGoCrawler: for future in as_completed(futures): result, used_endpoint = future.result() if result is not None: - print(f"Success using {used_endpoint}") + logger.info(f"Success using {used_endpoint}") return result raise RuntimeError("All Hermes endpoints failed.") @@ -351,8 +367,26 @@ class SQLAlchemyDuckDuckGoCrawler: ] return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS) + def _split_into_chunks(self, text, max_words=500): + """Split text into chunks of approximately max_words.""" + 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 + def cache_article(self, url, title, html, extracted, summary): - """Cache article in database, using extracted_content as summary.""" + """Cache article in database and store chunks in ChromaDB.""" with self.SessionLocal() as db: art = Article( url=url, @@ -367,17 +401,30 @@ class SQLAlchemyDuckDuckGoCrawler: db.commit() except Exception as e: db.rollback() - print(f"Error caching article {url}: {e}") + logger.error(f"Error caching article {url}: {e}") raise + # Store chunks in ChromaDB + chunks = self._split_into_chunks(extracted) + embeddings = self.embedder.encode(chunks, show_progress_bar=False) + for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): + chunk_id = f"{url}_chunk_{i}" + self.collection.upsert( + ids=[chunk_id], + embeddings=[embedding.tolist()], + documents=[chunk], + metadatas=[{"url": url, "title": title, "chunk_index": i}], + ) + logger.info(f"Stored {len(chunks)} chunks for {url} in ChromaDB") + def process_url(self, title_url): """Process a URL, extracting content but not summarizing.""" title, url = title_url with self.SessionLocal() as db: if db.query(Article).filter_by(url=url).first(): - print(f"Already cached: {url}") + logger.info(f"Already cached: {url}") return - print(f"Fetching: {url}") + logger.info(f"Fetching: {url}") html = self.fetch_webpage(url) if not html: return @@ -386,52 +433,44 @@ class SQLAlchemyDuckDuckGoCrawler: self.cache_article(url, title, html, extracted, extracted) def aggregate_and_answer(self, query): - """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 + """Aggregate relevant article chunks from ChromaDB and generate an answer.""" + # Embed the query + query_embedding = self.embedder.encode([query], show_progress_bar=False)[0] + # Query ChromaDB for relevant chunks + results = self.collection.query( + query_embeddings=[query_embedding.tolist()], + n_results=50, # Retrieve more to filter by token count + ) + + # Collect chunks up to token limit 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." + combined_content = [] + total_tokens = 0 + prompt_template = ( + f"Based on these article excerpts:\n{{}}\n\nProvide a comprehensive answer to: {query}" ) - summaries = [] - for _, extracted_content in content_list: - summary = self.summarize_with_hermes(extracted_content) - summaries.append(summary) + template_tokens = estimate_tokens(prompt_template.format("")) - combined = "\n\n".join(summaries) - 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.") + for doc, metadata in zip(results["documents"][0], results["metadatas"][0]): + chunk = doc + chunk_tokens = estimate_tokens(chunk) + if total_tokens + chunk_tokens + template_tokens <= HERMES_INPUT_TOKENS: + combined_content.append( + f"From {metadata['url']} (Title: {metadata['title']}):\n{chunk}" + ) + total_tokens += chunk_tokens + else: + break + if not combined_content: + logger.warning("No relevant chunks found within token limit.") + return "No relevant information found to answer the query." + + combined = "\n\n".join(combined_content) + logger.info(f"Using {total_tokens} tokens from {len(combined_content)} chunks") messages = [{"role": "user", "content": prompt_template.format(combined)}] return self._fanout_call( messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True @@ -443,8 +482,9 @@ class SQLAlchemyDuckDuckGoCrawler: 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)) + logger.info("Generating comprehensive answer...") + result = self.aggregate_and_answer(query) + print(result) if __name__ == "__main__":