Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """ | |
| Research Agent - Web Search and Summarization Tool | |
| Deployed on Hugging Face Spaces with Gradio | |
| """ | |
| import re | |
| import urllib.parse | |
| from ddgs import DDGS | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from sentence_transformers import SentenceTransformer | |
| import numpy as np | |
| import time | |
| import gradio as gr | |
| # Configuration | |
| SEARCH_RESULTS = 6 | |
| PASSAGES_PER_PAGE = 4 | |
| EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2" | |
| TOP_PASSAGES = 5 | |
| SUMMARY_SENTENCES = 3 | |
| TIMEOUT = 8 | |
| def unwrap_ddg(url): | |
| """If DuckDuckGo returns a redirect wrapper, extract the real URL.""" | |
| try: | |
| parsed = urllib.parse.urlparse(url) | |
| if "duckduckgo.com" in parsed.netloc: | |
| qs = urllib.parse.parse_qs(parsed.query) | |
| uddg = qs.get("uddg") | |
| if uddg: | |
| return urllib.parse.unquote(uddg[0]) | |
| except Exception: | |
| pass | |
| return url | |
| def search_web(query, max_results=SEARCH_RESULTS): | |
| """Search the web and return a list of URLs.""" | |
| urls = [] | |
| try: | |
| with DDGS() as ddgs: | |
| for r in ddgs.text(query, max_results=max_results): | |
| url = r.get("href") or r.get("url") | |
| if not url: | |
| continue | |
| url = unwrap_ddg(url) | |
| urls.append(url) | |
| except Exception as e: | |
| print(f"Search error: {e}") | |
| return urls | |
| def fetch_text(url, timeout=TIMEOUT): | |
| """Fetch and clean text content from a URL.""" | |
| headers = {"User-Agent": "Mozilla/5.0 (research-agent)"} | |
| try: | |
| r = requests.get(url, timeout=timeout, headers=headers, allow_redirects=True) | |
| if r.status_code != 200: | |
| return "" | |
| ct = r.headers.get("content-type", "") | |
| if "html" not in ct.lower(): | |
| return "" | |
| soup = BeautifulSoup(r.text, "html.parser") | |
| # Remove unnecessary tags | |
| for tag in soup(["script", "style", "noscript", "header", "footer", | |
| "svg", "iframe", "nav", "aside"]): | |
| tag.extract() | |
| # Get paragraph text | |
| paragraphs = [p.get_text(" ", strip=True) for p in soup.find_all("p")] | |
| text = " ".join([p for p in paragraphs if p]) | |
| if text.strip(): | |
| return re.sub(r"\s+", " ", text).strip() | |
| # Fallback to meta description | |
| meta = soup.find("meta", attrs={"name": "description"}) or \ | |
| soup.find("meta", attrs={"property": "og:description"}) | |
| if meta and meta.get("content"): | |
| return meta["content"].strip() | |
| if soup.title and soup.title.string: | |
| return soup.title.string.strip() | |
| except Exception as e: | |
| print(f"Fetch error for {url}: {e}") | |
| return "" | |
| def chunk_passages(text, max_words=120): | |
| """Split long text into smaller passages.""" | |
| words = text.split() | |
| if not words: | |
| return [] | |
| chunks = [] | |
| i = 0 | |
| while i < len(words): | |
| chunk = words[i : i + max_words] | |
| chunks.append(" ".join(chunk)) | |
| i += max_words | |
| return chunks | |
| def split_sentences(text): | |
| """A simple sentence splitter.""" | |
| parts = re.split(r'(?<=[.!?])\s+', text) | |
| return [p.strip() for p in parts if p.strip()] | |
| class ShortResearchAgent: | |
| def __init__(self, embed_model=EMBEDDING_MODEL): | |
| print(f"Loading embedder: {embed_model}...") | |
| self.embedder = SentenceTransformer(embed_model) | |
| def run(self, query, progress=gr.Progress()): | |
| """Run the research agent pipeline.""" | |
| start = time.time() | |
| # Step 1: Search | |
| progress(0.1, desc="🔍 Searching the web...") | |
| urls = search_web(query) | |
| if not urls: | |
| elapsed = time.time() - start | |
| return { | |
| "query": query, | |
| "passages": [], | |
| "summary": "⚠️ No search results found. Please try a different query.", | |
| "time": elapsed, | |
| "num_urls": 0 | |
| } | |
| # Step 2: Fetch & Chunk | |
| progress(0.3, desc=f"📥 Fetching content from {len(urls)} URLs...") | |
| docs = [] | |
| for u in urls: | |
| txt = fetch_text(u) | |
| if not txt: | |
| continue | |
| chunks = chunk_passages(txt, max_words=120) | |
| for c in chunks[:PASSAGES_PER_PAGE]: | |
| docs.append({"url": u, "passage": c}) | |
| if not docs: | |
| elapsed = time.time() - start | |
| return { | |
| "query": query, | |
| "passages": [], | |
| "summary": "⚠️ No content could be extracted from the search results.", | |
| "time": elapsed, | |
| "num_urls": len(urls) | |
| } | |
| # Step 3: Embed | |
| progress(0.5, desc="🧠 Analyzing content with AI...") | |
| texts = [d["passage"] for d in docs] | |
| emb_texts = self.embedder.encode(texts, convert_to_numpy=True, show_progress_bar=False) | |
| q_emb = self.embedder.encode([query], convert_to_numpy=True)[0] | |
| # Step 4: Rank | |
| progress(0.7, desc="📊 Ranking relevant passages...") | |
| def cosine(a, b): | |
| return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-10) | |
| sims = [cosine(e, q_emb) for e in emb_texts] | |
| top_idx = np.argsort(sims)[::-1][:TOP_PASSAGES] | |
| top_passages = [ | |
| { | |
| "url": docs[i]["url"], | |
| "passage": docs[i]["passage"], | |
| "score": float(sims[i]) | |
| } | |
| for i in top_idx | |
| ] | |
| # Step 5: Summarize | |
| progress(0.9, desc="✍️ Generating summary...") | |
| if not top_passages: | |
| summary = "⚠️ No relevant passages found for summarization." | |
| else: | |
| sentences = [] | |
| for tp in top_passages: | |
| for s in split_sentences(tp["passage"]): | |
| sentences.append({"sent": s, "url": tp["url"]}) | |
| if not sentences: | |
| summary = "⚠️ No sentences found in relevant passages." | |
| else: | |
| sent_texts = [s["sent"] for s in sentences] | |
| sent_embs = self.embedder.encode(sent_texts, convert_to_numpy=True, | |
| show_progress_bar=False) | |
| sent_sims = [cosine(e, q_emb) for e in sent_embs] | |
| top_sent_idx = np.argsort(sent_sims)[::-1][:SUMMARY_SENTENCES] | |
| chosen = [sentences[idx] for idx in top_sent_idx] | |
| # De-duplicate and format | |
| seen = set() | |
| lines = [] | |
| for s in chosen: | |
| key = s["sent"].lower()[:80] | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| lines.append(f"{s['sent']} [(Source)]({s['url']})") | |
| summary = "\n\n".join(lines) | |
| elapsed = time.time() - start | |
| progress(1.0, desc="✅ Complete!") | |
| return { | |
| "query": query, | |
| "passages": top_passages, | |
| "summary": summary, | |
| "time": elapsed, | |
| "num_urls": len(urls) | |
| } | |
| # Initialize the agent globally | |
| print("Initializing Research Agent...") | |
| agent = ShortResearchAgent() | |
| def research_interface(query): | |
| """Gradio interface function.""" | |
| if not query or len(query.strip()) < 3: | |
| return "❌ Please enter a valid query (at least 3 characters).", "" | |
| try: | |
| result = agent.run(query.strip()) | |
| # Format summary | |
| summary_md = f"""# 📝 Research Summary | |
| **Query:** {result['query']} | |
| **Time taken:** {result['time']:.2f} seconds | |
| **URLs searched:** {result['num_urls']} | |
| --- | |
| ## Summary | |
| {result['summary']} | |
| """ | |
| # Format detailed passages | |
| passages_md = "# 🔍 Top Relevant Passages\n\n" | |
| if result['passages']: | |
| for i, p in enumerate(result['passages'], 1): | |
| passages_md += f"""### Passage {i} (Relevance: {p['score']:.2%}) | |
| **Source:** [{p['url']}]({p['url']}) | |
| {p['passage']} | |
| --- | |
| """ | |
| else: | |
| passages_md += "No passages found." | |
| return summary_md, passages_md | |
| except Exception as e: | |
| error_msg = f"❌ **Error:** {str(e)}\n\nPlease try again with a different query." | |
| return error_msg, "" | |
| # Create Gradio interface | |
| with gr.Blocks(theme=gr.themes.Soft(), title="AI Research Agent") as demo: | |
| gr.Markdown(""" | |
| # 🤖 AI Research Agent | |
| ### Intelligent Web Search & Summarization Tool | |
| This tool searches the web, analyzes multiple sources, and provides you with: | |
| - **AI-generated summary** of the most relevant information | |
| - **Top passages** ranked by relevance with sources | |
| - **Fast results** powered by semantic search | |
| Simply enter your question below and let the AI do the research for you! | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=4): | |
| query_input = gr.Textbox( | |
| label="🔍 Enter your research query", | |
| placeholder="e.g., What causes urban heat islands and how can cities reduce them?", | |
| lines=2 | |
| ) | |
| with gr.Column(scale=1): | |
| search_btn = gr.Button("🚀 Research", variant="primary", size="lg") | |
| gr.Markdown("### 💡 Example Queries") | |
| with gr.Row(): | |
| example_btns = [ | |
| gr.Button("🌡️ Urban heat islands", size="sm"), | |
| gr.Button("🤖 Latest AI developments", size="sm"), | |
| gr.Button("🌱 Sustainable energy solutions", size="sm"), | |
| gr.Button("🧬 CRISPR gene editing", size="sm") | |
| ] | |
| gr.Markdown("---") | |
| with gr.Row(): | |
| with gr.Column(): | |
| summary_output = gr.Markdown(label="Summary") | |
| with gr.Accordion("📚 Detailed Passages", open=False): | |
| passages_output = gr.Markdown(label="Top Passages") | |
| # Event handlers | |
| search_btn.click( | |
| fn=research_interface, | |
| inputs=[query_input], | |
| outputs=[summary_output, passages_output] | |
| ) | |
| query_input.submit( | |
| fn=research_interface, | |
| inputs=[query_input], | |
| outputs=[summary_output, passages_output] | |
| ) | |
| # Example button handlers | |
| example_queries = [ | |
| "What causes urban heat islands and how can cities reduce them?", | |
| "What are the latest developments in artificial intelligence?", | |
| "What are the most promising sustainable energy solutions?", | |
| "How does CRISPR gene editing work and what are its applications?" | |
| ] | |
| for btn, query in zip(example_btns, example_queries): | |
| btn.click( | |
| fn=lambda q=query: q, | |
| outputs=[query_input] | |
| ) | |
| gr.Markdown(""" | |
| --- | |
| ### 📌 Tips | |
| - Be specific with your queries for better results | |
| - The tool analyzes 6 web sources by default | |
| - Results typically take 10-30 seconds depending on query complexity | |
| **Built with:** DuckDuckGo Search, Sentence Transformers, Gradio | |
| """) | |
| # Launch the app | |
| if __name__ == "__main__": | |
| demo.launch() |