temporalio · jssmith · Jul 26, 2025 · Jul 27, 2025 · Jul 27, 2025 · Jul 27, 2025
@@ -0,0 +1,61 @@
+# Financial Research Agent
+
+Multi-agent financial research system with specialized roles, extended with Temporal's durable execution.
+
+*Adapted from [OpenAI Agents SDK financial research agent](https://github.com/openai/openai-agents-python/tree/main/examples/financial_research_agent)*
+
+## Architecture
+
+This example shows how you might compose a richer financial research agent using the Agents SDK. The pattern is similar to the `research_bot` example, but with more specialized sub-agents and a verification step.
+
+The flow is:
+
+1. **Planning**: A planner agent turns the end user's request into a list of search terms relevant to financial analysis – recent news, earnings calls, corporate filings, industry commentary, etc.
+2. **Search**: A search agent uses the built-in `WebSearchTool` to retrieve terse summaries for each search term. (You could also add `FileSearchTool` if you have indexed PDFs or 10-Ks.)
+3. **Sub-analysts**: Additional agents (e.g. a fundamentals analyst and a risk analyst) are exposed as tools so the writer can call them inline and incorporate their outputs.
+4. **Writing**: A senior writer agent brings together the search snippets and any sub-analyst summaries into a long-form markdown report plus a short executive summary.
+5. **Verification**: A final verifier agent audits the report for obvious inconsistencies or missing sourcing.
+
+## Running the Example
+
+First, start the worker:
+```bash
+uv run openai_agents/financial_research_agent/run_worker.py
+```
+
+Then run the financial research workflow:
+```bash
+uv run openai_agents/financial_research_agent/run_financial_research_workflow.py
+```
+
+Enter a query like:
+```
+Write up an analysis of Apple Inc.'s most recent quarter.
+```
+
+You can also just hit enter to run this query, which is provided as the default.
+
+## Components
+
+### Agents
+
+- **Planner Agent**: Creates a search plan with 5-15 relevant search terms
+- **Search Agent**: Uses web search to gather financial information
+- **Financials Agent**: Analyzes company fundamentals (revenue, profit, margins)
+- **Risk Agent**: Identifies potential red flags and risk factors
+- **Writer Agent**: Synthesizes information into a comprehensive report
+- **Verifier Agent**: Audits the final report for consistency and accuracy
+
+### Writer Agent Tools
+
+The writer agent has access to tools that invoke the specialist analysts:
+- `fundamentals_analysis`: Get financial performance analysis
+- `risk_analysis`: Get risk factor assessment
+
+## Temporal Integration
+
+The example demonstrates several Temporal patterns:
+- Durable execution of multi-step research workflows
+- Parallel execution of web searches using `asyncio.create_task`
+- Use of `workflow.as_completed` for handling concurrent tasks
+- Proper import handling with `workflow.unsafe.imports_passed_through()`
@@ -0,0 +1,23 @@
+from agents import Agent
+from pydantic import BaseModel
+
+# A sub-agent focused on analyzing a company's fundamentals.
+FINANCIALS_PROMPT = (
+    "You are a financial analyst focused on company fundamentals such as revenue, "
+    "profit, margins and growth trajectory. Given a collection of web (and optional file) "
+    "search results about a company, write a concise analysis of its recent financial "
+    "performance. Pull out key metrics or quotes. Keep it under 2 paragraphs."
+)
+
+
+class AnalysisSummary(BaseModel):
+    summary: str
+    """Short text summary for this aspect of the analysis."""
+
+
+def new_financials_agent() -> Agent:
+    return Agent(
+        name="FundamentalsAnalystAgent",
+        instructions=FINANCIALS_PROMPT,
+        output_type=AnalysisSummary,
+    )
@@ -0,0 +1,35 @@
+from agents import Agent
+from pydantic import BaseModel
+
+# Generate a plan of searches to ground the financial analysis.
+# For a given financial question or company, we want to search for
+# recent news, official filings, analyst commentary, and other
+# relevant background.
+PROMPT = (
+    "You are a financial research planner. Given a request for financial analysis, "
+    "produce a set of web searches to gather the context needed. Aim for recent "
+    "headlines, earnings calls or 10-K snippets, analyst commentary, and industry background. "
+    "Output between 5 and 15 search terms to query for."
+)
+
+
+class FinancialSearchItem(BaseModel):
+    reason: str
+    """Your reasoning for why this search is relevant."""
+
+    query: str
+    """The search term to feed into a web (or file) search."""
+
+
+class FinancialSearchPlan(BaseModel):
+    searches: list[FinancialSearchItem]
+    """A list of searches to perform."""
+
+
+def new_planner_agent() -> Agent:
+    return Agent(
+        name="FinancialPlannerAgent",
+        instructions=PROMPT,
+        model="o3-mini",
+        output_type=FinancialSearchPlan,
+    )
@@ -0,0 +1,22 @@
+from agents import Agent
+from pydantic import BaseModel
+
+# A sub-agent specializing in identifying risk factors or concerns.
+RISK_PROMPT = (
+    "You are a risk analyst looking for potential red flags in a company's outlook. "
+    "Given background research, produce a short analysis of risks such as competitive threats, "
+    "regulatory issues, supply chain problems, or slowing growth. Keep it under 2 paragraphs."
+)
+
+
+class AnalysisSummary(BaseModel):
+    summary: str
+    """Short text summary for this aspect of the analysis."""
+
+
+def new_risk_agent() -> Agent:
+    return Agent(
+        name="RiskAnalystAgent",
+        instructions=RISK_PROMPT,
+        output_type=AnalysisSummary,
+    )
@@ -0,0 +1,20 @@
+from agents import Agent, WebSearchTool
+from agents.model_settings import ModelSettings
+
+# Given a search term, use web search to pull back a brief summary.
+# Summaries should be concise but capture the main financial points.
+INSTRUCTIONS = (
+    "You are a research assistant specializing in financial topics. "
+    "Given a search term, use web search to retrieve up-to-date context and "
+    "produce a short summary of at most 300 words. Focus on key numbers, events, "
+    "or quotes that will be useful to a financial analyst."
+)
+
+
+def new_search_agent() -> Agent:
+    return Agent(
+        name="FinancialSearchAgent",
+        instructions=INSTRUCTIONS,
+        tools=[WebSearchTool()],
+        model_settings=ModelSettings(tool_choice="required"),
+    )
@@ -0,0 +1,27 @@
+from agents import Agent
+from pydantic import BaseModel
+
+# Agent to sanity-check a synthesized report for consistency and recall.
+# This can be used to flag potential gaps or obvious mistakes.
+VERIFIER_PROMPT = (
+    "You are a meticulous auditor. You have been handed a financial analysis report. "
+    "Your job is to verify the report is internally consistent, clearly sourced, and makes "
+    "no unsupported claims. Point out any issues or uncertainties."
+)
+
+
+class VerificationResult(BaseModel):
+    verified: bool
+    """Whether the report seems coherent and plausible."""
+
+    issues: str
+    """If not verified, describe the main issues or concerns."""
+
+
+def new_verifier_agent() -> Agent:
+    return Agent(
+        name="VerificationAgent",
+        instructions=VERIFIER_PROMPT,
+        model="gpt-4o",
+        output_type=VerificationResult,
+    )
@@ -0,0 +1,34 @@
+from agents import Agent
+from pydantic import BaseModel
+
+# Writer agent brings together the raw search results and optionally calls out
+# to sub-analyst tools for specialized commentary, then returns a cohesive markdown report.
+WRITER_PROMPT = (
+    "You are a senior financial analyst. You will be provided with the original query and "
+    "a set of raw search summaries. Your task is to synthesize these into a long-form markdown "
+    "report (at least several paragraphs) including a short executive summary and follow-up "
+    "questions. If needed, you can call the available analysis tools (e.g. fundamentals_analysis, "
+    "risk_analysis) to get short specialist write-ups to incorporate."
+)
+
+
+class FinancialReportData(BaseModel):
+    short_summary: str
+    """A short 2-3 sentence executive summary."""
+
+    markdown_report: str
+    """The full markdown report."""
+
+    follow_up_questions: list[str]
+    """Suggested follow-up questions for further research."""
+
+
+# Note: We will attach tools to specialist analyst agents at runtime in the manager.
+# This shows how an agent can use tools to delegate to specialized subagents.
+def new_writer_agent() -> Agent:
+    return Agent(
+        name="FinancialWriterAgent",
+        instructions=WRITER_PROMPT,
+        model="gpt-4.1-2025-04-14",
+        output_type=FinancialReportData,
+    )
@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Sequence
+
+from agents import RunConfig, Runner, RunResult, custom_span, trace
+from temporalio import workflow
+
+from openai_agents.financial_research_agent.agents.financials_agent import (
+    new_financials_agent,
+)
+from openai_agents.financial_research_agent.agents.planner_agent import (
+    FinancialSearchItem,
+    FinancialSearchPlan,
+    new_planner_agent,
+)
+from openai_agents.financial_research_agent.agents.risk_agent import new_risk_agent
+from openai_agents.financial_research_agent.agents.search_agent import new_search_agent
+from openai_agents.financial_research_agent.agents.verifier_agent import (
+    VerificationResult,
+    new_verifier_agent,
+)
+from openai_agents.financial_research_agent.agents.writer_agent import (
+    FinancialReportData,
+    new_writer_agent,
+)
+
+
+async def _summary_extractor(run_result: RunResult) -> str:
+    """Custom output extractor for sub-agents that return an AnalysisSummary."""
+    # The financial/risk analyst agents emit an AnalysisSummary with a `summary` field.
+    # We want the tool call to return just that summary text so the writer can drop it inline.
+    return str(run_result.final_output.summary)
+
+
+class FinancialResearchManager:
+    """
+    Orchestrates the full flow: planning, searching, sub-analysis, writing, and verification.
+    """
+
+    def __init__(self) -> None:
+        self.run_config = RunConfig()
+        self.planner_agent = new_planner_agent()
+        self.search_agent = new_search_agent()
+        self.financials_agent = new_financials_agent()
+        self.risk_agent = new_risk_agent()
+        self.writer_agent = new_writer_agent()
+        self.verifier_agent = new_verifier_agent()
+
+    async def run(self, query: str) -> str:
+        with trace("Financial research trace"):
+            search_plan = await self._plan_searches(query)
+            search_results = await self._perform_searches(search_plan)
+            report = await self._write_report(query, search_results)
+            verification = await self._verify_report(report)
+
+        # Return formatted output
+        result = f"""=====REPORT=====
+
+{report.markdown_report}
+
+=====FOLLOW UP QUESTIONS=====
+
+{chr(10).join(report.follow_up_questions)}
+
+=====VERIFICATION=====
+
+Verified: {verification.verified}
+Issues: {verification.issues}"""
+
+        return result
+
+    async def _plan_searches(self, query: str) -> FinancialSearchPlan:
+        result = await Runner.run(
+            self.planner_agent,
+            f"Query: {query}",
+            run_config=self.run_config,
+        )
+        return result.final_output_as(FinancialSearchPlan)
+
+    async def _perform_searches(
+        self, search_plan: FinancialSearchPlan
+    ) -> Sequence[str]:
+        with custom_span("Search the web"):
+            tasks = [
+                asyncio.create_task(self._search(item)) for item in search_plan.searches
+            ]
+            results: list[str] = []
+            for task in workflow.as_completed(tasks):
+                result = await task
+                if result is not None:
+                    results.append(result)
+            return results
+
+    async def _search(self, item: FinancialSearchItem) -> str | None:
+        input_data = f"Search term: {item.query}\nReason: {item.reason}"
+        try:
+            result = await Runner.run(
+                self.search_agent,
+                input_data,
+                run_config=self.run_config,
+            )
+            return str(result.final_output)
+        except Exception:
+            return None
+
+    async def _write_report(
+        self, query: str, search_results: Sequence[str]
+    ) -> FinancialReportData:
+        # Expose the specialist analysts as tools so the writer can invoke them inline
+        # and still produce the final FinancialReportData output.
+        fundamentals_tool = self.financials_agent.as_tool(
+            tool_name="fundamentals_analysis",
+            tool_description="Use to get a short write-up of key financial metrics",
+            custom_output_extractor=_summary_extractor,
+        )
+        risk_tool = self.risk_agent.as_tool(
+            tool_name="risk_analysis",
+            tool_description="Use to get a short write-up of potential red flags",
+            custom_output_extractor=_summary_extractor,
+        )
+        writer_with_tools = self.writer_agent.clone(
+            tools=[fundamentals_tool, risk_tool]
+        )
+
+        input_data = (
+            f"Original query: {query}\nSummarized search results: {search_results}"
+        )
+        result = await Runner.run(
+            writer_with_tools,
+            input_data,
+            run_config=self.run_config,
+        )
+        return result.final_output_as(FinancialReportData)
+
+    async def _verify_report(self, report: FinancialReportData) -> VerificationResult:
+        result = await Runner.run(
+            self.verifier_agent,
+            report.markdown_report,
+            run_config=self.run_config,
+        )
+        return result.final_output_as(VerificationResult)