Upload folder using huggingface_hub

README.md

@@ -1,12 +1,6 @@
 ---
-title: Deep Research Assistant Agent
-emoji: 🏃
-colorFrom: green
-colorTo: pink
+title: Deep_Research_Assistant_Agent
+app_file: deep_research.py
 sdk: gradio
-sdk_version: 5.29.0
-app_file: app.py
-pinned: false
+sdk_version: 4.44.1
 ---
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

clarifier_agent.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel
+from agents import Agent
+
+class ClarifyingQuestions(BaseModel):
+    questions: list[str]
+    """Three clarifying questions to better understand the user's query."""
+
+clarifier_agent = Agent(
+    name="ClarifierAgent",
+    instructions=(
+        "You are a research assistant. Your task is to ask 3 clarifying questions that help refine and understand "
+        "a research query better. After the user answers them, hand off control to the Research Coordinator to perform the full research."
+    ),
+    model="gpt-4o-mini",
+    output_type=ClarifyingQuestions,
+)

deep_research.py

@@ -0,0 +1,138 @@
+# deep_research.py
+
+import gradio as gr
+from dotenv import load_dotenv
+from clarifier_agent import clarifier_agent
+from research_manager import ResearchManagerAgent
+from agents import Runner
+from collections import defaultdict
+from datetime import datetime
+import time
+import logging
+
+load_dotenv(override=True)
+
+# --- Rate Limiter ---
+class RateLimiter:
+    # Rate limit to 2 requests per minute, 10 requests per day
+    def __init__(self, max_requests=2, time_window=60, daily_quota=10):
+        self.max_requests = max_requests
+        self.time_window = time_window  # seconds
+        self.request_history = defaultdict(list)
+        self.daily_quota = daily_quota
+        self.daily_counts = defaultdict(lambda: {'date': self._today(), 'count': 0})
+
+    def _today(self):
+        return datetime.utcnow().strftime('%Y-%m-%d')
+
+    def is_rate_limited(self, user_id):
+        now = time.time()
+        self.request_history[user_id] = [
+            t for t in self.request_history[user_id] if now - t < self.time_window
+        ]
+        if len(self.request_history[user_id]) >= self.max_requests:
+            return True
+        self.request_history[user_id].append(now)
+        return False
+
+    def is_quota_exceeded(self, user_id):
+        today = self._today()
+        user_quota = self.daily_counts[user_id]
+        if user_quota['date'] != today:
+            user_quota['date'] = today
+            user_quota['count'] = 0
+        if user_quota['count'] >= self.daily_quota:
+            return True
+        user_quota['count'] += 1
+        self.daily_counts[user_id] = user_quota
+        return False
+
+rate_limiter = RateLimiter()
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+async def get_user_id(request: gr.Request = None):
+    user_id = "default_user"
+    if request is not None:
+        try:
+            forwarded = request.headers.get("X-Forwarded-For")
+            if forwarded:
+                user_id = forwarded.split(",")[0].strip()
+            else:
+                user_id = getattr(request.client, 'host', 'default_user')
+        except Exception:
+            pass
+    logger.debug(f"[RateLimiter] user_id={user_id}")
+    return user_id
+
+# Step 1 — Generate clarifying questions
+async def get_clarifying_questions(query, request: gr.Request = None):
+    user_id = await get_user_id(request)
+    if rate_limiter.is_rate_limited(user_id):
+        return ["Rate limit exceeded. Please wait a minute."], "", "", ""
+    if rate_limiter.is_quota_exceeded(user_id):
+        return ["Daily quota exceeded. Try again tomorrow."], "", "", ""
+
+    result = await Runner.run(clarifier_agent, input=query)
+    return result.final_output.questions
+
+# Step 2 — Run full research pipeline via coordinator agent (handoff style)
+async def run_with_handoff(query, q1, q2, q3, a1, a2, a3, send_email_flag, recipient_email, request: gr.Request = None):
+    user_id = await get_user_id(request)
+    if rate_limiter.is_rate_limited(user_id):
+        yield "Rate limit exceeded. Please wait a minute."
+        return
+    if rate_limiter.is_quota_exceeded(user_id):
+        yield "You have reached your daily quota. Try again tomorrow."
+        return
+
+    questions = [q1, q2, q3]
+    answers = [a1, a2, a3]
+    async for chunk in ResearchManagerAgent().run(
+        query,
+        questions,
+        answers,
+        send_email_flag=send_email_flag,
+        recipient_email=recipient_email,
+    ):
+        yield chunk
+
+with gr.Blocks(theme=gr.themes.Default(primary_hue="sky")) as ui:
+    gr.Markdown("# 🔍 Deep Research Agent (Clarify ➡️ Research ➡️ Email)")
+
+    query = gr.Textbox(label="🔎 What would you like to research?")
+
+    get_questions_btn = gr.Button("Generate Clarifying Questions", variant="primary")
+
+    clar_q1 = gr.Textbox(label="Clarifying Question 1", interactive=False)
+    clar_q2 = gr.Textbox(label="Clarifying Question 2", interactive=False)
+    clar_q3 = gr.Textbox(label="Clarifying Question 3", interactive=False)
+
+    answer_1 = gr.Textbox(label="Your Answer to Q1")
+    answer_2 = gr.Textbox(label="Your Answer to Q2")
+    answer_3 = gr.Textbox(label="Your Answer to Q3")
+
+    send_email_checkbox = gr.Checkbox(label="📧 Send Report via Email?")
+    email_box = gr.Textbox(label="Recipient Email", visible=False)
+
+    # Show/hide email textbox based on checkbox
+    send_email_checkbox.change(fn=lambda checked: gr.update(visible=checked), inputs=send_email_checkbox, outputs=email_box)
+
+    submit_answers_btn = gr.Button("✅ Submit & Run Full Research")
+    report = gr.Markdown(label="📄 Research Report")
+
+    # Step 1
+    get_questions_btn.click(
+        fn=get_clarifying_questions,
+        inputs=query,
+        outputs=[clar_q1, clar_q2, clar_q3]
+    ).then(lambda *_: "", outputs=report)
+
+    # Step 2
+    submit_answers_btn.click(
+        fn=run_with_handoff,
+        inputs=[query, clar_q1, clar_q2, clar_q3, answer_1, answer_2, answer_3, send_email_checkbox, email_box],
+        outputs=report
+    )
+
+ui.launch(inbrowser=True)

