2026/9/16 13:55:26
FEATURED · 精选文章 
基于 Agents SDK 构建金融研究多 Agent 流水线规划、搜索、子分析师与验证全流程实战CAI 仓库 financial_research_agent 示例解析【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai本指南以 CAI 仓库中的 financial_research_agent 示例 为核心完整讲解如何用 Agents SDK 组装一个规划 → 搜索 → 子分析师 → 写作 → 验证五段式金融研究 Agent 流水线。读完本文你将掌握多 Agent 编排的完整套路如何让 Writer Agent 通过as_tool内联调用专家子 Agent、如何用并发搜索与流式输出提升体验、如何用结构化输出Pydantic 模型约束每个环节的产物并能直接套用这套骨架搭建你自己的研究报告生成系统。示例定位从 research_bot 到专业金融研究流水线该示例位于 examples/financial_research_agent/其设计模式与仓库中的 research_bot 示例 一脉相承但在两个维度上做了深化子 Agent 更专业不再只有一个泛化的 writer而是拆出了 fundamentals基本面分析师与 risk风险分析师两个专职专家引入验证环节最终由 verifier审计 Agent对整篇报告做一致性、来源与主张核查形成写完再审的闭环。README 明确给出了五步流程这正是整篇文章的骨架Planning规划planner Agent 把用户请求转化为一组与财务分析相关的搜索词——近期新闻、财报电话会、公司备案文件、行业评论等Search搜索search Agent 使用内置的WebSearchTool为每个搜索词抓取精简摘要若已索引 PDF 或 10-K 文档还可追加FileSearchToolSub-analysts子分析师fundamentals analyst 与 risk analyst 以工具形式暴露给 writerwriter 可内联调用并吸收其输出Writing写作senior writer Agent 把搜索片段与子分析师摘要整合为长文 Markdown 报告外加一段精简的 executive summaryVerification验证verifier Agent 最终审计报告标记明显不一致或来源缺失之处。目录结构一个 Agent 一个文件examples/financial_research_agent/ ├── __init__.py ├── main.py # 命令行入口 ├── manager.py # 全流程编排FinancialResearchManager ├── printer.py # Rich 实时状态打印 └── agents/ ├── __init__.py ├── planner_agent.py # 规划 Agent FinancialSearchPlan ├── search_agent.py # 搜索 AgentWebSearchTool ├── financials_agent.py # 基本面分析师AnalysisSummary ├── risk_agent.py # 风险分析师AnalysisSummary ├── writer_agent.py # 高级写作 AgentFinancialReportData └── verifier_agent.py # 审计 AgentVerificationResult每个 Agent 与每种结构化输出类型各自独立成文件职责单一、便于按需替换——这也是该示例希望传达的工程组织方式。运行方式与示例查询README 给出的启动命令是模块化执行而非直接运行脚本python -m examples.financial_research_agent.main从 main.py 的实现可以看到入口逻辑程序读取用户输入后交由FinancialResearchManager.run()处理——async def main() - None: query input(Enter a financial research query: ) mgr FinancialResearchManager() await mgr.run(query) if __name__ __main__: asyncio.run(main())输入一个示例查询即可体验完整流水线Write up an analysis of Apple Inc.s most recent quarter.五段式工作流的逐环节剖析阶段一规划Planning——把问题拆成搜索清单planner_agent.py 中定义了规划 Agent 及其结构化输出。其系统提示词明确要求产出近期头条、财报电话会或 10-K 片段、分析师评论、行业背景四类检索目标且搜索词数量控制在 515 个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. )输出模型FinancialSearchPlan由若干FinancialSearchItem组成每个搜索项都带query要喂给搜索的检索词与reason该检索为何相关的理由这样后续搜索阶段可以拿到带上下文的输入class FinancialSearchItem(BaseModel): reason: str # 该搜索为何相关的理由 query: str # 喂给 web/file search 的检索词 class FinancialSearchPlan(BaseModel): searches: list[FinancialSearchItem] # 要执行的搜索列表规划 Agent 声明output_typeFinancialSearchPlan模型选用o3-mini保证返回结果严格符合该结构manager 侧再通过result.final_output_as(FinancialSearchPlan)取回。阶段二搜索Search——WebSearchTool 并行抓取search_agent.py 是最简单的单工具 Agent只挂一个WebSearchTool()并通过ModelSettings(tool_choicerequired)强制模型必须调用搜索工具避免模型跳过检索直接编造search_agent Agent( nameFinancialSearchAgent, instructionsINSTRUCTIONS, tools[WebSearchTool()], model_settingsModelSettings(tool_choicerequired), )其指令把摘要长度压到300 词以内并要求聚焦关键数字、事件、引用确保下游 writer 拿到的是高密度、可引用的素材。并发是这一阶段的亮点。manager.py 的_perform_searches用asyncio.create_task为每个搜索项建独立任务再用asyncio.as_completed边完成边汇总并实时更新进度Searching... 3/7 completed单条搜索失败则捕获异常返回None不拖垮整条流水线async def _search(self, item: FinancialSearchItem) - str | None: input_data fSearch term: {item.query}\nReason: {item.reason} try: result await Runner.run(search_agent, input_data) return str(result.final_output) except Exception: return None整个搜索阶段还被包进一个custom_span(Search the web)用于链路追踪详见后文 Tracing 一节。README 还提示若已有索引化的 PDF 或 10-K 文档可在tools列表中追加FileSearchTool()实现Web 检索 私有文档检索双通道。阶段三子分析师Sub-analysts——专家 Agent 化身工具financials_agent与risk_agent是两个专科医生都输出统一的AnalysisSummary单字段summary并且提示词都把篇幅压在两段以内fundamentalsfinancials_agent.py聚焦营收、利润、利润率与增长轨迹提取关键指标或引语riskrisk_agent.py排查竞争威胁、监管问题、供应链风险、增长放缓等 red flags。关键设计在于 README 第 3 条所说的exposed as tools。manager 在写作阶段用as_tool()把这两个 Agent 包装成 writer 可内联调用的工具详见阶段四这样 writer 可以在行文过程中按需咨询专家而不是把整条管线串行化。阶段四写作Writing——Writer Agent 与内联专家工具writer_agent.py 定义了写作 Agent其输出模型FinancialReportData含三个字段正好对应 README 里长文报告 执行摘要 后续问题的三重要求class FinancialReportData(BaseModel): short_summary: str # 2-3 句执行摘要 markdown_report: str # 完整 Markdown 报告 follow_up_questions: list[str] # 建议的后续研究问题README 给出的 Starter prompt 正是WRITER_PROMPT的原文两者一致明确 writer 的职责是把原始查询与搜索摘要合成为至少数段的长文 Markdown 报告 简短执行摘要并可调用fundamentals_analysis、risk_analysis工具获取专家短文最后追加几个后续研究问题。运行时动态挂载工具是这段代码最值得学习的地方。manager.py 没有在定义 writer 时写死工具而是在运行前用clone()生成一个带工具的副本fundamentals_tool financials_agent.as_tool( tool_namefundamentals_analysis, tool_descriptionUse to get a short write‑up of key financial metrics, custom_output_extractor_summary_extractor, ) risk_tool risk_agent.as_tool( tool_namerisk_analysis, tool_descriptionUse 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])这里的custom_output_extractor是点睛之笔子分析师返回的是AnalysisSummary对象而 writer 内联调用时希望拿到的是纯文本摘要于是通过提取器只把summary字段作为工具返回值注入正文async def _summary_extractor(run_result: RunResult) - str: return str(run_result.final_output.summary)clone()、as_tool()、custom_output_extractor均可在 SDK 源码中查到实现agent.py 中clone用于派生子 Agentas_tool在custom_output_extractor存在时调用它转换RunResult输出。这意味着把任意 Agent 降格为另一 Agent 的工具是该框架的一等公民能力。写作阶段还使用了Runner.run_streamed流式执行并在 manager.py 中每 5 秒切换一次提示文案Planning report structure... → Writing sections... → Finalizing report...让长时生成过程对用户可见、不卡顿。阶段五验证Verification——审计 Agent 兜底verifier_agent.py 扮演苛刻审计员检查报告内部一致性、来源是否清晰、是否存在无依据的主张。输出模型VerificationResult给出二元结论与问题说明class VerificationResult(BaseModel): verified: bool # 报告是否连贯可信 issues: str # 未通过时描述主要问题/疑虑manager 把report.markdown_report直接喂给 verifier得到VerificationResult后随报告一并打印见下节。这一步为整条流水线补上了质量闸门。编排中枢FinancialResearchManager 与链路追踪manager.py 的FinancialResearchManager.run()是整条流水线的指挥中心其调用顺序严格对应 README 的五步search_plan await self._plan_searches(query) # 1 规划 search_results await self._perform_searches(search_plan) # 2 并发搜索 report await self._write_report(query, search_results) # 34 子分析 写作 verification await self._verify_report(report) # 5 验证两个值得注意的工程细节链路追踪整个run()被包进with trace(Financial research trace, trace_idtrace_id):其中trace_id由gen_trace_id()生成manager 会把追踪链接打印到终端View trace: https://platform.openai.com/traces/{trace_id}搜索阶段另有custom_span细分便于回放每一步的输入输出Rich 实时状态面板printer.py 用rich.live.Live维护一个动态面板未完成项显示旋转 SpinnerPlanning searches...完成项打上 ✅ 对勾hide_done_checkmark则用于 trace_id 这类无需对勾的行。输出形态报告、后续问题与验证结论流水线跑完后manager 在 manager.py 中分四段打印结果REPORT Report: {report.markdown_report} # 完整长文 Markdown 报告 FOLLOW UP QUESTIONS {每条 follow_up_questions} # 建议的后续研究问题 VERIFICATION {VerificationResult} # verified 布尔值 issues 说明同时实时面板会先展示精简执行摘要Report summary\n\n{report.short_summary}。这种摘要先行、细节后置的输出结构非常适合接入邮件摘要、终端简报等场景。定制化指南把它改造成你自己的研究管线README 结尾明确指出这些 prompt 与子 Agent 完全可以按你的数据源与报告偏好进行调整。结合源码可落地的改造点包括定制维度修改位置说明换报告风格writer_agent.py 的WRITER_PROMPT改指令如输出中文增加估值章节或改FinancialReportData字段增加检索源search_agent.py 的tools追加FileSearchTool()检索已索引的 10-K / PDF扩充专家团队新建 Agent 在_write_report中as_tool如增加 ESG、供应链、宏观等分析师注意各专家的AnalysisSummary输出与专属提取器调整检索深度planner 的PROMPT修改搜索词数量区间当前 515 个与检索主题类型强化质控verifier 的VERIFIER_PROMPT增加对数据引用必须带日期/来源等硬性要求更换模型各Agent(model...)当前示例分别使用o3-mini规划、gpt-4.5-preview-2025-02-27写作、gpt-4o验证可按预算与延迟调整小结financial_research_agent示例虽然体量不大却浓缩了 Agents SDK 的五项核心能力结构化输出Pydantic 约束每个环节产物、Agent 即工具as_toolcustom_output_extractor让专家内联协作、并发执行asyncio并行搜索、流式运行run_streamed长任务体验优化、链路追踪trace/custom_span全程可回放。把这套规划-搜索-分析-写作-验证的骨架迁移到投研、尽调、合规审查等任意先取证、再成文、后校验的任务上即可快速构建你自己的多 Agent 研究报告系统。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
基于 Agents SDK 构建金融研究多 Agent 流水线：规划、搜索、子分析师与验证全流程实战（CAI 仓库 financial_research_agent 示例解析）
基于 Agents SDK 构建金融研究多 Agent 流水线规划、搜索、子分析师与验证全流程实战CAI 仓库 financial_research_agent 示例解析【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai本指南以 CAI 仓库中的 financial_research_agent 示例 为核心完整讲解如何用 Agents SDK 组装一个规划 → 搜索 → 子分析师 → 写作 → 验证五段式金融研究 Agent 流水线。读完本文你将掌握多 Agent 编排的完整套路如何让 Writer Agent 通过as_tool内联调用专家子 Agent、如何用并发搜索与流式输出提升体验、如何用结构化输出Pydantic 模型约束每个环节的产物并能直接套用这套骨架搭建你自己的研究报告生成系统。示例定位从 research_bot 到专业金融研究流水线该示例位于 examples/financial_research_agent/其设计模式与仓库中的 research_bot 示例 一脉相承但在两个维度上做了深化子 Agent 更专业不再只有一个泛化的 writer而是拆出了 fundamentals基本面分析师与 risk风险分析师两个专职专家引入验证环节最终由 verifier审计 Agent对整篇报告做一致性、来源与主张核查形成写完再审的闭环。README 明确给出了五步流程这正是整篇文章的骨架Planning规划planner Agent 把用户请求转化为一组与财务分析相关的搜索词——近期新闻、财报电话会、公司备案文件、行业评论等Search搜索search Agent 使用内置的WebSearchTool为每个搜索词抓取精简摘要若已索引 PDF 或 10-K 文档还可追加FileSearchToolSub-analysts子分析师fundamentals analyst 与 risk analyst 以工具形式暴露给 writerwriter 可内联调用并吸收其输出Writing写作senior writer Agent 把搜索片段与子分析师摘要整合为长文 Markdown 报告外加一段精简的 executive summaryVerification验证verifier Agent 最终审计报告标记明显不一致或来源缺失之处。目录结构一个 Agent 一个文件examples/financial_research_agent/ ├── __init__.py ├── main.py # 命令行入口 ├── manager.py # 全流程编排FinancialResearchManager ├── printer.py # Rich 实时状态打印 └── agents/ ├── __init__.py ├── planner_agent.py # 规划 Agent FinancialSearchPlan ├── search_agent.py # 搜索 AgentWebSearchTool ├── financials_agent.py # 基本面分析师AnalysisSummary ├── risk_agent.py # 风险分析师AnalysisSummary ├── writer_agent.py # 高级写作 AgentFinancialReportData └── verifier_agent.py # 审计 AgentVerificationResult每个 Agent 与每种结构化输出类型各自独立成文件职责单一、便于按需替换——这也是该示例希望传达的工程组织方式。运行方式与示例查询README 给出的启动命令是模块化执行而非直接运行脚本python -m examples.financial_research_agent.main从 main.py 的实现可以看到入口逻辑程序读取用户输入后交由FinancialResearchManager.run()处理——async def main() - None: query input(Enter a financial research query: ) mgr FinancialResearchManager() await mgr.run(query) if __name__ __main__: asyncio.run(main())输入一个示例查询即可体验完整流水线Write up an analysis of Apple Inc.s most recent quarter.五段式工作流的逐环节剖析阶段一规划Planning——把问题拆成搜索清单planner_agent.py 中定义了规划 Agent 及其结构化输出。其系统提示词明确要求产出近期头条、财报电话会或 10-K 片段、分析师评论、行业背景四类检索目标且搜索词数量控制在 515 个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. )输出模型FinancialSearchPlan由若干FinancialSearchItem组成每个搜索项都带query要喂给搜索的检索词与reason该检索为何相关的理由这样后续搜索阶段可以拿到带上下文的输入class FinancialSearchItem(BaseModel): reason: str # 该搜索为何相关的理由 query: str # 喂给 web/file search 的检索词 class FinancialSearchPlan(BaseModel): searches: list[FinancialSearchItem] # 要执行的搜索列表规划 Agent 声明output_typeFinancialSearchPlan模型选用o3-mini保证返回结果严格符合该结构manager 侧再通过result.final_output_as(FinancialSearchPlan)取回。阶段二搜索Search——WebSearchTool 并行抓取search_agent.py 是最简单的单工具 Agent只挂一个WebSearchTool()并通过ModelSettings(tool_choicerequired)强制模型必须调用搜索工具避免模型跳过检索直接编造search_agent Agent( nameFinancialSearchAgent, instructionsINSTRUCTIONS, tools[WebSearchTool()], model_settingsModelSettings(tool_choicerequired), )其指令把摘要长度压到300 词以内并要求聚焦关键数字、事件、引用确保下游 writer 拿到的是高密度、可引用的素材。并发是这一阶段的亮点。manager.py 的_perform_searches用asyncio.create_task为每个搜索项建独立任务再用asyncio.as_completed边完成边汇总并实时更新进度Searching... 3/7 completed单条搜索失败则捕获异常返回None不拖垮整条流水线async def _search(self, item: FinancialSearchItem) - str | None: input_data fSearch term: {item.query}\nReason: {item.reason} try: result await Runner.run(search_agent, input_data) return str(result.final_output) except Exception: return None整个搜索阶段还被包进一个custom_span(Search the web)用于链路追踪详见后文 Tracing 一节。README 还提示若已有索引化的 PDF 或 10-K 文档可在tools列表中追加FileSearchTool()实现Web 检索 私有文档检索双通道。阶段三子分析师Sub-analysts——专家 Agent 化身工具financials_agent与risk_agent是两个专科医生都输出统一的AnalysisSummary单字段summary并且提示词都把篇幅压在两段以内fundamentalsfinancials_agent.py聚焦营收、利润、利润率与增长轨迹提取关键指标或引语riskrisk_agent.py排查竞争威胁、监管问题、供应链风险、增长放缓等 red flags。关键设计在于 README 第 3 条所说的exposed as tools。manager 在写作阶段用as_tool()把这两个 Agent 包装成 writer 可内联调用的工具详见阶段四这样 writer 可以在行文过程中按需咨询专家而不是把整条管线串行化。阶段四写作Writing——Writer Agent 与内联专家工具writer_agent.py 定义了写作 Agent其输出模型FinancialReportData含三个字段正好对应 README 里长文报告 执行摘要 后续问题的三重要求class FinancialReportData(BaseModel): short_summary: str # 2-3 句执行摘要 markdown_report: str # 完整 Markdown 报告 follow_up_questions: list[str] # 建议的后续研究问题README 给出的 Starter prompt 正是WRITER_PROMPT的原文两者一致明确 writer 的职责是把原始查询与搜索摘要合成为至少数段的长文 Markdown 报告 简短执行摘要并可调用fundamentals_analysis、risk_analysis工具获取专家短文最后追加几个后续研究问题。运行时动态挂载工具是这段代码最值得学习的地方。manager.py 没有在定义 writer 时写死工具而是在运行前用clone()生成一个带工具的副本fundamentals_tool financials_agent.as_tool( tool_namefundamentals_analysis, tool_descriptionUse to get a short write‑up of key financial metrics, custom_output_extractor_summary_extractor, ) risk_tool risk_agent.as_tool( tool_namerisk_analysis, tool_descriptionUse 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])这里的custom_output_extractor是点睛之笔子分析师返回的是AnalysisSummary对象而 writer 内联调用时希望拿到的是纯文本摘要于是通过提取器只把summary字段作为工具返回值注入正文async def _summary_extractor(run_result: RunResult) - str: return str(run_result.final_output.summary)clone()、as_tool()、custom_output_extractor均可在 SDK 源码中查到实现agent.py 中clone用于派生子 Agentas_tool在custom_output_extractor存在时调用它转换RunResult输出。这意味着把任意 Agent 降格为另一 Agent 的工具是该框架的一等公民能力。写作阶段还使用了Runner.run_streamed流式执行并在 manager.py 中每 5 秒切换一次提示文案Planning report structure... → Writing sections... → Finalizing report...让长时生成过程对用户可见、不卡顿。阶段五验证Verification——审计 Agent 兜底verifier_agent.py 扮演苛刻审计员检查报告内部一致性、来源是否清晰、是否存在无依据的主张。输出模型VerificationResult给出二元结论与问题说明class VerificationResult(BaseModel): verified: bool # 报告是否连贯可信 issues: str # 未通过时描述主要问题/疑虑manager 把report.markdown_report直接喂给 verifier得到VerificationResult后随报告一并打印见下节。这一步为整条流水线补上了质量闸门。编排中枢FinancialResearchManager 与链路追踪manager.py 的FinancialResearchManager.run()是整条流水线的指挥中心其调用顺序严格对应 README 的五步search_plan await self._plan_searches(query) # 1 规划 search_results await self._perform_searches(search_plan) # 2 并发搜索 report await self._write_report(query, search_results) # 34 子分析 写作 verification await self._verify_report(report) # 5 验证两个值得注意的工程细节链路追踪整个run()被包进with trace(Financial research trace, trace_idtrace_id):其中trace_id由gen_trace_id()生成manager 会把追踪链接打印到终端View trace: https://platform.openai.com/traces/{trace_id}搜索阶段另有custom_span细分便于回放每一步的输入输出Rich 实时状态面板printer.py 用rich.live.Live维护一个动态面板未完成项显示旋转 SpinnerPlanning searches...完成项打上 ✅ 对勾hide_done_checkmark则用于 trace_id 这类无需对勾的行。输出形态报告、后续问题与验证结论流水线跑完后manager 在 manager.py 中分四段打印结果REPORT Report: {report.markdown_report} # 完整长文 Markdown 报告 FOLLOW UP QUESTIONS {每条 follow_up_questions} # 建议的后续研究问题 VERIFICATION {VerificationResult} # verified 布尔值 issues 说明同时实时面板会先展示精简执行摘要Report summary\n\n{report.short_summary}。这种摘要先行、细节后置的输出结构非常适合接入邮件摘要、终端简报等场景。定制化指南把它改造成你自己的研究管线README 结尾明确指出这些 prompt 与子 Agent 完全可以按你的数据源与报告偏好进行调整。结合源码可落地的改造点包括定制维度修改位置说明换报告风格writer_agent.py 的WRITER_PROMPT改指令如输出中文增加估值章节或改FinancialReportData字段增加检索源search_agent.py 的tools追加FileSearchTool()检索已索引的 10-K / PDF扩充专家团队新建 Agent 在_write_report中as_tool如增加 ESG、供应链、宏观等分析师注意各专家的AnalysisSummary输出与专属提取器调整检索深度planner 的PROMPT修改搜索词数量区间当前 515 个与检索主题类型强化质控verifier 的VERIFIER_PROMPT增加对数据引用必须带日期/来源等硬性要求更换模型各Agent(model...)当前示例分别使用o3-mini规划、gpt-4.5-preview-2025-02-27写作、gpt-4o验证可按预算与延迟调整小结financial_research_agent示例虽然体量不大却浓缩了 Agents SDK 的五项核心能力结构化输出Pydantic 约束每个环节产物、Agent 即工具as_toolcustom_output_extractor让专家内联协作、并发执行asyncio并行搜索、流式运行run_streamed长任务体验优化、链路追踪trace/custom_span全程可回放。把这套规划-搜索-分析-写作-验证的骨架迁移到投研、尽调、合规审查等任意先取证、再成文、后校验的任务上即可快速构建你自己的多 Agent 研究报告系统。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
