Upload 7 files

app.py

@@ -0,0 +1,30 @@
+from dotenv import load_dotenv
+
+import gradio as gr
+
+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 user_interface:
+
+    gr.Markdown("Digital Research Assistant")
+
+    query = gr.Textbox(label="Hi there, what would you like to research on today?")
+
+    execute_button = gr.Button("Execute", variant="primary")
+
+    research_report = gr.Markdown(label="Research Report")
+
+    execute_button.click(fn= run, inputs=query, outputs=research_report)
+
+    query.submit(fn = run, inputs = query, outputs = research_report)
+
+
+user_interface.launch()

email_agent.py

@@ -0,0 +1,54 @@
+import os
+
+from typing import Dict
+
+import sendgrid
+
+from sendgrid import SendGridAPIClient
+
+from sendgrid.helpers.mail import To, Email, Mail, Content
+
+from agents import Agent, function_tool
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+@function_tool
+def send_email(subject: str, html_body: str) -> Dict[str, str]:
+
+  """Send out a given email with the given subject and html body. """
+
+  sendgrid_client = SendGridAPIClient(api_key = os.environ.get("SENDGRID_API_KEY"))
+
+  to_email = To("faiaza037@gmail.com")
+
+  from_email = Email("faiazrex8@gmail.com")
+
+  content = Content("text/html", html_body)
+
+  mail = Mail(to_emails=to_email, from_email=from_email, subject=subject, html_content=content)
+
+  response = sendgrid_client.send(mail)
+
+  return {"status" : "success"}
+
+email_agent_instructions = """
+
+You send nice HTML formatted emails based on a detailed report as you will be provided a detailed report.
+
+Use your available tool to send one email providing the report is converted to a clean, well structured HTML with appropriate subject line.
+
+"""
+
+email_agent = Agent(
+    
+                    name = "Email Agent",
+
+                    instructions = email_agent_instructions,
+
+                    tools = [send_email],
+
+                    model = "gpt-4o"
+                    
+                    )

requirements.txt

@@ -0,0 +1,12 @@
+# Core dependencies
+requests>=2.31.0
+python-dotenv>=1.0.0
+gradio>=4.19.2
+sendgrid>=6.11.0
+openai>=1.12.0
+openai-agents>=0.0.17
+
+# Development dependencies
+pytest>=7.4.0
+black>=23.7.0
+flake8>=6.1.0 

research_manager.py

@@ -0,0 +1,143 @@
+import asyncio
+
+from typing import Dict
+
+from agents import trace, gen_trace_id, Runner
+
+from web_searcher import web_searcher
+
+from research_planner import research_planner, WebSearchItem, WebSearchResultsList
+
+from research_writer import research_writer, ReportFormat
+
+from email_agent import email_agent
+
+class ResearchManager:
+
+    async def run(self, query: str):
+
+        """Runs the deep research process, yielding the status updates and final report."""
+
+        trace_id = gen_trace_id()
+
+        with trace("Research in progress", trace_id=trace_id):
+
+            print(f"View trace: https://platform.openai.com/traces/{trace_id}")
+
+            yield f"View trace: https://platform.openai.com/traces/{trace_id}"
+
+            print("Initializing research.....")
+
+            research_plan = await self.plan_search(query)
+
+            yield "Search planning completed, preparing for search"
+
+            search_results = await self.perform_search(research_plan)
+
+            yield "Searches complete, drafting report...."
+
+            research_report = await self.draft_research_report(query, search_results)
+
+            yield "Research report drafted. Sending email with the report attached."
+
+            await self.send_email(research_report)
+
+            yield "Email sent successfully! Research Task completed."
+
+            yield research_report.markdown_report
+
+
+    async def plan_search(self, query: str) -> WebSearchResultsList:
+
+        """Plans search to be perfomed for the query"""   
+
+        print("Planning searches")
+
+        results = await Runner.run(research_planner, f"Query:{query}")
+
+        print(f"Will perform {len(results.final_output.web_search_results)} searches")
+
+        return results.final_output_as(WebSearchResultsList)
+    
+
+    async def perform_search(self, research_plan: WebSearchResultsList)-> list[str]:
+
+        """Runs the searches to perform for the query"""
+
+        print("Searching....")
+
+        num_completed = 0
+
+        tasks = [asyncio.create_task(self.search(item)) for item in research_plan.web_search_results]
+
+        results = []
+
+        for task in asyncio.as_completed(tasks):
+
+            result = await task
+
+            if result is not None:
+
+              results.append(result)
+
+            num_completed += 1
+
+            print(f"Searches completed: {num_completed}/{len(tasks)} completed.")
+
+        print("Search Complete!")
+
+        return results
+        
+    
+    async def search(self, item: WebSearchItem) -> str | None:
+
+        """Performs a search for the query"""
+
+        input = f"Query : {item.query}, Reason for searching/querying: {item.query}"
+
+        try:
+
+            result = await Runner.run(web_searcher, input)
+             
+        except Exception:
+
+            return None
+
+    async def draft_research_report(self, query:str, search_results: list[str]) -> ReportFormat:
+
+        """Drafting a research report for the query"""
+
+        print("Preparing research report....")
+
+        input = f"Original query: {query}, Summarized searches: {search_results}"
+
+        result = await Runner.run(research_writer, input)
+
+        print("Draft of research report completed!")
+
+        return result.final_output_as(ReportFormat)
+    
+
+    async def send_email(self, research_report: ReportFormat) -> None:
+
+        print("Sending email with research report attached....")
+
+        result = await Runner.run(email_agent, research_report.markdown_report)
+
+        print("Email sent sucessfully with research report attached!")
+
+        return research_report
+ 
+
+
+    
+
+
+
+
+
+
+
+
+
+

