File size: 4,137 Bytes
fb15b03 525cd32 fb15b03 37c13e3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
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],
):
""" 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)
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, 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) -> None:
print("Writing email...")
result = await Runner.run(
email_agent,
report.markdown_report,
)
print("Email sent")
return report |