Upload folder using huggingface_hub

appagents/deep_research.py

@@ -0,0 +1,23 @@
+import gradio as gr
+from dotenv import load_dotenv
+from research_manager import ResearchManager
+
+load_dotenv(override=True)
+
+
+async def run(query: str):
+    async for chunk in ResearchManager().run(query):
+        yield chunk
+
+
+with gr.Blocks(theme=gr.themes.Default(primary_hue="sky")) as ui:
+    gr.Markdown("# Deep Research")
+    query_textbox = gr.Textbox(label="What topic would you like to research?")
+    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(inbrowser=True)
+

appagents/email_agent.py

@@ -0,0 +1,29 @@
+import os
+from typing import Dict
+
+import sendgrid
+from sendgrid.helpers.mail import Email, Mail, Content, To
+from agents import Agent, function_tool
+
+@function_tool
+def send_email(subject: str, html_body: str) -> Dict[str, str]:
+    """ Send an email with the given subject and HTML body """
+    sg = sendgrid.SendGridAPIClient(api_key=os.environ.get('SENDGRID_API_KEY'))
+    from_email = Email("ed@edwarddonner.com") # put your verified sender here
+    to_email = To("ed.donner@gmail.com") # put your recipient here
+    content = Content("text/html", html_body)
+    mail = Mail(from_email, to_email, subject, content).get()
+    response = sg.client.mail.send.post(request_body=mail)
+    print("Email response", response.status_code)
+    return {"status": "success"}
+
+INSTRUCTIONS = """You are able to send a nicely formatted HTML email based on a detailed report.
+You will be provided with a detailed report. You should use your tool to send one email, providing the 
+report converted into clean, well presented HTML with an appropriate subject line."""
+
+email_agent = Agent(
+    name="Email agent",
+    instructions=INSTRUCTIONS,
+    tools=[send_email],
+    model="gpt-4o-mini",
+)

appagents/planner_agent.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, Field
+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 = 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,
+)

appagents/research_agent.py

@@ -2,7 +2,7 @@ from tools.google_tools import GoogleTools
 from tools.news_tools import NewsTools
 from tools.yahoo_tools import FinanceTools
 from tools.time_tools import TimeTools
-from agents import Agent
+from agents import Agent, WebSearchTool
 
 class MarketResearchAgent:
     """
@@ -16,10 +16,11 @@ class MarketResearchAgent:
         """
         tools = [
             TimeTools.current_datetime,
-            GoogleTools.search,
-            FinanceTools.get_summary,
+            # GoogleTools.search,
+            WebSearchTool(search_context_size="low"),
+            # FinanceTools.get_summary,
             FinanceTools.get_market_sentiment,
-            FinanceTools.get_history,
+            # FinanceTools.get_history,
             # NewsTools.top_headlines,
             # NewsTools.search_news,
         ]
@@ -45,6 +46,6 @@ class MarketResearchAgent:
             name="AI Assistant",
             tools=tools,
             instructions=instructions,
-            model="gpt-4o"
+            model="gpt-4o-mini"
         )
         return agent

appagents/research_manager.py

@@ -0,0 +1,84 @@
+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
+
+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(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}"
+            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 written, sending email..."
+            await self.send_email(report)
+            yield "Email sent, research complete"
+            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)
+    
+    async def send_email(self, report: ReportData) -> None:
+        print("Writing email...")
+        result = await Runner.run(
+            email_agent,
+            report.markdown_report,
+        )
+        print("Email sent")
+        return report

appagents/search_agent.py

@@ -0,0 +1,17 @@
+from agents import Agent, WebSearchTool
+from agents.model_settings import 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 it's 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"),
+)
+

appagents/writer_agent.py

@@ -0,0 +1,27 @@
+from pydantic import BaseModel, Field
+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 = 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,
+)