TradingAgents-CN 的 LangGraph stream_mode 双模策略：节点级进度跟踪与状态累积的实现剖析-北京尧图网络

TradingAgents-CN 的 LangGraph stream_mode 双模策略节点级进度跟踪与状态累积的实现剖析【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN本文围绕 stream_mode 修改影响分析展开解析 TradingAgents-CN基于多智能体 LLM 的中文金融交易框架为何将 LangGraph 图执行的stream_mode从固定的values改为按“是否携带进度回调”动态选择并给出 propagator、图执行主流程、后端分析服务三处源码级的实现证据与兼容性验证方法。读完后你将理解两种 stream 模式的 chunk 格式差异、updates 模式下的状态累积手法以及如何在保持 CLI、示例脚本等调用方零改动的同时为 Web 后端提供节点级实时进度。1. 背景为什么一个 stream_mode 参数会牵连整个调用链TradingAgents-CN 的核心执行引擎是一张 LangGraph 有向图市场分析师、基本面分析师、新闻分析师、社交媒体分析师产出报告看涨/看跌研究员进行辩论交易员给出决策风险团队激进/保守/中性分析师风险经理做最终风控。图的执行入口是TradingAgentsGraph.propagate()方法tradingagents/graph/trading_graph.py它通过self.graph.stream(init_agent_state, **args)流式消费每个节点的输出。LangGraph 的stream()支持不同的stream_mode其中最常用的两种在本项目中都出现了stream_modevalues每个 chunk 是完整的累积状态包含全部字段例如{messages: [...], market_report: ..., ...}。消费方可以直接读取chunk[messages]但无法区分“本次是哪个节点产出的”也就无法做节点粒度的进度上报。stream_modeupdates每个 chunk 只包含当前节点的增量更新键是节点名如Market Analyst值是该节点写入的状态字段。天然适合进度跟踪但消费方必须自己累积才能得到完整状态。项目早期get_graph_args()固定返回values。这带来一个矛盾后端 Web API 希望前端能看到“当前是市场分析师还是风险经理在跑”的实时进度对应前端轮询/api/analysis/tasks/{task_id}/status的进度条而 CLI 与示例脚本又依赖chunk[messages]这种完整状态格式直接取消息打印。二选一都会破坏另一方。文档给出的方案是把 stream_mode 的选择权交给调用方——有进度回调就走updates没有就保持values。2. 核心修改get_graph_args 的动态模式选择修改集中在 Propagator.get_graph_argstradingagents/graph/propagation.py。原文档记录的修改前后对比如下# 修改前 def get_graph_args(self) - Dict[str, Any]: return { stream_mode: values, config: {recursion_limit: self.max_recur_limit}, } # 修改后 def get_graph_args(self, use_progress_callback: bool False) - Dict[str, Any]: stream_mode updates if use_progress_callback else values return { stream_mode: stream_mode, config: {recursion_limit: self.max_recur_limit}, }当前仓库源码确认该修改已落地propagation.pydef get_graph_args(self, use_progress_callback: bool False) - Dict[str, Any]: Get arguments for the graph invocation. Args: use_progress_callback: If True, use updates mode for node-level progress tracking. If False, use values mode for complete state updates. # 使用 updates 模式可以获取节点级别的更新用于进度跟踪 # 使用 values 模式可以获取完整的状态更新 stream_mode updates if use_progress_callback else values return { stream_mode: stream_mode, config: {recursion_limit: self.max_recur_limit}, }设计要点有三个默认参数保持兼容use_progress_callback默认False任何旧调用get_graph_args()拿到的仍是values行为与修改前完全一致模式与能力绑定只有明确声明“我要节点级进度”的调用方才切换到updates避免无谓的状态累积成本recursion_limit不受影响max_recur_limitPropagator.__init__中默认 100见 propagation.py始终通过config下发防止多轮辩论导致递归超限。Propagator同时负责构建图的初始状态 create_initial_state注入一条明确的任务指令消息请对股票 {company_name} 进行全面分析交易日期为 {trade_date}。以及company_of_interest、trade_date、辩论状态InvestDebateState/RiskDebateState和四个空报告字段。理解这一点很重要updates 模式的状态累积正是以这份init_agent_state为基底的见第 4 节。3. propagate() 内的三分支执行逻辑stream_mode的选择最终由propagate()落地。在 trading_graph.py 中def propagate(self, company_name, trade_date, progress_callbackNone, task_idNone): ... init_agent_state self.propagator.create_initial_state(company_name, trade_date) ... # 根据是否有进度回调选择不同的stream_mode args self.propagator.get_graph_args(use_progress_callbackbool(progress_callback))注意原文档引用的行号第 372-402 行对应的是旧版本代码以当前仓库实际内容为准该逻辑位于 trading_graph.py 第 685-786 行。实际代码比原文档描述的更细致——根据self.debug开关与progress_callback是否存在共形成三条消费路径3.1 debug 分支按 stream_mode 区分处理trading_graph.py 第 688-729 行if self.debug: trace [] final_state None for chunk in self.graph.stream(init_agent_state, **args): # 记录节点计时略 if progress_callback and args.get(stream_mode) updates: # updates 模式chunk {Market Analyst: {...}} self._send_progress_update(chunk, progress_callback) # 累积状态更新 if final_state is None: final_state init_agent_state.copy() for node_name, node_update in chunk.items(): if not node_name.startswith(__): final_state.update(node_update) else: # values 模式chunk {messages: [...], ...} if len(chunk.get(messages, [])) 0: chunk[messages][-1].pretty_print() trace.append(chunk) final_state chunkupdates 路径每个 chunk 经_send_progress_update()转换为进度消息同时把节点增量合并进final_statevalues 路径保留原有行为取最后一条消息pretty_print()打印完整状态直接就是最新 chunk。3.2 非 debug 有进度回调updates 模式累积这是后端 Web API 实际走的路径trading_graph.py 第 730-759 行else: if progress_callback: # 使用 updates 模式以便获取节点级别的进度 trace [] final_state None for chunk in self.graph.stream(init_agent_state, **args): # 记录节点计时略 self._send_progress_update(chunk, progress_callback) # 累积状态更新 if final_state is None: final_state init_agent_state.copy() for node_name, node_update in chunk.items(): if not node_name.startswith(__): final_state.update(node_update)3.3 非 debug 无回调values 模式else: # 原有的invoke模式也需要计时 logger.info(⏱️ 使用 invoke 模式执行分析无进度回调) # 使用stream模式以便计时但不发送进度更新 for chunk in self.graph.stream(init_agent_state, **args): ... if final_state is None: final_state init_agent_state.copy() for node_name, node_update in chunk.items(): if not node_name.startswith(__): final_state.update(node_update)从源码结构看即便无回调的分支stream_mode仍是valuesget_graph_args未传True只是实现上仍用stream()而非invoke()以便记录节点计时累积写法保持一致保证三条路径返回的final_state语义相同。三条路径完成后propagate()统一追加性能指标并返回trading_graph.py 第 811-834 行final_state[performance_metrics] performance_data self.curr_state final_state self._log_state(trade_date, final_state) decision self.process_signal(final_state[final_trade_decision], company_name) decision[model_info] model_info return final_state, decision也就是说无论走哪种 stream_mode调用方拿到的(final_state, decision)形态不变——这是本次修改“对外透明”的关键。4. 状态累积逻辑updates 模式如何拼回完整状态原文档强调“状态累积逻辑已实现”当前仓库源码印证如下。累积的核心是四行trading_graph.py 第 754-759 行# 累积状态更新 if final_state is None: final_state init_agent_state.copy() # ✅ 从初始状态开始 for node_name, node_update in chunk.items(): if not node_name.startswith(__): final_state.update(node_update) # ✅ 逐步累积每个节点的更新验证要点逐项对应初始状态正确复制基底来自create_initial_state()保证messages、company_of_interest、trade_date、辩论状态等字段从第一步就存在逐节点合并updates 模式下chunk形如{Market Analyst: {messages: [...], market_report: ...}}final_state.update(node_update)把每个节点写入的字段覆盖进总状态跳过特殊键node_name.startswith(__)过滤掉__end__这类控制节点避免其空更新污染状态最终状态完整所有分析师报告、辩论历史、final_trade_decision等字段按执行顺序依次落位第 3 节末尾的final_state[final_trade_decision]取值依赖这一点。原文档对累积逻辑的验证结论初始状态正确复制、每个节点的更新正确累积、跳过特殊键、最终状态包含所有字段与当前源码完全吻合。5. 进度消息的节点映射_send_progress_updateupdates 模式的 chunk 键是 LangGraph 的节点名英文而前端进度条展示的是中文步骤文案。中间的翻译工作由 _send_progress_updatetrading_graph.py 第 836-919 行完成内置一张节点映射表node_mapping { # 分析师节点 Market Analyst: 市场分析师, Fundamentals Analyst: 基本面分析师, News Analyst: 新闻分析师, Social Analyst: 社交媒体分析师, # 工具节点不发送进度更新避免重复 tools_market: None, tools_fundamentals: None, tools_news: None, tools_social: None, # 消息清理节点不发送进度更新 Msg Clear Market: None, Msg Clear Fundamentals: None, Msg Clear News: None, Msg Clear Social: None, # 研究员节点 Bull Researcher: 看涨研究员, Bear Researcher: 看跌研究员, Research Manager: 研究经理, # 交易员节点 Trader: 交易员决策, # 风险评估节点 Risky Analyst: 激进风险评估, Safe Analyst: ️ 保守风险评估, Neutral Analyst: ⚖️ 中性风险评估, Risk Judge: 风险经理, }实现上有两个值得注意的细节工具节点与消息清理节点映射为None即跳过这些节点紧跟在对应分析师之后执行若也发进度会造成进度条“回跳”或重复闪烁源码注释明确写了“避免重复”__end__哨兵检测到结束节点时回调progress_callback( 生成报告)让进度条进入收尾阶段未知节点兜底映射表未覆盖的节点会以 {node_name}形式透出并打 warning 日志方便后续补全映射。回调异常被try/except捕获并记录堆栈第 918-919 行进度上报失败不会影响图本身的执行——这也是把 stream_mode 改动的风险面控制在进度链路内的重要原因。6. chunk 格式对比values 与 updates这是原文档的核心内容之一也是排查KeyError: messages类问题的基础。6.1stream_modevalues默认chunk { messages: [ HumanMessage(...), AIMessage(...), ToolMessage(...), ... ], company_of_interest: 工商银行, trade_date: 2025-10-03, market_report: ..., fundamentals_report: ..., ... }特点包含完整状态可以直接访问chunk[messages]适合需要完整状态、逐条打印消息的场景如 CLI 交互输出。6.2stream_modeupdates进度跟踪chunk { Market Analyst: { messages: [AIMessage(...)], market_report: ... } } # 或 chunk { Bull Researcher: { messages: [AIMessage(...)], ... } }特点只包含当前节点的更新键名是节点名称适合进度跟踪场景⚠️ 需要累积状态才能获得完整状态第 4 节的四行代码。两种格式不可互换values 的 chunk 没有节点名键updates 的 chunk 没有顶层messages键。任何直接写死chunk[messages]的消费方一旦误入 updates 模式就会抛KeyError。这正是原文档将“向后兼容”作为首要设计目标的原因。7. 后端 API 侧进度回调如何落到 Redis 与 MongoDB原文档指出后端 API 是updates模式的唯一受益方。结合 app/services/simple_analysis_service.py 的源码完整链路是第一步定义节点 → 百分比映射表simple_analysis_service.py 第 1353-1372 行。百分比与RedisProgressTracker的步骤权重对齐node_progress_map { # 分析师阶段 (10% → 45%) 市场分析师: 27.5, 基本面分析师: 45, 新闻分析师: 27.5, 社交媒体分析师: 27.5, # 研究辩论阶段 (45% → 70%) 看涨研究员: 51.25, 看跌研究员: 57.5, 研究经理: 70, # 交易员阶段 (70% → 78%) 交易员决策: 78, # 风险评估阶段 (78% → 93%) 激进风险评估: 81.75, ️ 保守风险评估: 85.5, ⚖️ 中性风险评估: 89.25, 风险经理: 93, # 最终阶段 (93% → 100%) 生成报告: 97, }注意映射的键与第 5 节_send_progress_update的输出文案一一对应含 emoji两处必须同步维护否则会命中“未知节点只更新消息”的兜底分支。第二步定义回调并更新三处存储simple_analysis_service.py 第 1374-1471 行。graph_progress_callback(message)的核心策略是“只在进度增加时更新百分比避免覆盖虚拟步骤进度”def graph_progress_callback(message: str): try: if not progress_tracker: return progress_pct node_progress_map.get(message) if progress_pct is not None: current_progress progress_tracker.progress_data.get(progress_percentage, 0) if int(progress_pct) current_progress: # 更新 Redis 进度跟踪器 progress_tracker.update_progress({ progress_percentage: int(progress_pct), last_message: message }) # 同时更新内存和 MongoDB事件循环内用 create_task # 无事件循环时回退到同步 MongoClient 更新 analysis_tasks 集合 ... else: # 进度没有增加只更新消息 progress_tracker.update_progress({last_message: message}) else: # 未知节点只更新消息 progress_tracker.update_progress({last_message: message}) except Exception as e: logger.error(f❌ Graph进度回调失败: {e}, exc_infoTrue)一个实现细节回调运行在asyncio.to_thread拉起的同步工作线程里此时未必存在运行的事件循环因此代码做了双路径处理——有事件循环时用asyncio.create_task提交异步更新没有时回退到同步MongoClient直接更新analysis_tasks集合第 1403-1454 行。这说明进度链路要同时服务 Redis轮询接口、MongoDB任务持久化和内存缓存memory_manager三份状态。第三步把回调传入图执行simple_analysis_service.py 第 1476-1481 行state, decision trading_graph.propagate( request.stock_code, analysis_date, progress_callbackgraph_progress_callback, task_idtask_id )到这里闭环形成updates模式产出节点级 chunk →_send_progress_update翻译成中文文案 →graph_progress_callback映射为百分比 → 写入 Redis/MongoDB/内存 → 前端轮询任务状态接口渲染进度条。8. 各调用方式兼容性分析原文档的“调用方式分析”覆盖了五类调用方以下结合当前仓库源码逐一确认原文档引用的 CLI 行号 1244/1267 在现行代码中为 1247/1270已按实际内容更新8.1 后端 API 调用受影响已正确处理文件app/services/simple_analysis_service.py✅ 传递progress_callback自动进入updates模式✅ 节点级进度更新经第 7 节链路送达前端✅ 状态累积逻辑在 trading_graph.py 中实现返回的state完整。8.2 CLI 命令行调用无影响文件cli/main.pyargs graph.propagator.get_graph_args() # 第 1247 行不传参数 → values 模式 for chunk in graph.graph.stream(init_agent_state, **args): # 第 1270 行 if len(chunk[messages]) 0: # 第 1271 行直接访问 messages last_message chunk[messages][-1] # 处理消息...✅ 不传use_progress_callback默认values✅ chunk 格式仍是{messages: [...], ...}chunk[messages]取值安全✅ 代码逻辑完全兼容。8.3 示例脚本调用无影响文件examples/dashscope_examples/demo_dashscope_chinese.py 等state, decision ta.propagate(AAPL, 2024-01-15) # 不传递 progress_callback✅ 走values模式返回完整的最终状态✅propagate()的返回值契约final_state, decision未变。8.4 Web 界面调用无影响Web 界面不直接调用propagate而是经后端 API 发起分析任务由后端统一处理进度跟踪见第 7 节前端轮询/api/analysis/tasks/{task_id}/status获取进度。8.5 调试模式已正确处理文件tradingagents/graph/trading_graph.pyif progress_callback and args.get(stream_mode) updates: # updates 模式处理节点更新 self._send_progress_update(chunk, progress_callback) # 累积状态 else: # values 模式打印消息 if len(chunk.get(messages, [])) 0: chunk[messages][-1].pretty_print()✅ 以args.get(stream_mode)显式区分两种模式的处理逻辑✅ updates 模式发进度累积状态values 模式打印消息原有行为。9. 兼容性矩阵与关键优势原文档的结论表格如下“受影响”指会切换到updates模式而非产生故障调用方式是否受影响兼容性说明后端 API✅ 受影响✅ 兼容使用updates模式状态累积逻辑已实现CLI 命令行❌ 不受影响✅ 兼容使用默认的values模式示例脚本❌ 不受影响✅ 兼容使用默认的values模式Web 界面❌ 不受影响✅ 兼容通过后端 API 调用调试模式✅ 受影响✅ 兼容根据模式选择不同处理逻辑由此可以归纳出这一设计的四个关键优势向后兼容默认参数保持原有行为存量调用方零改动按需启用只有传递progress_callback时才使用updates模式能力随需求开启状态完整累积逻辑初始状态复制逐节点合并过滤特殊键确保final_state与values模式一样完整propagate()的返回值契约不变逻辑清晰代码中显式按stream_mode分支处理模式与消费方式一一对应便于排查。10. 测试建议原文档给出的验证清单可直接作为回归测试基线后端 API 测试验证进度更新是否正常Redis 中progress_percentage是否随节点推进而上升且不回退验证最终状态是否完整final_state中各报告字段、final_trade_decision均非空;验证分析结果是否正确decision 内容与values模式基线一致。CLI 测试验证命令行分析是否正常跑完验证消息打印是否正常确认没有KeyError: messages。示例脚本测试运行examples/dashscope_examples/demo_dashscope_chinese.py验证分析结果是否正确。补充一条源码可推导出的检查点node_progress_mapsimple_analysis_service.py的键必须与node_mappingtrading_graph.py的输出文案严格一致若日后改动任一侧的文案包括 emoji另一侧需同步更新否则进度条会停留在上一个节点只刷新消息。11. 故障排查原文档的三个典型问题对照当前源码更新排查位置如下问题 1后端进度不更新原因stream_mode仍然是valuesupdates 特有的节点级 chunk 根本没有产生排查确认 propagation.py 第 54-68 行的动态选择逻辑存在且propagate()调用处传入了progress_callbacktrading_graph.py 第 686 行的bool(progress_callback)会把它转成模式开关。问题 2CLI 报错KeyError: messages原因CLI 误用了updates模式chunk 里没有顶层messages键排查确认 cli/main.py 第 1247 行调用get_graph_args()时不传递任何参数保持values默认值。问题 3最终状态不完整原因updates 模式下的状态累积逻辑缺失或顺序错误排查检查 trading_graph.py 第 754-759 行非 debug 路径与第 712-717 行debug 路径的累积代码是否完整重点确认final_state init_agent_state.copy()的基底初始化与__前缀键的过滤。12. 结论这次修改的安全性来自一条清晰的兼容性边界stream_mode 的选择被收敛到get_graph_args()的默认参数上只有显式声明“需要进度”的调用方才会进入updates模式。后端 API 因此获得了节点级进度能力配合 Redis/MongoDB/内存三写与前端轮询形成完整进度链路而 CLI、示例脚本、Web 界面与调试模式要么走默认values模式、要么按模式分支处理行为均不受破坏。对维护多智能体框架的开发者而言这份分析也提供了一个通用范式当流式执行的消费格式发生变化时用“默认参数保兼容消费端自累积”组合可以把一次 chunk 格式变更的影响面精确控制在新增功能一侧。如需进一步理解整体数据流设计可继续阅读同目录下的 dataflows 架构分析与 dataflows 保守重构方案。【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

