跳到主要内容

第 1 章 · LangGraph 编排与状态机

本章讲整个流程怎么被组装成一台状态机、又怎么跑起来。读完你能在脑子里画出完整的图，并说清“辩论为什么会停、崩溃为什么不会连累全图”。

1.1 一句话：它是一台共享状态的接力机​

TradingAgents 的“图”是 LangGraph 的 StateGraph。核心只有三样东西：

  1. 一个共享状态 AgentState——一个大 TypedDict，装着报告、辩论历史、最终决策等所有字段。
  2. 一堆节点——每个 agent 是一个节点函数：读 state 里它要的字段，返回一个 dict 去更新 state。
  3. 一堆边——决定节点之间怎么走；有的是固定边，有的是条件边(看 state 决定去哪)。

组装发生在 GraphSetup.setup_graph(graph/setup.py:61)，编译和运行发生在 TradingAgentsGraph(graph/trading_graph.py)。

1.2 共享状态 AgentState 长什么样​

AgentState 继承 LangGraph 的 MessagesState(自带 messages 列表)，再挂上业务字段(agent_states.py:47)：

字段谁写干什么
company_of_interest / trade_date / asset_type初始化本次分析的标的、日期、资产类型
instrument_context运行开始确定性解析出的标的身份(公司名/行业)，防幻觉的关键
market_report / sentiment_report / news_report / fundamentals_report四个分析师四份分析报告
investment_debate_state多空研究员 + 研究经理多空辩论的全部历史与裁决(见下)
investment_plan研究经理给交易员的投资计划
trader_investment_plan交易员可执行的交易提案
risk_debate_state风险三方 + 组合经理风险辩论历史与裁决
final_trade_decision组合经理最终决策(整段文字)
past_context运行开始从记忆日志注入的历史教训

两个辩论子状态是嵌套 TypedDict，关键是各自带一个 count(轮数计数器)和 current_response/latest_speaker(谁刚说完)——这两样就是辩论循环的“节拍器”(agent_states.py:8-44)。

1.3 图的骨架：节点和边怎么连​

setup_graph 的组装顺序(graph/setup.py:95-154)：

START
  │  (固定边)
  ▼
[分析师 1]──条件边──▶ tools_X ──固定边──▶ 回到[分析师 1]   ← 取数循环
  │  (要么进 tools_X，要么进 Msg Clear X)
  ▼ Msg Clear X (清空消息)
[分析师 2] … 同样的结构 … [分析师 N]
  │  最后一个分析师 clear 后固定边
  ▼
[Bull Researcher]──条件边(should_continue_debate)
  ├─▶ Bear Researcher ──条件边──┐
  │        ▲                     │
  └────────┴─────────────────────┘   ← 多空辩论循环
  │  count 到顶 → 
  ▼
[Research Manager] ──固定边──▶ [Trader] ──固定边──▶ [Aggressive Analyst]
                                                        │ 条件边(should_continue_risk_analysis)
     Aggressive → Conservative → Neutral 轮转 ◀─────────┘   ← 风险辩论循环
  │  count 到顶 →
  ▼
[Portfolio Manager] ──固定边──▶ END

分析师节点是按 selected_analysts 动态生成的：每个分析师配三件套——agent 节点、清理节点、工具节点(setup.py:98-101)。这三件套的名字由 build_analyst_execution_plan 统一分配(见 1.6)。

1.4 机制一：分析师的“取数循环”​

它要解决的小问题: 分析师不能一上来就写报告——它得先决定要哪些指标、调工具拉数据，可能来回好几次。

思路: 用一条条件边在“继续调工具”和“收工清理”之间二选一。判据极简：LLM 这轮的回复里有没有 tool_calls。

真实实现(conditional_logic.py:14-20，市场分析师为例)：

