LangGraph 02：路由、循环、并发与 Command

2026-09-01

983 字约 4 分钟

...

1、本篇任务：让控制流明确、有限且可预测

上一课的图会在输入非法时继续查询。本课增加四类控制：条件路由决定走哪条边；循环允许有限重试；并发加速多个独立检索；Command 将状态更新与跳转绑定。

2、条件路由用于确定性业务判断

from typing import Literal

def route_after_validate(state: RefundState) -> Literal["load", "reject"]:
    return "reject" if state.get("errors") else "load"

builder.add_conditional_edges(
    "validate",
    route_after_validate,
    {"load": "load_order", "reject": "abstain"},
)

权限、金额、订单状态、重试次数等确定性条件用代码。模型只在“用户意图是什么、证据如何概括”这类语义任务中提供结构化枚举，路由函数再根据枚举跳转。

3、循环必须先写退出条件

检索不足时可以改写 query 再检索，但循环至少有四个出口：证据足够；达到最大尝试；deadline/token 预算耗尽；遇到不可重试错误。

def after_grade(state) -> Literal["answer", "rewrite", "abstain"]:
    if state["evidence_score"] >= 0.8:
        return "answer"
    if state["attempt"] >= 2 or state["budget_remaining"] <= 0:
        return "abstain"
    return "rewrite"

recursion limit 是最后保险，不是业务设计。若触发它，说明状态机退出条件有问题，应记录当前 step 和 state 摘要。

4、Command 绑定更新和跳转

当一个节点既要修改 state 又要决定目的地时，用 Command 可避免边读取旧值：

from langgraph.types import Command

def grade(state) -> Command:
    score = grade_evidence(state["evidence"])
    if score >= 0.8:
        return Command(update={"evidence_score": score}, goto="answer")
    if state["attempt"] >= 2:
        return Command(update={"evidence_score": score}, goto="abstain")
    return Command(update={"evidence_score": score, "attempt": state["attempt"] + 1}, goto="rewrite")

TypeScript 对应 new Command({ update: {...}, goto: "answer" })。节点返回类型最好列出所有可能目的地，帮助类型检查和代码审阅。

5、静态并发与 fan-in

订单和政策相互独立，可从同一节点分别连边并行执行；它们都结束后，汇总节点进入下一 super-step。

prepare ─┬→ load_order ───┐
         ├→ search_policy ├→ merge → grade
         └→ search_faq ───┘

并行分支不要共同覆盖单值字段。各分支写自己的字段，或写带 reducer 的 evidence；最终由 merge/grade 单独写 decision。

6、动态 fan-out 与 Send

来源数量在运行时才知道时，用 Send 创建工作单元：

from langgraph.types import Send

def create_research_jobs(state):
    return [
        Send("research_one", {"source_id": source_id, "question": state["question"]})
        for source_id in state["candidate_sources"][:5]
    ]

def research_one(state):
    return {"evidence": [lookup(state["source_id"], state["question"])]}