TradingAgents-CN 的 LangGraph stream_mode 双模策略：节点级进度跟踪与状态累积的实现剖析

相关推荐

用 Aliasing XOR Mutability 强化 API 不变量：comprehensive-rust 课程中的借阅检查器设计模式

OpenHuman 工具级记忆（Tool-Scoped Memory）：把“绝不要给 Sarah 发邮件”变成 Agent 必须遵守的硬约束

Django电商项目实战：SPU/SKU、Redis缓存与订单事务设计

更多资讯

索纳拉完成收购，宣称让大脑活动测量如常规生理信号般简单助力可穿戴设备

Trivy config 命令完全指南：用 `trivy config` 扫描 IaC 与配置文件中的错误配置

Karakeep 高级工作流实战：规则引擎、REST API 与 Webhooks 的自动化集成指南

Mojo 编译期值物化（Materialization）实战指南：把 comptime 数据安全地带到运行时

Hyperframes 嵌入式字幕审美验收：用 reference bar 与五项正向检查把关“世界级”渲染

YOLOv8自行车检测资源详解：数据集、权重训练与C++部署

今日更新

CopilotKit Vue Demo 与 React Demo 的 Parity 实现指南：路由、运行时端点与交互语义的全框架对齐

Harbor RBAC 实战：db_auth 模式下非管理员用户管理项目成员的完整验证与权限模型解析

STM32平衡车纯平衡程序从拆包到参数整定的完整调试指南

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？