| """ |
| BeastBullet + RAG: Local Perplexity Clone |
| Build your own Perplexity for $0/month |
| """ |
|
|
| import requests |
| from typing import List, Dict |
| import json |
| from pathlib import Path |
|
|
| class WebRAG: |
| """Web search and retrieval for BeastBullet.""" |
| |
| def __init__(self, search_engine="duckduckgo"): |
| self.search_engine = search_engine |
| |
| def search_web(self, query: str, num_results: int = 5) -> List[Dict]: |
| """Search the web and return results.""" |
| |
| try: |
| from duckduckgo_search import DDGS |
| |
| results = [] |
| with DDGS() as ddgs: |
| for r in ddgs.text(query, max_results=num_results): |
| results.append({ |
| "title": r.get("title", ""), |
| "url": r.get("href", ""), |
| "snippet": r.get("body", "") |
| }) |
| return results |
| except ImportError: |
| print("⚠️ Install duckduckgo-search: pip install duckduckgo-search") |
| return [] |
| |
| def fetch_content(self, url: str) -> str: |
| """Fetch full content from URL.""" |
| try: |
| from bs4 import BeautifulSoup |
| |
| response = requests.get(url, timeout=5) |
| soup = BeautifulSoup(response.content, 'html.parser') |
| |
| |
| for script in soup(["script", "style"]): |
| script.decompose() |
| |
| text = soup.get_text() |
| lines = (line.strip() for line in text.splitlines()) |
| chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) |
| text = ' '.join(chunk for chunk in chunks if chunk) |
| |
| return text[:5000] |
| except: |
| return "" |
| |
| def search_and_retrieve(self, query: str) -> Dict: |
| """Search web and retrieve relevant content.""" |
| print(f"🔍 Searching: {query}") |
| |
| |
| results = self.search_web(query) |
| |
| if not results: |
| return {"sources": [], "context": ""} |
| |
| |
| sources = [] |
| context_parts = [] |
| |
| for i, result in enumerate(results[:3], 1): |
| print(f" 📄 Fetching: {result['title'][:50]}...") |
| content = self.fetch_content(result['url']) |
| |
| if content: |
| sources.append({ |
| "title": result['title'], |
| "url": result['url'], |
| "snippet": result['snippet'] |
| }) |
| context_parts.append(f"[Source {i}] {result['title']}\n{content[:1000]}") |
| |
| return { |
| "sources": sources, |
| "context": "\n\n".join(context_parts) |
| } |
|
|
|
|
| class LocalPerplexity: |
| """Local Perplexity clone using BeastBullet + RAG.""" |
| |
| def __init__(self): |
| from code.integration_example import TinyLlamaIntegration, ExpertLoader |
| |
| print("🧠 Loading BeastBullet...") |
| loader = ExpertLoader("experts/") |
| loader.load_encoder() |
| loader.load_all_experts() |
| |
| self.bb = TinyLlamaIntegration( |
| experts=loader.experts, |
| encoder=loader.encoder |
| ) |
| |
| self.rag = WebRAG() |
| print("✅ Local Perplexity ready!\n") |
| |
| def query(self, question: str) -> Dict: |
| """Query with web search + expert reasoning.""" |
| print("=" * 70) |
| print(f"Question: {question}") |
| print("=" * 70) |
| |
| |
| retrieval = self.rag.search_and_retrieve(question) |
| |
| if not retrieval["sources"]: |
| print("\n⚠️ No web results found, using BeastBullet knowledge only") |
| result = self.bb.query(question) |
| return result |
| |
| |
| context_query = f"""Based on the following sources, answer: {question} |
| |
| Sources: |
| {retrieval['context']} |
| |
| Provide a clear answer with citations.""" |
| |
| |
| result = self.bb.query(context_query) |
| |
| |
| result["sources"] = retrieval["sources"] |
| |
| return result |
| |
| def display_result(self, result: Dict): |
| """Display result in Perplexity-style format.""" |
| print("\n" + "=" * 70) |
| print("🧠 Answer:") |
| print("=" * 70) |
| print(result.get("answer", "No answer generated")) |
| |
| if "sources" in result and result["sources"]: |
| print("\n" + "=" * 70) |
| print("📚 Sources:") |
| print("=" * 70) |
| for i, source in enumerate(result["sources"], 1): |
| print(f"\n[{i}] {source['title']}") |
| print(f" {source['url']}") |
| print(f" {source['snippet'][:100]}...") |
| |
| print("\n" + "=" * 70) |
| print("📊 Metadata:") |
| print("=" * 70) |
| print(f"Experts Used: {', '.join(result.get('experts_used', []))}") |
| print(f"Confidence: {result.get('confidence', 0):.1%}") |
| print("=" * 70) |
|
|
|
|
| |
| |
| |
|
|
| def demo(): |
| """Run demo queries.""" |
| perplexity = LocalPerplexity() |
| |
| |
| queries = [ |
| "What are the latest AI developments in December 2025?", |
| "How does quantum entanglement work?", |
| "What is the current price of Bitcoin?", |
| ] |
| |
| for query in queries: |
| result = perplexity.query(query) |
| perplexity.display_result(result) |
| print("\n\n") |
|
|
|
|
| def interactive(): |
| """Interactive mode.""" |
| perplexity = LocalPerplexity() |
| |
| print("🧠 Local Perplexity - Interactive Mode") |
| print("=" * 70) |
| print("Ask any question. Type 'exit' to quit.") |
| print("=" * 70) |
| print() |
| |
| while True: |
| try: |
| question = input("You: ").strip() |
| |
| if question.lower() in ['exit', 'quit', 'q']: |
| print("\n👋 Goodbye!") |
| break |
| |
| if not question: |
| continue |
| |
| result = perplexity.query(question) |
| perplexity.display_result(result) |
| print("\n") |
| |
| except KeyboardInterrupt: |
| print("\n\n👋 Goodbye!") |
| break |
|
|
|
|
| if __name__ == "__main__": |
| import sys |
| |
| if len(sys.argv) > 1 and sys.argv[1] == "demo": |
| demo() |
| else: |
| interactive() |
|
|
