Create app.py

app.py

@@ -0,0 +1,358 @@
+# -*- 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()