email_agent.py

@@ -0,0 +1,43 @@
+import os
+import base64
+import requests
+from agents import Agent, function_tool
+
+@function_tool
+def send_email(subject: str, html_body: str, to: str):
+    """Send out an email with the given subject and HTML body to a specified recipient using Mailgun"""
+    MAILGUN_API_KEY = os.environ.get('MAILGUN_API_KEY')
+    MAILGUN_DOMAIN = os.environ.get('MAILGUN_DOMAIN')
+
+    if not all([MAILGUN_API_KEY, MAILGUN_DOMAIN, to]):
+        return {"status": "failure", "response": "Missing configuration or recipient"}
+
+    auth = base64.b64encode(f'api:{MAILGUN_API_KEY}'.encode()).decode()
+    response = requests.post(
+        f'https://api.mailgun.net/v3/{MAILGUN_DOMAIN}/messages',
+        headers={
+            'Authorization': f'Basic {auth}'
+        },
+        data={
+            'from': f'Research Agent <mailgun@{MAILGUN_DOMAIN}>',
+            'to': to,
+            'subject': subject,
+            'html': html_body
+        }
+    )
+
+    return {
+        "status": "success" if response.status_code == 200 else "failure",
+        "response": response.text
+    }
+
+INSTRUCTIONS = """You are able to send a nicely formatted HTML email based on a detailed report.
+You will be provided with a detailed report and a recipient email. Use your tool to send one email, 
+providing the report as HTML with an appropriate subject line."""
+
+email_agent = Agent(
+    name="Email agent",
+    instructions=INSTRUCTIONS,
+    tools=[send_email],
+    model="gpt-4o-mini",
+)

planner_agent.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+from agents import Agent
+
+HOW_MANY_SEARCHES = 5
+
+INSTRUCTIONS = f"You are a helpful research assistant. Given a query, come up with a set of web searches \
+to perform to best answer the query. Output {HOW_MANY_SEARCHES} terms to query for."
+
+
+class WebSearchItem(BaseModel):
+    reason: str
+    "Your reasoning for why this search is important to the query."
+
+    query: str
+    "The search term to use for the web search."
+
+
+class WebSearchPlan(BaseModel):
+    searches: list[WebSearchItem]
+    """A list of web searches to perform to best answer the query."""
+
+
+planner_agent = Agent(
+    name="PlannerAgent",
+    instructions=INSTRUCTIONS,
+    model="gpt-4o-mini",
+    output_type=WebSearchPlan,
+)

requirements.txt

@@ -0,0 +1,6 @@
+requests
+python-dotenv
+gradio
+pypdf
+openai
+openai-agents

research_manager.py

