add basic version

.gitignore

@@ -0,0 +1,13 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+# Environment variables
+.env

.python-version

@@ -0,0 +1 @@
+3.12

pyproject.toml

@@ -0,0 +1,14 @@
+[project]
+name = "deep-research"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "black>=25.1.0",
+    "dotenv>=0.9.9",
+    "gradio>=5.33.1",
+    "isort>=6.0.1",
+    "openai>=1.86.0",
+    "openai-agents>=0.0.17",
+]

src/main.py

@@ -0,0 +1,26 @@
+import gradio as gr
+from dotenv import load_dotenv
+
+from research_manager import ResearchManager
+
+load_dotenv()
+
+
+async def run(query: str):
+    async for chunk in ResearchManager().run(query):
+        yield chunk
+
+
+with gr.Blocks(theme=gr.themes.Default(primary_hue="yellow")) as ui:
+    gr.Markdown("# Deep Research")
+    query_textbox = gr.Textbox(
+        label="What topic would you like to research?",
+        value="e.g. How to create a Deep Research Agent?",
+    )
+    run_button = gr.Button("Run", variant="primary")
+    report = gr.Markdown(label="Report")
+
+    run_button.click(fn=run, inputs=query_textbox, outputs=report)
+    query_textbox.submit(fn=run, inputs=query_textbox, outputs=report)
+
+ui.launch()

src/planner.py

@@ -0,0 +1,28 @@
+from agents import Agent
+from pydantic import BaseModel, Field
+
+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 = Field(
+        description="Your reasoning for why this search is important to the query."
+    )
+    query: str = Field(description="The search term to use for the web search.")
+
+
+class WebSearchPlan(BaseModel):
+    searches: list[WebSearchItem] = Field(
+        description="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,
+)

src/report_generator.py

@@ -0,0 +1,31 @@
+from agents import Agent
+from pydantic import BaseModel, Field
+
+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 = Field(
+        description="A short 2-3 sentence summary of the findings."
+    )
+
+    markdown_report: str = Field(description="The final report")
+
+    follow_up_questions: list[str] = Field(
+        description="Suggested topics to research further"
+    )
+
+
+writer_agent = Agent(
+    name="WriterAgent",
+    instructions=INSTRUCTIONS,
+    model="gpt-4o-mini",
+    output_type=ReportData,
+)

src/research_manager.py

@@ -0,0 +1,73 @@
+import asyncio
+
+from agents import Runner, gen_trace_id, trace
+
+from planner import WebSearchItem, WebSearchPlan, planner_agent
+from report_generator import ReportData, writer_agent
+from web_search import search_agent
+
+
+class ResearchManager:
+
+    async def run(self, query: str):
+        """Run the deep research process, yielding the status updates and the final report"""
+        trace_id = gen_trace_id()
+        with trace("Research trace", trace_id=trace_id):
+            print("Starting research...")
+            search_plan = await self.plan_searches(query)
+            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)
+            yield report.markdown_report
+
+    async def plan_searches(self, query: str) -> WebSearchPlan:
+        """Plan the searches to perform for the query"""
+        print("Planning searches...")
+        result = await Runner.run(
+            planner_agent,
+            f"Query: {query}",
+        )
+        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 to perform for the query"""
+        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) -> str | None:
+        """Perform a search for the query"""
+        input = f"Search term: {item.query}\nReason for searching: {item.reason}"
+        try:
+            result = await Runner.run(
+                search_agent,
+                input,
+            )
+            return str(result.final_output)
+        except Exception:
+            return None
+
+    async def write_report(self, query: str, search_results: list[str]) -> ReportData:
+        """Write the report for the query"""
+        print("Thinking about report...")
+        input = f"Original query: {query}\nSummarized search results: {search_results}"
+        result = await Runner.run(
+            writer_agent,
+            input,
+        )
+
+        print("Finished writing report")
+        return result.final_output_as(ReportData)

src/web_search.py

@@ -0,0 +1,17 @@
+from agents import Agent, ModelSettings, WebSearchTool
+
+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"),
+)