1. 从零拆解 TradingAgents一个中文 AI 股神多智能体系统到底在做什么第一次看到 TradingAgents 这个项目名我脑子里蹦出来的不是又一个股票分析工具而是终于有人把多智能体协作这套东西搬到 A 股场景里了。市面上大部分所谓 AI 炒股工具本质就是拿一个大模型套个壳喂几条新闻让它输出买入/卖出/持有这种单智能体方案最大的问题在于它既当分析师又当交易员还当风控角色混在一起输出自然是一锅粥。TradingAgents 的思路完全不同它用 LangGraph 把多个各司其职的智能体编排成一条流水线让每个 Agent 只干自己最擅长的那一段活最后汇总成一份带推理链的分析报告。这个项目解决的核心痛点其实很明确散户做股票分析时信息是碎片化的——基本面数据在一个地方、技术指标在另一个地方、新闻情绪又散落在各处人脑很难在短时间内把这些维度整合起来。TradingAgents 想做的就是把这套研究员团队的工作流自动化一个 Agent 负责读财报和基本面一个 Agent 盯技术面 K 线和量价一个 Agent 扫新闻和舆情还有一个专门唱反调做风险审查最后由一个决策 Agent综合所有意见给出结论。这套架构在英文社区已经有类似实现但针对中文语境、中文财经数据源、A 股特有逻辑做适配的还真不多见。适合谁来参考这个项目我把它分成三类人。第一类是懂点 Python、想入门 AI Agent 开发但不知道拿什么练手的开发者TradingAgents 是一个非常好的 LangGraph 实战案例因为它的图结构清晰、节点职责分明比那些玩具级的 demo 有嚼头得多。第二类是对量化分析感兴趣、想用 AI 辅助自己做研究的老股民你可以把它当成一个研究助理来用而不是当成自动赚钱机器。第三类是做 AI 应用的产品和技术负责人想看看多智能体协作在垂直领域怎么落地这个项目的编排思路可以直接迁移到其他行业。需要先把丑话说在前面任何 AI 股票分析系统的输出都只是参考它不能预测市场更不能替你做决策。我见过太多人把这类工具当成圣杯结果亏得一塌糊涂。TradingAgents 的价值在于帮你把分析流程结构化、把信息整合效率提上去而不是给你一个稳赚的答案。理解这一点后面的内容你才能看得进去。2. 多智能体架构设计为什么不用一个大模型硬扛2.1 单智能体方案的三个致命伤在讲 TradingAgents 的架构之前我得先说说为什么一个大模型包打天下这条路走不通。我早期也试过用单个大模型做股票分析提示词里塞满你是资深分析师请综合基本面、技术面、消息面给出建议结果输出质量惨不忍睹主要栽在三个地方。第一个是角色冲突。当你让一个模型同时扮演基本面分析师和风险控制官时它的输出会倾向于和稀泥。基本面看多、技术面看空的时候单模型往往给出一个模棱两可的谨慎乐观这种废话对决策毫无价值。而多智能体方案里基本面 Agent 可以坚定看多技术面 Agent 可以坚定看空最后由决策层去权衡这个分歧反而更接近真实投研团队的运作方式。第二个是上下文污染。股票分析涉及的数据量很大财报几十页、K 线几百根、新闻几十条全塞进一个上下文窗口模型注意力会被稀释重要信息被淹没。多智能体方案把任务拆开每个 Agent 只处理自己那一小块数据上下文干净推理质量自然高。第三个是无法追溯。单模型给出买入结论你根本不知道它是基于哪条信息得出的出了问题也没法复盘。多智能体方案每个节点都有独立的输出你能清楚看到技术面 Agent 因为 MACD 金叉看多但风险 Agent 因为估值过高提出警告这种可追溯性对投研来说太重要了。2.2 LangGraph 作为编排骨架的选型逻辑TradingAgents 选择 LangGraph 而不是 LangChain 的 AgentExecutor 或者自己手写调度这个选择我认为非常对。LangGraph 的核心抽象是状态图——你把整个分析流程定义成一张有向图节点是智能体边是数据流转路径整个图共享一个 State 对象。这个模型天然适配投研流程因为投研本身就是有先后依赖的你得先拿到数据才能做分析先做完各维度分析才能做综合决策。对比一下其他方案你就明白差距了。用 LangChain 的 AgentExecutor本质是 ReAct 循环模型自己决定下一步调什么工具流程是动态的、不可控的。股票分析这种场景流程其实是相对固定的你不需要模型自己想下一步干嘛你需要的是稳定、可复现的流水线。LangGraph 的静态图正好满足这个需求而且它支持条件边比如如果风险评分超过阈值就跳到风险复审节点这种分支逻辑用图来表达非常自然。还有一个关键优势是状态持久化。LangGraph 的 checkpointer 机制可以把每一步的 State 存下来这意味着分析过程可以中断、可以恢复、可以回放。对于耗时较长的股票分析任务要调多个数据源、跑多次模型推理这个特性很实用。我实测下来一次完整的 TradingAgents 分析流程大概要跑 8 到 12 个节点耗时从几十秒到几分钟不等中间任何一步失败都能从断点续跑不用从头再来。2.3 智能体角色划分与职责边界TradingAgents 的智能体划分遵循了真实投研团队的分工逻辑我把它整理成一张表方便你对照理解每个角色的定位。智能体角色核心职责输入数据输出内容基本面分析师解读财报、估值指标财务数据、PE/PB/ROE基本面评分与理由技术面分析师分析 K 线、量价、指标历史行情、MACD/RSI技术面评分与信号新闻舆情分析师扫描新闻、公告、情绪新闻文本、公告内容情绪倾向与事件影响多头研究员构建看多逻辑上述三方分析结果看多论据清单空头研究员构建看空逻辑上述三方分析结果看空论据清单风险控制官评估下行风险多空论据、持仓约束风险等级与警示决策智能体综合权衡出结论所有上游输出最终建议与置信度这张表里最值得说的是多头和空头研究员这对设计。很多同类项目只做分析不做辩论但真实投研里看多和看空两方的对抗恰恰是发现盲点的关键。TradingAgents 让两个 Agent 分别站在对立面构建论据然后交给风险控制官和决策智能体去裁决这种对抗式推理能有效减少单一视角带来的偏差。我自己测试的时候发现当基本面和技术面信号矛盾时多空辩论环节往往能挖出一些被忽略的风险点比如虽然技术面走强但大股东近期有减持公告这类信息。2.4 状态流转与数据契约设计多智能体系统最容易翻车的地方就是状态管理。如果每个 Agent 各写各的最后汇总时字段对不上整个系统就崩了。TradingAgents 用 LangGraph 的 State 对象作为全局数据契约所有 Agent 读写同一个结构化状态这个设计看似简单但落地时有几个细节必须处理好。首先是字段命名规范。我建议用统一的前缀区分数据来源比如fundamental_*开头的字段归基本面 Agent 管technical_*归技术面news_*归舆情。这样在决策节点汇总时一眼就能看出数据来自哪个维度不会混淆。其次是中间结果的版本管理。股票分析里同一个指标可能被多个 Agent 引用比如当前股价既被技术面用也被估值计算用。如果每个 Agent 各自去拉一次数据不仅浪费调用还可能出现数据不一致拉取时间不同导致价格不同。正确做法是在图的入口处统一拉取基础数据存进 State后续所有 Agent 都从这个 State 里读。最后是错误传播机制。如果某个数据源挂了比如新闻接口超时不能让整个流程崩掉。我的做法是给每个 Agent 的输出加一个status字段标记success、partial或failed决策节点根据这些状态决定是否降级处理。比如新闻数据缺失时决策智能体可以只基于基本面和技术面给结论并在报告里注明舆情维度数据缺失。3. 核心环节实操把 TradingAgents 跑起来的关键步骤3.1 环境准备与依赖安装动手之前先把环境理清楚。TradingAgents 这类项目对 Python 版本有要求我建议用 3.10 或 3.11太新的版本3.12有时候会遇到某些库的兼容问题。虚拟环境是必须的别图省事装在全局环境里依赖冲突会让你怀疑人生。# 创建虚拟环境 python -m venv trading_env source trading_env/bin/activate # Windows 用 trading_env\Scripts\activate # 安装核心依赖 pip install langgraph langchain langchain-openai pip install pandas numpy akshare # 数据源和数据处理 pip install pydantic # 状态对象定义这里重点说下数据源的选择。A 股数据我推荐用akshare它是开源免费的接口覆盖行情、财务、新闻等多个维度虽然偶尔有接口不稳定的时候但胜在免费且社区活跃。如果你要更稳定的数据可以考虑付费数据源但初期验证阶段没必要花这个钱。提示akshare 的接口经常因为上游数据源调整而失效建议在代码里做好异常捕获和重试逻辑别让一个接口挂掉拖垮整个流程。关于大模型的选择TradingAgents 需要模型具备较强的中文理解和推理能力。我实测下来国产模型在中文财经语境下的表现普遍不错尤其是对 A 股特有概念比如涨停板北向资金龙虎榜的理解比一些英文为主的模型更到位。你可以根据自己的预算和可用性选择关键是确保模型支持结构化输出JSON mode因为 Agent 之间的数据传递需要严格的格式。3.2 定义全局状态对象State 对象是整个系统的骨架定义得好不好直接决定后续开发顺不顺。我用 Pydantic 来定义因为它自带类型校验能在早期发现字段类型错误。from pydantic import BaseModel, Field from typing import Optional, List, Literal class AnalysisState(BaseModel): # 基础信息 stock_code: str stock_name: str analysis_date: str # 原始数据 price_data: Optional[dict] None financial_data: Optional[dict] None news_data: Optional[list] None # 各维度分析结果 fundamental_analysis: Optional[str] None technical_analysis: Optional[str] None news_analysis: Optional[str] None # 多空论据 bull_arguments: List[str] Field(default_factorylist) bear_arguments: List[str] Field(default_factorylist) # 风险与决策 risk_level: Optional[Literal[low, medium, high]] None final_decision: Optional[str] None confidence: Optional[float] None # 流程控制 data_status: dict Field(default_factorydict) error_log: List[str] Field(default_factorylist)这个 State 定义有几个设计考量。第一我把原始数据和分析结果分开存这样调试时能清楚看到是数据错了还是分析错了。第二data_status字段记录每个数据源的获取状态方便降级处理。第三error_log累积错误信息最后可以输出到报告里让用户知道哪些环节出了问题。3.3 构建 LangGraph 分析流程图图结构是整个项目的灵魂我把它拆成三个阶段数据采集、多维分析、综合决策。每个阶段内部又有若干节点节点之间用边连接条件边负责分支逻辑。from langgraph.graph import StateGraph, END def build_analysis_graph(): workflow StateGraph(AnalysisState) # 第一阶段数据采集 workflow.add_node(fetch_price, fetch_price_node) workflow.add_node(fetch_financial, fetch_financial_node) workflow.add_node(fetch_news, fetch_news_node) # 第二阶段多维分析 workflow.add_node(analyze_fundamental, fundamental_agent) workflow.add_node(analyze_technical, technical_agent) workflow.add_node(analyze_news, news_agent) # 第三阶段辩论与决策 workflow.add_node(bull_research, bull_agent) workflow.add_node(bear_research, bear_agent) workflow.add_node(risk_check, risk_agent) workflow.add_node(make_decision, decision_agent) # 定义流转路径 workflow.set_entry_point(fetch_price) workflow.add_edge(fetch_price, fetch_financial) workflow.add_edge(fetch_financial, fetch_news) # 数据采集完成后并行进入分析 workflow.add_edge(fetch_news, analyze_fundamental) workflow.add_edge(fetch_news, analyze_technical) workflow.add_edge(fetch_news, analyze_news) # 三个分析汇聚到辩论环节 workflow.add_edge(analyze_fundamental, bull_research) workflow.add_edge(analyze_technical, bull_research) workflow.add_edge(analyze_news, bull_research) workflow.add_edge(analyze_fundamental, bear_research) workflow.add_edge(analyze_technical, bear_research) workflow.add_edge(analyze_news, bear_research) # 辩论后进入风控和决策 workflow.add_edge(bull_research, risk_check) workflow.add_edge(bear_research, risk_check) workflow.add_edge(risk_check, make_decision) workflow.add_edge(make_decision, END) return workflow.compile()这里有个细节值得展开数据采集为什么是串行而不是并行。理论上三个数据源可以同时拉但实测下来串行更稳。原因是很多免费数据接口有频率限制并发请求容易触发限流反而更慢。串行虽然总耗时略长但成功率更高。如果你用的是付费数据源可以改成并行用 LangGraph 的并行边实现。3.4 单个智能体的实现范式每个 Agent 的实现套路其实高度相似读 State 里的相关数据构造提示词调用大模型解析输出写回 State。我以基本面分析师为例把关键代码和设计思路讲清楚。def fundamental_agent(state: AnalysisState) - dict: financial state.financial_data if not financial: return { fundamental_analysis: 财务数据缺失无法分析, data_status: {**state.data_status, fundamental: failed} } prompt f你是一位资深基本面分析师请基于以下财务数据分析 {state.stock_name} 营收与利润{financial.get(revenue_growth)} / {financial.get(profit_growth)} 估值指标PE{financial.get(pe)}, PB{financial.get(pb)} 盈利能力ROE{financial.get(roe)}, 毛利率{financial.get(gross_margin)} 请从成长性、估值合理性、盈利质量三个维度给出分析输出 JSON 格式 {{score: 0-100, reasoning: 分析理由, key_risks: [风险点]}} response llm.invoke(prompt) result parse_json(response.content) return { fundamental_analysis: result[reasoning], data_status: {**state.data_status, fundamental: success} }这段代码里有几个经验点。第一提示词里明确要求 JSON 输出并且给出格式示例这样解析起来稳定得多。第二数据缺失时返回降级结果而不是抛异常保证流程能继续。第三score 用 0-100 的数值方便后续决策节点做加权计算比让模型输出看多/看空这种离散值更有信息量。3.5 多空辩论环节的实现技巧多空辩论是 TradingAgents 最有特色的部分实现上也有讲究。核心思路是让多头 Agent 和空头 Agent 分别读取前面三个分析师的输出然后各自构建论据。这里的关键是提示词要引导对抗性思维。多头 Agent 的提示词我会这样写你是一位坚定的多头研究员请从所有分析结果中挖掘支持买入的证据即使整体信号偏空也要找出可能的反转逻辑。空头 Agent 则反过来你是一位谨慎的空头研究员请找出所有可能的风险点和看空理由即使整体信号偏多也要指出潜在的隐患。这种强制站队的设计能逼出模型更深层的推理。我实测发现如果提示词写得太中立两个 Agent 的输出会高度雷同辩论就失去意义了。强制站队后多头会去强调估值处于历史低位行业景气度回升这类积极因素空头则会揪住现金流紧张大股东质押比例高这类风险点两边一碰撞决策者能看到更全面的图景。3.6 决策智能体的加权逻辑最后的决策智能体不是简单地把前面所有输出拼起来让模型总结而是有一套结构化的加权逻辑。我的做法是给每个维度分配权重然后计算综合得分。维度权重说明基本面30%长期价值锚点技术面25%短期趋势参考舆情面15%事件驱动因素多空辩论20%分歧度调整风险控制10%一票否决项权重不是拍脑袋定的而是根据投资周期调整。如果你是做中长线基本面权重可以提到 40%做短线技术面权重提到 35%。风险控制虽然权重只有 10%但它有一票否决权——如果风控 Agent 判定风险等级为 high无论综合得分多高最终建议都要降级为观望。决策 Agent 的提示词里我会明确要求输出置信度这个置信度不是模型随便给的而是基于信号一致性计算的如果基本面、技术面、舆情面三个维度都指向同一方向置信度高如果三个维度分歧很大置信度低。这个设计能帮用户判断这个结论有多可靠。4. 常见问题与排查技巧实录4.1 数据源相关的坑做 A 股数据采集最大的坑就是接口不稳定。akshare 的很多接口底层是爬取公开数据上游页面一改版接口就失效。我踩过的坑包括财务数据接口突然返回空值、新闻接口返回的日期格式变了导致解析失败、行情接口在收盘后一段时间内数据不更新。应对策略是多数据源备份 字段级容错。比如行情数据主用 akshare备用可以接一个其他免费源主源失败时自动切换。字段解析时不要假设格式固定用 try-except 包住每个字段的解析单个字段失败不影响整体。还有一个隐蔽的坑是复权问题。技术面分析必须用复权价否则除权除息那天会出现巨大的价格跳空导致技术指标全部失真。我建议统一用前复权数据并且在 State 里明确标记复权方式避免不同 Agent 用了不同口径的数据。4.2 大模型输出不稳定的处理大模型输出 JSON 格式不稳定是另一个高频问题。即使你在提示词里明确要求 JSON模型偶尔还是会输出带 markdown 代码块包裹的内容或者字段名拼错。我的处理方案是三层防护第一层提示词里给完整的 JSON 示例第二层用正则先提取 JSON 部分再解析第三层解析失败时重试一次重试还失败就降级处理。import json import re def parse_json_safely(text: str, max_retry: int 2): for attempt in range(max_retry): try: # 尝试直接解析 return json.loads(text) except json.JSONDecodeError: # 尝试提取代码块内的 JSON match re.search(r(?:json)?\s*(\{.*?\})\s*, text, re.DOTALL) if match: try: return json.loads(match.group(1)) except: pass # 尝试提取第一个完整的大括号块 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group(0)) except: pass return None # 全部失败调用方做降级处理注意不要因为解析失败就无限重试那会浪费大量 token。设置合理的重试上限失败后走降级路径保证流程能走完。4.3 分析结果自相矛盾的排查多智能体系统一个典型问题是各 Agent 结论互相矛盾比如基本面说估值合理技术面说严重超买舆情说利空频出。这种矛盾本身不是 bug反而是有价值的信息但如果决策 Agent 处理不好就会输出一个自相矛盾的最终报告。排查这类问题的思路是先看是不是数据口径不一致导致的假矛盾。比如基本面 Agent 用的是最新季度财报技术面 Agent 用的是实时行情两者时间点不同结论自然可能矛盾。解决方法是统一数据时间戳在 State 里记录每个数据的采集时间分析时明确标注基于 X 月 X 日数据。如果排除了数据问题那矛盾就是真实的这时候决策 Agent 的任务不是消除矛盾而是呈现矛盾并给出权衡。我会在决策提示词里明确要求如果各维度信号存在分歧请在报告中明确指出分歧点并说明你的权衡逻辑不要强行给出统一结论。4.4 性能优化与成本控制跑一次完整分析如果每个 Agent 都调用一次大模型总共要调用 8 到 10 次token 消耗不小。我做过统计一次完整分析大概消耗 1.5 万到 3 万 token如果用贵价模型单次成本可能到几毛钱。频繁分析的话成本会累积。优化手段有几个。第一缓存基础数据同一只股票同一天的数据只拉一次多个 Agent 共享。第二精简提示词把不必要的数据字段去掉只保留分析真正需要的。第三分级调用模型简单的数据提取用便宜的小模型复杂的推理用强模型。第四结果缓存如果同一只股票在短时间内被重复分析直接返回缓存结果。4.5 常见问题速查表问题现象可能原因排查方向解决方案流程卡在某节点不动接口超时无超时设置检查网络请求加 timeout 和重试JSON 解析失败模型输出格式漂移打印原始输出三层解析防护分析结果为空数据源返回空值检查 data_status降级处理结论自相矛盾数据时间戳不一致核对各数据采集时间统一时间基准成本过高重复调用大模型统计 token 消耗加缓存和分级调用技术指标失真未使用复权价检查价格数据统一前复权5. 从 TradingAgents 延伸出的多智能体落地经验5.1 智能体粒度怎么把握做多智能体系统最容易纠结的就是到底该拆几个 Agent。拆太少跟单智能体没区别拆太多协调成本高还容易出现三个和尚没水喝的情况。我的经验是按决策维度拆而不是按数据源拆。什么意思比如基本面是一个决策维度它可能涉及财报、估值、行业对比多个数据源但这些数据最终服务于基本面判断这一个决策所以归一个 Agent 管。而技术面和舆情面是独立的决策维度各自成 Agent。这样拆出来的 Agent 数量通常在 5 到 8 个之间既能覆盖主要维度又不至于太碎。TradingAgents 的拆分就符合这个原则基本面、技术面、舆情是三个决策维度多空辩论是两个立场维度风控和决策是流程维度。每个 Agent 都有明确的决策产出而不是单纯的数据搬运工。5.2 状态设计的可扩展性前面讲的 State 对象设计时要考虑未来扩展。我见过一些项目State 字段写死了后来想加一个新维度比如资金流向分析发现要改的地方太多牵一发动全身。我的做法是给 State 留一个extra_analysis字典字段任何新增的分析维度都可以往里面塞不影响已有字段。同时决策 Agent 的加权逻辑用配置化的方式实现权重存在配置文件里加新维度时只改配置不改代码。# 权重配置化 DIMENSION_WEIGHTS { fundamental: 0.30, technical: 0.25, news: 0.15, debate: 0.20, risk: 0.10, # 新增维度只需在这里加一行 capital_flow: 0.0, # 初始权重为0验证后再调整 }5.3 提示词工程在多智能体里的特殊考量单智能体的提示词工程核心是把任务说清楚。多智能体的提示词工程多了一层角色隔离的要求。每个 Agent 的提示词里必须明确它的角色边界防止它越界去干别的 Agent 的活。比如技术面 Agent 的提示词里我会明确写你只负责技术面分析不要评论公司基本面不要预测消息面影响这些由其他分析师负责。这样能保证各 Agent 输出的纯粹性避免重复劳动和结论冲突。另一个技巧是给每个 Agent 设定输出格式模板。多智能体系统里Agent 之间的数据传递是结构化的如果上游 Agent 输出格式不规范下游 Agent 解析就会出问题。所以每个 Agent 的提示词里都要带上明确的输出格式要求最好给出完整的示例。5.4 系统评估与迭代方法多智能体系统上线后怎么评估它好不好用不能只看预测准不准因为股票预测本身就不可能准。我用的评估维度有三个。第一是流程完整性每次分析是否所有节点都跑通了有没有降级或失败。这个指标反映系统稳定性。第二是结论一致性同一只股票在相近时间点分析结论是否稳定如果今天说买入明天说卖出说明系统不稳定。第三是人工复核通过率把系统输出给有经验的投资者看看他们觉得分析逻辑是否合理这个指标反映分析质量。迭代时优先解决流程完整性问题因为流程都跑不通谈分析质量没意义。流程稳定后再优化提示词提升分析质量最后再考虑加新维度、调权重这些精细化操作。5.5 合规与风险提示的边界做股票分析类 AI 应用有一条红线必须守住不能给出明确的买卖指令。系统可以输出综合评分 75 分偏乐观但不能输出建议买入目标价 XX 元。前者是分析参考后者是投资建议性质完全不同。我的做法是在最终报告里加一段固定声明明确说明本分析仅供研究参考不构成投资建议据此操作风险自负。同时决策 Agent 的输出措辞要克制用偏乐观/中性/偏谨慎这种程度词而不是买入/卖出这种指令词。这不是为了规避责任而是对用户负责——AI 的分析再全面也无法替代投资者自己的判断和风险承受能力评估。6. 我在实际搭建这类系统时踩过的坑说几个只有真正动手做过才会遇到的坑。第一个是时区问题。A 股数据的时间戳有时候是北京时间有时候是 UTC如果不统一会导致用未来数据做分析这种低级错误。我现在的做法是所有时间戳进系统就统一转成北京时间并且在 State 里明确标注。第二个是停牌股票的处理。停牌期间没有行情数据技术面 Agent 会拿到空数据如果不做特殊处理要么报错要么输出垃圾结论。正确做法是检测到停牌就跳过技术面分析在报告里注明该股票当前停牌。第三个是新股数据不足。上市不满一年的股票历史数据太少技术指标算不出来财务数据也只有一两个季度。这种情况要主动降级只做有限的基本面分析并提示用户数据不足分析可靠性降低。第四个是大模型的知识截止问题。模型训练数据有截止日期对最近发生的事件可能不了解。所以舆情分析不能只靠模型知道必须把实时新闻喂进去让模型基于给定文本分析而不是靠它自己的记忆。这些坑文档里不会写只有自己跑一遍才会遇到。我建议你在搭建类似系统时先把这些边界情况列个清单逐个处理能省下大量调试时间。最后分享一个实用技巧给系统加一个分析日志功能把每个 Agent 的输入、输出、耗时、token 消耗都记录下来。这个日志在调试时是救命稻草能让你快速定位是哪个环节出了问题。而且积累一段时间后你还能分析出哪些 Agent 最耗时、哪些提示词效果最好为后续优化提供数据支撑。