Building a deep research agent with Temporal

export OPENAI_API_KEY="your-openai-api-key"
export BRAINTRUST_API_KEY="your-braintrust-api-key"
export BRAINTRUST_PROJECT_NAME="deep-research"  # optional

git clone https://github.com/braintrustdata/braintrust-cookbook.git
cd braintrust-cookbook/examples/TemporalDeepResearch/temporal-deep-research
uv sync

temporal server start-dev

uv run python -m worker

uv run python -m start_workflow "What's the best restaurant in San Francisco?"

class ResearchPlan(BaseModel):
    research_question: str
    key_aspects: List[ResearchAspect]
    expected_sources: List[str]
    search_strategy: str
    success_criteria: List[str]

class SearchQuery(BaseModel):
    query: str
    rationale: str
    expected_info_type: str
    priority: int

class SearchResult(BaseModel):
    query: str
    sources: List[str]
    key_findings: str
    relevance_score: float
    citations: List[str]

class ResearchReport(BaseModel):
    executive_summary: str
    detailed_analysis: str
    key_findings: List[str]
    confidence_assessment: str
    citations: List[str]
    follow_up_questions: List[str]

RESEARCH_PLANNING_INSTRUCTIONS = """
You are a research planning specialist who creates focused research strategies.

CORE RESPONSIBILITIES:
1. Decompose the user's question into 3-7 key research aspects
2. Identify required sources and evidence types
3. Design a practical search strategy
4. Set clear success criteria
"""

async def plan_research(query: str) -> ResearchPlan:
    result = await workflow.execute_activity(
        invoke_model,
        InvokeModelRequest(
            model=COMPLEX_REASONING_MODEL,
            instructions=RESEARCH_PLANNING_INSTRUCTIONS,
            input=f"Research query: {query}",
            response_format=ResearchPlan,
        ),
        start_to_close_timeout=timedelta(seconds=300),
    )
    return result.response

QUERY_GENERATION_INSTRUCTIONS = """
You are a search query specialist who crafts effective web searches.

CORE RESPONSIBILITIES:
1. Generate 3-5 diverse search queries based on the research plan
2. Balance specificity with discoverability
3. Target different information types (factual, analytical, recent, historical)
"""

async def generate_queries(research_plan: ResearchPlan) -> QueryPlan:
    result = await workflow.execute_activity(
        invoke_model,
        InvokeModelRequest(
            model=EFFICIENT_PROCESSING_MODEL,
            instructions=QUERY_GENERATION_INSTRUCTIONS,
            input=plan_context,
            response_format=QueryPlan,
        ),
        start_to_close_timeout=timedelta(seconds=300),
    )
    return result.response

WEB_SEARCH_INSTRUCTIONS = """
You are a web research specialist who finds and evaluates information.

CORE RESPONSIBILITIES:
1. Execute web searches using the web search tool
2. Prioritize authoritative sources
3. Extract key information relevant to the research question
4. Provide proper citations and assess reliability
"""

async def search_web(query: SearchQuery) -> SearchResult:
    result = await workflow.execute_activity(
        invoke_model,
        InvokeModelRequest(
            model=EFFICIENT_PROCESSING_MODEL,
            instructions=WEB_SEARCH_INSTRUCTIONS,
            input=search_input,
            response_format=SearchResult,
            tools=[{"type": "web_search"}],
        ),
        start_to_close_timeout=timedelta(seconds=300),
    )
    return result.response

REPORT_SYNTHESIS_INSTRUCTIONS = """
You are a research synthesis expert who creates comprehensive reports.

CORE RESPONSIBILITIES:
1. Synthesize all research into a coherent narrative
2. Structure information logically with evidence support
3. Provide comprehensive citations
4. Assess confidence levels and acknowledge limitations
"""

async def generate_synthesis(
    original_query: str, research_plan: ResearchPlan, search_results: List[SearchResult]
) -> ResearchReport:
    result = await workflow.execute_activity(
        invoke_model,
        InvokeModelRequest(
            model=COMPLEX_REASONING_MODEL,
            instructions=REPORT_SYNTHESIS_INSTRUCTIONS,
            input=synthesis_input,
            response_format=ResearchReport,
        ),
        start_to_close_timeout=timedelta(seconds=300),
    )
    return result.response

@workflow.defn
class DeepResearchWorkflow:
    @workflow.run
    async def run(self, query: str) -> str:
        # Step 1: Research Planning
        research_plan = await plan_research(query)

        # Step 2: Query Generation
        query_plan = await generate_queries(research_plan)

        # Step 3: Web Search (parallel execution with resilience)
        search_results = await self._execute_searches(query_plan.queries)

        if not search_results:
            raise ApplicationError(
                "All web searches failed - cannot generate report",
                non_retryable=True,
            )

        # Step 4: Report Synthesis
        final_report = await generate_synthesis(query, research_plan, search_results)

        return self._format_final_report(query, final_report)

async def _execute_searches(self, search_queries) -> List[SearchResult]:
    async def execute_single_search(search_query):
        try:
            return await search_web(search_query)
        except Exception as e:
            workflow.logger.exception(f"Search failed: {e}")
            return None

    search_tasks = [execute_single_search(q) for q in search_queries]
    results = await asyncio.gather(*search_tasks)
    return [r for r in results if r is not None]

from braintrust import init_logger
from braintrust.contrib.temporal import BraintrustPlugin

init_logger(project=os.environ.get("BRAINTRUST_PROJECT_NAME", "deep-research"))

worker = Worker(
    client,
    task_queue="deep-research-task-queue",
    workflows=[DeepResearchWorkflow],
    activities=[invoke_model.invoke_model],
    plugins=[BraintrustPlugin()],
)

client = wrap_openai(AsyncOpenAI())
resp = await client.responses.parse(**kwargs)