@@ -0,0 +1,114 @@
+from agents import Runner, trace, gen_trace_id
+from search_agent import search_agent
+from planner_agent import planner_agent, WebSearchItem, WebSearchPlan
+from writer_agent import writer_agent, ReportData
+from email_agent import email_agent
+import asyncio
+from typing import Optional
+
+class ResearchManagerAgent:
+
+    async def run(
+        self,
+        query: str,
+        clarifying_questions: list[str],
+        clarifying_answers: list[str],
+        send_email_flag: bool = False,
+        recipient_email: Optional[str] = None,
+    ):
+        """ Run the deep research process using user-provided clarification answers. """
+        trace_id = gen_trace_id()
+        with trace("Research trace", trace_id=trace_id):
+            print(f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}")
+            yield f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}"
+            yield "Planning search based on clarifications..."
+
+            print(f"Clarifying questions: {clarifying_questions}")
+            print(f"Clarifying answers: {clarifying_answers}")
+
+            # Plan searches using clarifications and user answers
+            search_plan = await self.plan_searches(query, clarifying_questions, clarifying_answers)
+
+            yield "Searches planned, starting to search..."
+            search_results = await self.perform_searches(search_plan)
+
+            yield "Searches complete, writing report..."
+            report = await self.write_report(query, search_results)
+
+            if send_email_flag and recipient_email:
+                yield f"Sending report to {recipient_email}..."
+                await self.send_email(report, recipient_email)
+                yield "Email sent"
+            else:
+                yield "Skipping email step"
+               
+            yield "Email sent"
+            yield report.markdown_report
+
+    async def plan_searches(self, query: str, questions: list[str], answers: list[str]) -> WebSearchPlan:
+        """ Plan the searches to perform based on clarifications """
+        print("Planning searches...")
+
+        # Combine clarifying Q&A into structured prompt
+        clarifying_context = "\n".join(
+            f"Q: {q}\nA: {a}" for q, a in zip(questions, answers)
+        )
+        final_prompt = f"Query: {query}\nClarifications:\n{clarifying_context}"
+
+        result = await Runner.run(
+            planner_agent,
+            input=final_prompt,
+        )
+        print(f"Will perform {len(result.final_output.searches)} searches")
+        return result.final_output_as(WebSearchPlan)
+
+    async def perform_searches(self, search_plan: WebSearchPlan) -> list[str]:
+        """ Perform the searches for the planned queries """
+        print("Searching...")
+        num_completed = 0
+        tasks = [asyncio.create_task(self.search(item)) for item in search_plan.searches]
+        results = []
+        for task in asyncio.as_completed(tasks):
+            result = await task
+            if result is not None:
+                results.append(result)
+            num_completed += 1
+            print(f"Searching... {num_completed}/{len(tasks)} completed")
+        print("Finished searching")
+        return results
+
+    async def search(self, item: WebSearchItem) -> Optional[str]:
+        """ Perform a single web search """
+        input_text = f"Search term: {item.query}\nReason for searching: {item.reason}"
+        try:
+            result = await Runner.run(
+                search_agent,
+                input_text,
+            )
+            return str(result.final_output)
+        except Exception as e:
+            print(f"Search failed: {e}")
+            return None
+
+    async def write_report(self, query: str, search_results: list[str]) -> ReportData:
+        """ Write a markdown report from search results """
+        print("Thinking about report...")
+        input_text = f"Original query: {query}\nSummarized search results: {search_results}"
+        result = await Runner.run(
+            writer_agent,
+            input_text,
+        )
+        print("Finished writing report")
+        return result.final_output_as(ReportData)
+
+    async def send_email(self, report: ReportData, recipient_email: str) -> None:
+        """ Send the report via email """
+
+        email_prompt = f"""Send the following report as an email.
+        To: {recipient_email}
+        Body (HTML):
+        {report.markdown_report}
+        """
+        print(f"Sending email to: {recipient_email}")
+        await Runner.run(email_agent, input=email_prompt)
+        print("✅ Email sent")

search_agent.py

@@ -0,0 +1,17 @@
+from agents import Agent, WebSearchTool, ModelSettings
+
+INSTRUCTIONS = (
+    "You are a research assistant. Given a search term, you search the web for that term and "
+    "produce a concise summary of the results. The summary must 2-3 paragraphs and less than 300 "
+    "words. Capture the main points. Write succintly, no need to have complete sentences or good "
+    "grammar. This will be consumed by someone synthesizing a report, so its vital you capture the "
+    "essence and ignore any fluff. Do not include any additional commentary other than the summary itself."
+)
+
+search_agent = Agent(
+    name="Search agent",
+    instructions=INSTRUCTIONS,
+    tools=[WebSearchTool(search_context_size="low")],
+    model="gpt-4o-mini",
+    model_settings=ModelSettings(tool_choice="required"),
+)

writer_agent.py

@@ -0,0 +1,30 @@
+from pydantic import BaseModel
+from agents import Agent
+
+INSTRUCTIONS = (
+    "You are a senior researcher tasked with writing a cohesive report for a research query. "
+    "You will be provided with the original query, and some initial research done by a research assistant.\n"
+    "You should first come up with an outline for the report that describes the structure and "
+    "flow of the report. Then, generate the report and return that as your final output.\n"
+    "The final output should be in markdown format, and it should be lengthy and detailed. Aim "
+    "for 5-10 pages of content, at least 1000 words."
+)
+
+
+class ReportData(BaseModel):
+    short_summary: str
+    """A short 2-3 sentence summary of the findings."""
+
+    markdown_report: str
+    """The final report"""
+
+    follow_up_questions: list[str]
+    """Suggested topics to research further"""
+
+
+writer_agent = Agent(
+    name="WriterAgent",
+    instructions=INSTRUCTIONS,
+    model="gpt-4o-mini",
+    output_type=ReportData,
+)