def should_continue_market(self, state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:      # 模型还想要数据
        return "tools_market"        # → 去工具节点执行，然后回到分析师
    return "Msg Clear Market"        # 模型不再要数据 → 收工，进清理节点

工具节点执行完用固定边回到分析师(setup.py:129)，于是形成 分析师 ⇄ 工具 的循环，直到模型不再要数据、写出最终报告。四个分析师各有一个 should_continue_<key> 方法，结构完全一样。

清理节点为什么必要: 每个分析师收工后进一个 create_msg_delete() 节点，把 messages 全清掉再塞一句锚定标的的占位消息。这样下一个分析师不会被上一个的工具消息污染上下文。占位消息为什么不能是裸 "Continue"——是个踩过的坑，见第 2 章。

1.5 机制二：两个辩论循环怎么“数着轮数”停​

它要解决的小问题: 多空/风险辩论不能无限来回，得有个明确的停止条件，还得决定“下一个该谁说”。

思路: 用 count 计数 + 判断“上一个说话的是谁”来轮转。两处逻辑同构。

多空辩论(conditional_logic.py:52-61)：

def should_continue_debate(self, state):
    if state["investment_debate_state"]["count"] >= 2 * self.max_debate_rounds:
        return "Research Manager"                    # 轮数到顶 → 裁决
    if state["investment_debate_state"]["current_response"].startswith("Bull"):
        return "Bear Researcher"                     # 刚说完的是多头 → 换空头
    return "Bull Researcher"                         # 否则换多头

每个研究员节点跑完会把自己的 count 加 1、并把 current_response 标上 "Bull Analyst: …" 之类前缀。默认 max_debate_rounds=1，阈值 2*1=2：多头(count→1)、空头(count→2)各说一次就到顶，进研究经理。

一个要诚实指出的细节: 代码旁的注释写的是 “3 rounds of back-and-forth”，但按 count >= 2 * max_debate_rounds 实际是每轮 2 次发言(多+空各一)。注释与代码不一致，以代码为准(conditional_logic.py:56 附近)。

风险辩论(conditional_logic.py:63-73)阈值是 3 * max_risk_discuss_rounds，因为是三方轮转：

Aggressive(count→1) → Conservative(count→2) → Neutral(count→3) → 到顶 → Portfolio Manager

判据用 latest_speaker.startswith(...) 决定下一位：激进后接保守、保守后接中性、否则回激进。

1.6 机制三：路由边全挂 path_map，防止 fall-through 崩图​

它要解决的小问题: LangGraph 的条件边需要一张 path_map(路由函数的返回值 → 目标节点)。如果路由函数因为 prompt/i18n/重构漂移，返回了一个不在 map 里的字符串，LangGraph 会运行中途崩溃。

巧妙做法: 把每条辩论边都挂上完整的 path_map，覆盖该路由器能返回的所有值——即便某条边逻辑上“只会去某几个地方”，也把全集给它。这样 fall-through 也一定命中某个键，不会崩(setup.py:32-42、setup.py:137-152，issue #1088)：

DEBATE_PATH_MAP = {
    "Bull Researcher": "Bull Researcher",
    "Bear Researcher": "Bear Researcher",
    "Research Manager": "Research Manager",
}
# 多头、空头两条边都用同一张完整 map
for debate_node in ("Bull Researcher", "Bear Researcher"):
    workflow.add_conditional_edges(debate_node, should_continue_debate, DEBATE_PATH_MAP)

分析师节点的动态命名也服务于同一目标——AnalystNodeSpec 把 agent_node/clear_node/tool_node/report_key 四个名字集中定义(analyst_execution.py:20-53)，路由函数返回的标签和图里注册的节点名同源，不会对不上。注意 social 这个 wire key 的用户可见名是 “Sentiment Analyst”(v0.2.5 改名，为兼容旧配置保留 key)。

1.7 跑图：propagate() 做了哪些事​

入口 TradingAgentsGraph.propagate(trading_graph.py:362)不止“跑一遍图”，它按顺序做了 5 件事：

propagate(ticker, date, asset_type)
  1. _resolve_pending_entries(ticker)   ← 先把上次的 pending 决策用真实行情结算(见第4章)
  2. 若 checkpoint_enabled: 用 per-ticker SqliteSaver 重编译图，算出可续跑的 step
  3. _run_graph():
       - 组装初始 state: 注入 past_context(历史教训) + instrument_context(标的身份)
       - debug 模式 stream 逐节点打印；否则 graph.invoke 一把跑完
       - 把最终 state 落盘 JSON、把决策记为 pending(store_decision)
       - 成功后清掉 checkpoint
  4. return final_state, process_signal(final_decision)   ← 从决策里抽出五档评级

process_signal 不再额外调 LLM——因为组合经理用结构化输出保证了决策里一定有 **Rating**: X，用确定性正则 parse_rating 抽就够(signal_processing.py:29、rating.py:28)。

初始状态和跑图参数由 Propagator 提供(propagation.py:18、propagation.py:71)，其中 recursion_limit(默认 100)是 LangGraph 的安全阀，防止某个循环失控无限跑。

1.8 代码地图(本章)​

主题文件符号
图组装tradingagents/graph/setup.pyGraphSetup.setup_graph、DEBATE_PATH_MAP、RISK_ANALYSIS_PATH_MAP
条件路由tradingagents/graph/conditional_logic.pyshould_continue_market、should_continue_debate、should_continue_risk_analysis
共享状态tradingagents/agents/utils/agent_states.pyAgentState、InvestDebateState、RiskDebateState
节点命名tradingagents/graph/analyst_execution.pyAnalystNodeSpec、build_analyst_execution_plan
总编排/跑图tradingagents/graph/trading_graph.pyTradingAgentsGraph.propagate、_run_graph
初始状态tradingagents/graph/propagation.pyPropagator.create_initial_state、get_graph_args