Adds example for financial agent

examples/financial_research_agent/README.md

@@ -0,0 +1,38 @@
+# Financial Research Agent Example
+
+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.
+
+You can run the example with:
+
+```bash
+python -m examples.financial_research_agent.main
+```
+
+and enter a query like:
+
+```
+Write up an analysis of Apple Inc.'s most recent quarter.
+```
+
+### Starter prompt
+
+The writer agent is seeded with instructions similar to:
+
+```
+You are a senior financial analyst. You will be provided with the original query
+and a set of raw search summaries. Your job is to synthesize these into a
+long‑form markdown report (at least several paragraphs) with a short executive
+summary. You also have access to tools like `fundamentals_analysis` and
+`risk_analysis` to get short specialist write‑ups if you want to incorporate them.
+Add a few follow‑up questions for further research.
+```
+
+You can tweak these prompts and sub‑agents to suit your own data sources and preferred report structure.

examples/financial_research_agent/agents/financials_agent.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+# 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."""
+
+
+financials_agent = Agent(
+    name="FundamentalsAnalystAgent",
+    instructions=FINANCIALS_PROMPT,
+    output_type=AnalysisSummary,
+)

examples/financial_research_agent/agents/planner_agent.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+# 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."""
+
+
+planner_agent = Agent(
+    name="FinancialPlannerAgent",
+    instructions=PROMPT,
+    model="o3-mini",
+    output_type=FinancialSearchPlan,
+)

examples/financial_research_agent/agents/risk_agent.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+# 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."""
+
+
+risk_agent = Agent(
+    name="RiskAnalystAgent",
+    instructions=RISK_PROMPT,
+    output_type=AnalysisSummary,
+)

examples/financial_research_agent/agents/search_agent.py

@@ -0,0 +1,18 @@
+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."
+)
+
+search_agent = Agent(
+    name="FinancialSearchAgent",
+    instructions=INSTRUCTIONS,
+    tools=[WebSearchTool()],
+    model_settings=ModelSettings(tool_choice="required"),
+)

examples/financial_research_agent/agents/verifier_agent.py

@@ -0,0 +1,27 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+# 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."""
+
+
+verifier_agent = Agent(
+    name="VerificationAgent",
+    instructions=VERIFIER_PROMPT,
+    model="gpt-4o",
+    output_type=VerificationResult,
+)

examples/financial_research_agent/agents/writer_agent.py

@@ -0,0 +1,34 @@
+from pydantic import BaseModel
+
+from agents import Agent
+
+# 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 handoffs to specialist analyst agents at runtime in the manager.
+# This shows how an agent can use handoffs to delegate to specialized subagents.
+writer_agent = Agent(
+    name="FinancialWriterAgent",
+    instructions=WRITER_PROMPT,
+    model="gpt-4.5-preview-2025-02-27",
+    output_type=FinancialReportData,
+)

examples/financial_research_agent/main.py

@@ -0,0 +1,17 @@
+import asyncio
+
+from .manager import FinancialResearchManager
+
+
+# Entrypoint for the financial bot example.
+# Run this as `python -m examples.financial_bot.main` and enter a
+# financial research query, for example:
+# "Write up an analysis of Apple Inc.'s most recent quarter."
+async def main() -> None:
+    query = input("Enter a financial research query: ")
+    mgr = FinancialResearchManager()
+    await mgr.run(query)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/financial_research_agent/manager.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from collections.abc import Sequence
+
+from rich.console import Console
+
+from agents import Runner, RunResult, custom_span, gen_trace_id, trace
+
+from .agents.financials_agent import financials_agent
+from .agents.planner_agent import FinancialSearchItem, FinancialSearchPlan, planner_agent
+from .agents.risk_agent import risk_agent
+from .agents.search_agent import search_agent
+from .agents.verifier_agent import VerificationResult, verifier_agent
+from .agents.writer_agent import FinancialReportData, writer_agent
+from .printer import Printer
+
+
+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.console = Console()
+        self.printer = Printer(self.console)
+
+    async def run(self, query: str) -> None:
+        trace_id = gen_trace_id()
+        with trace("Financial research trace", trace_id=trace_id):
+            self.printer.update_item(
+                "trace_id",
+                f"View trace: https://platform.openai.com/traces/{trace_id}",
+                is_done=True,
+                hide_checkmark=True,
+            )
+            self.printer.update_item(
+                "start", "Starting financial research...", is_done=True)
+            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)
+
+            final_report = f"Report summary\n\n{report.short_summary}"
+            self.printer.update_item(
+                "final_report", final_report, is_done=True)
+
+            self.printer.end()
+
+        # Print to stdout
+        print("\n\n=====REPORT=====\n\n")
+        print(f"Report:\n{report.markdown_report}")
+        print("\n\n=====FOLLOW UP QUESTIONS=====\n\n")
+        print("\n".join(report.follow_up_questions))
+        print("\n\n=====VERIFICATION=====\n\n")
+        print(verification)
+
+    async def _plan_searches(self, query: str) -> FinancialSearchPlan:
+        self.printer.update_item("planning", "Planning searches...")
+        result = await Runner.run(planner_agent, f"Query: {query}")
+        self.printer.update_item(
+            "planning",
+            f"Will perform {len(result.final_output.searches)} searches",
+            is_done=True,
+        )
+        return result.final_output_as(FinancialSearchPlan)
+
+    async def _perform_searches(self, search_plan: FinancialSearchPlan) -> Sequence[str]:
+        with custom_span("Search the web"):
+            self.printer.update_item("searching", "Searching...")
+            tasks = [asyncio.create_task(self._search(item))
+                     for item in search_plan.searches]
+            results: list[str] = []
+            num_completed = 0
+            for task in asyncio.as_completed(tasks):
+                result = await task
+                if result is not None:
+                    results.append(result)
+                num_completed += 1
+                self.printer.update_item(
+                    "searching", f"Searching... {num_completed}/{len(tasks)} completed"
+                )
+            self.printer.mark_item_done("searching")
+            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(search_agent, input_data)
+            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 = 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 = 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 = writer_agent.clone(
+            tools=[fundamentals_tool, risk_tool])
+        self.printer.update_item("writing", "Thinking about report...")
+        input_data = f"Original query: {query}\nSummarized search results: {search_results}"
+        result = Runner.run_streamed(writer_with_tools, input_data)
+        update_messages = [
+            "Planning report structure...",
+            "Writing sections...",
+            "Finalizing report...",
+        ]
+        last_update = time.time()
+        next_message = 0
+        async for _ in result.stream_events():
+            if time.time() - last_update > 5 and next_message < len(update_messages):
+                self.printer.update_item(
+                    "writing", update_messages[next_message])
+                next_message += 1
+                last_update = time.time()
+        self.printer.mark_item_done("writing")
+        return result.final_output_as(FinancialReportData)
+
+    async def _verify_report(self, report: FinancialReportData) -> VerificationResult:
+        self.printer.update_item("verifying", "Verifying report...")
+        result = await Runner.run(verifier_agent, report.markdown_report)
+        self.printer.mark_item_done("verifying")
+        return result.final_output_as(VerificationResult)