research_planner.py

@@ -0,0 +1,32 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+how_many_searches = 3
+
+research_planner_agent_instructions = f"You are a fellow and a very helpful research assistant. Given a query, come up with a set of web searches to best answer the query.Output {how_many_searches} terms to query for"
+
+class WebSearchItem(BaseModel):
+
+  reason: str
+  "Your reasoning and thought process of why this search is important to the query."
+
+  query: str
+  "The search term for to use for web searches"
+
+class WebSearchResultsList(BaseModel):
+
+  web_search_results : list[WebSearchItem]
+
+  """List of relevant web links of the query search result."""
+
+research_planner = Agent(
+
+                               name = "Research Planner",
+
+                               instructions = research_planner_agent_instructions,
+
+                               output_type = WebSearchResultsList,
+
+                               model = "gpt-4o"
+)

research_writer.py

@@ -0,0 +1,42 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+research_writer_instructions =  """
+
+You are a well experienced research writer tasked to preparing a cohesive report for a research query.
+
+You will be provided with the original query, and a few initial research draft done by a research assistant.
+
+You must first come up with the outline of the report that talks about the structure and flow of the report.
+
+Finally, generate the report and give that as your final output.
+
+The final output should be in a markdown format, and it should be lengthy and detailed, yet easy to understand.
+
+Deliver at least 5 to 7 pages of content of minimum 500 to 700 words.
+
+"""
+
+class ReportFormat(BaseModel):
+
+  summary: str
+  "A summary of the report generated"
+
+  markdown_report: str
+  "The final report"
+
+  follow_up_questions: list[str]
+  "Further relevant and associated research areas to discuss and explore further"
+
+
+research_writer = Agent(
+    
+                        name = "Research Writer",
+
+                        instructions = research_writer_instructions,
+
+                        model = "gpt-4o",
+
+                        output_type = ReportFormat
+)

web_searcher.py

@@ -0,0 +1,34 @@
+from agents import Agent
+
+from agents import WebSearchTool
+
+from agents import ModelSettings
+
+
+web_searcher_instructions = """ 
+
+You are a very sucessful and award-winning research assistant.
+
+Given a search term, you will always search the web for that term and generate a robust summary of the results.
+
+The summary must be strictly only 2 to 3 paragraphs and not more than 400 words.
+
+Capture only the essential and feasible points, write it in a very easy to understand way yet in extremely impacting tone.
+
+Do not include any unneccasary information other than the summary itself
+
+"""
+
+web_searcher = Agent(
+    
+                       name = "Research Assistant",
+
+                       instructions = web_searcher_instructions,
+
+                       tools = [WebSearchTool(search_context_size = "low")],
+
+                       model_settings = ModelSettings(tool_choice = "required"),
+
+                       model = "gpt-4o"
+)
+