LangGraph Command：现代路由方式

1.1 conditional_edges 的局限

LangGraph Command 对比条件边——Command 对象将路由与状态更新合一，代码更简洁

LangGraph 最初的路由方式是 add_conditional_edges：在节点外部定义一个路由函数，根据状态判断下一个节点。

python

def route_after_analysis(state: AgentState) -> str:
    if state["needs_retrieval"]:
        return "retrieval_node"
    elif state["needs_calculation"]:
        return "calculation_node"
    else:
        return "answer_node"

builder.add_conditional_edges(
    "analysis_node",
    route_after_analysis,
    {
        "retrieval_node": "retrieval_node",
        "calculation_node": "calculation_node",
        "answer_node": "answer_node",
    }
)

这种方式在图规模较小时尚可接受，但随着节点增多，问题逐渐暴露：

路由逻辑与节点逻辑分离。 节点函数执行完毕，开发者需要跳到另一个地方查看路由函数，才能知道接下来会发生什么。节点的"意图"和"去向"分散在两处代码中，阅读体验割裂。

更新状态与路由必须分两步完成。 节点函数返回状态更新，路由函数返回下一节点，两者不能在同一处完成，带来了不必要的间接层。

多目标跳转复杂。 想从一个节点同时派发到多个节点，需要结合 Send API，写法不够直观。

Command 对象正是为解决这些问题而设计的。

1.2 Command 对象：让节点声明"去哪里"

Command 允许节点函数直接返回路由指令和状态更新，二者合一：

python

from langgraph.types import Command

def analysis_node(state: AgentState) -> Command:
    result = perform_analysis(state["query"])

    if result["needs_retrieval"]:
        return Command(
            goto="retrieval_node",
            update={"analysis_result": result, "needs_retrieval": True}
        )
    else:
        return Command(
            goto="answer_node",
            update={"analysis_result": result}
        )

节点函数现在同时完成两件事：更新状态（update）和声明下一步（goto）。阅读节点函数就能完整了解这个节点的行为。

Command 的基本参数：

python

Command(
    goto="target_node",        # 目标节点名称（必填）
    update={"field": value},   # 状态更新（可选）
    graph=Command.PARENT,      # 目标图（用于子图场景，可选）
)

1.3 Command vs conditional_edges 对比

对比维度	conditional_edges	Command
路由逻辑位置	独立的路由函数，与节点分离	在节点函数内部，与逻辑合并
可读性	需要跳转查看路由函数	节点内部直接可见
状态更新与路由	分两步完成	同一个 return 语句
多目标跳转	需结合 Send，写法繁琐	`goto=[Send(...), "node_name"]`
图结构声明	必须在 add_conditional_edges 中列出所有目标	无需预先声明，动态决定
调试体验	路由逻辑在外部，断点设置分散	断点直接设在节点函数内
子图通信	不支持跨图路由	支持 `graph=Command.PARENT`
适用场景	路由条件简单，目标固定	复杂条件、动态目标、需要清晰代码组织

1.4 完整示例：动态路由

以下示例展示一个意图识别 → 分类处理的图，不同意图走不同节点：

python

from typing import TypedDict
from langgraph.types import Command
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json

class RouterState(TypedDict):
    user_input: str
    intent: str
    result: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def intent_classifier(state: RouterState) -> Command:
    """识别用户意图，直接路由到对应节点"""
    response = llm.invoke([
        HumanMessage(content=f"""将以下用户输入分类为：search（搜索信息）、calculate（计算任务）、chat（闲聊）
用户输入：{state['user_input']}
只输出分类结果，不加任何解释。""")
    ])

    intent = response.content.strip().lower()
    # 保证路由到有效节点
    valid_intents = {"search", "calculate", "chat"}
    if intent not in valid_intents:
        intent = "chat"

    return Command(
        goto=f"handle_{intent}",
        update={"intent": intent}
    )

def handle_search(state: RouterState) -> Command:
    response = llm.invoke([
        HumanMessage(content=f"请搜索并回答：{state['user_input']}")
    ])
    return Command(goto=END, update={"result": response.content})

def handle_calculate(state: RouterState) -> Command:
    response = llm.invoke([
        HumanMessage(content=f"请计算：{state['user_input']}，只输出计算结果和简要过程。")
    ])
    return Command(goto=END, update={"result": response.content})

def handle_chat(state: RouterState) -> Command:
    response = llm.invoke([
        HumanMessage(content=state["user_input"])
    ])
    return Command(goto=END, update={"result": response.content})

# 构建图：使用 Command 后，无需 add_conditional_edges
builder = StateGraph(RouterState)
builder.add_node("intent_classifier", intent_classifier)
builder.add_node("handle_search", handle_search)
builder.add_node("handle_calculate", handle_calculate)
builder.add_node("handle_chat", handle_chat)

builder.add_edge(START, "intent_classifier")
# 不需要 add_conditional_edges！Command 内部已经处理了路由

graph = builder.compile()

result = graph.invoke({"user_input": "北京今天的天气怎么样？", "intent": "", "result": ""})
print(f"意图：{result['intent']}")
print(f"回答：{result['result']}")

1.5 Command + interrupt：Human-in-the-loop 的新写法

interrupt 是 LangGraph 暂停执行、等待人工输入的机制。结合 Command，可以写出清晰的人工审核流程：

python

from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class ReviewState(TypedDict):
    topic: str           # 文章主题（初始输入）
    draft_content: str
    human_feedback: str
    approved: bool
    final_content: str

def generate_draft(state: ReviewState) -> Command:
    """生成草稿内容"""
    draft = f"这是一份关于'{state['topic']}' 的草稿内容..."
    return Command(
        goto="human_review",
        update={"draft_content": draft}
    )

def human_review(state: ReviewState) -> Command:
    """暂停执行，等待人工审核"""
    # interrupt 会暂停图的执行
    # 恢复时，人工传入的数据作为 interrupt 的返回值
    human_input = interrupt({
        "draft": state["draft_content"],
        "instruction": "请审核以上草稿，输入 'approve' 批准，或输入修改意见"
    })

    if human_input.lower() == "approve":
        return Command(
            goto="finalize",
            update={"approved": True, "human_feedback": "approved"}
        )
    else:
        return Command(
            goto="revise_draft",
            update={"approved": False, "human_feedback": human_input}
        )

def revise_draft(state: ReviewState) -> Command:
    """根据人工反馈修改草稿"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke([
        HumanMessage(content=f"""根据反馈修改以下内容：

原稿：{state['draft_content']}
修改意见：{state['human_feedback']}

请输出修改后的内容：""")
    ])
    return Command(
        goto="human_review",  # 返回审核节点，形成审核循环
        update={"draft_content": response.content}
    )

def finalize(state: ReviewState) -> dict:
    return {"final_content": state["draft_content"]}

# 构建带持久化的图（interrupt 需要 checkpointer）
builder = StateGraph(ReviewState)
builder.add_node("generate_draft", generate_draft)
builder.add_node("human_review", human_review)
builder.add_node("revise_draft", revise_draft)
builder.add_node("finalize", finalize)

builder.add_edge(START, "generate_draft")
builder.add_edge("finalize", END)

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "review_001"}}

# 第一次执行：会在 human_review 暂停
result = graph.invoke(
    {"topic": "AI 应用", "draft_content": "", "human_feedback": "", "approved": False, "final_content": ""},
    config=config,
)
# result 是 interrupt 的暂停状态

# 人工审核后，用 Command 恢复执行
from langgraph.types import Command as ResumeCommand
graph.invoke(ResumeCommand(resume="请增加更多代码示例"), config=config)

1.6 多目标跳转：结合 Send

Command 的 goto 参数可以接受 Send 对象列表，实现从单个节点同时派发到多个目标：

python

from langgraph.types import Command, Send
from typing import Annotated
import operator

class MultiTargetState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]
    summary: str
    status: str  # 任务状态字段

def dispatch_and_monitor(state: MultiTargetState) -> Command:
    """同时启动批处理任务和监控任务"""
    sends = [Send("process_item", {"item": item, "results": []}) for item in state["items"]]
    # 同时发送到多个处理节点
    return Command(
        goto=sends,
        update={"status": "processing"}
    )

1.7 完整重构：用 Command 改写 Supervisor 模式

Supervisor 模式（主管协调多个 Worker）是多智能体系统的经典架构。传统写法依赖 conditional_edges，以下展示用 Command 改写后的清晰版本：

python

from typing import TypedDict, Literal, Annotated
import operator
from langgraph.types import Command
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import json

class SupervisorState(TypedDict):
    task: str
    worker_outputs: Annotated[list[str], operator.add]
    current_worker: str
    iteration: int
    final_answer: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

WORKERS = ["researcher", "coder", "writer"]

def supervisor(state: SupervisorState) -> Command:
    """主管节点：决定派发给哪个 Worker，或者结束任务"""
    iteration = state.get("iteration", 0)

    # 防止无限循环
    if iteration >= 5:
        return Command(goto="final_aggregator", update={"iteration": iteration})

    context = "\n".join(state.get("worker_outputs", []))
    response = llm.invoke([
        SystemMessage(content="""你是一个任务协调者。根据任务和已有输出，决定下一步：
- 如果需要研究信息，回复：{"next": "researcher"}
- 如果需要编写代码，回复：{"next": "coder"}
- 如果需要撰写文档，回复：{"next": "writer"}
- 如果任务已完成，回复：{"next": "FINISH"}
只输出 JSON，不要其他内容。"""),
        HumanMessage(content=f"""任务：{state['task']}
已完成的工作：
{context if context else '（暂无）'}
下一步应该做什么？""")
    ])

    try:
        decision = json.loads(response.content)
        next_node = decision.get("next", "FINISH")
    except json.JSONDecodeError:
        next_node = "FINISH"

    if next_node == "FINISH":
        return Command(
            goto="final_aggregator",
            update={"iteration": iteration + 1}
        )
    else:
        return Command(
            goto=next_node,
            update={"current_worker": next_node, "iteration": iteration + 1}
        )

def researcher(state: SupervisorState) -> Command:
    """研究员 Worker"""
    response = llm.invoke([
        HumanMessage(content=f"请研究并提供关于以下任务的背景信息：{state['task']}")
    ])
    return Command(
        goto="supervisor",  # 完成后返回主管
        update={"worker_outputs": [f"[研究员] {response.content}"]}
    )

def coder(state: SupervisorState) -> Command:
    """程序员 Worker"""
    response = llm.invoke([
        HumanMessage(content=f"请为以下任务提供代码实现：{state['task']}")
    ])
    return Command(
        goto="supervisor",
        update={"worker_outputs": [f"[程序员] {response.content}"]}
    )

def writer(state: SupervisorState) -> Command:
    """文档撰写 Worker"""
    response = llm.invoke([
        HumanMessage(content=f"请撰写关于以下任务的说明文档：{state['task']}")
    ])
    return Command(
        goto="supervisor",
        update={"worker_outputs": [f"[文档] {response.content}"]}
    )

def final_aggregator(state: SupervisorState) -> dict:
    """汇总所有 Worker 的输出"""
    all_outputs = "\n\n".join(state["worker_outputs"])
    response = llm.invoke([
        HumanMessage(content=f"请将以下工作成果整合为最终答案：\n\n{all_outputs}")
    ])
    return {"final_answer": response.content}

# 构建图：Command 负责路由，无需任何 add_conditional_edges
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_node("writer", writer)
builder.add_node("final_aggregator", final_aggregator)

builder.add_edge(START, "supervisor")
builder.add_edge("final_aggregator", END)
# 注意：researcher/coder/writer 到 supervisor 的边由 Command(goto="supervisor") 处理
# 无需在这里手动添加

graph = builder.compile()

result = graph.invoke({
    "task": "实现一个 Python 函数，用于计算斐波那契数列",
    "worker_outputs": [],
    "current_worker": "",
    "iteration": 0,
    "final_answer": "",
})
print(result["final_answer"])

1.7.1 Command 路由流程图

1.8 使用 Command 的注意事项

节点返回类型变化。 使用 Command 时，节点函数返回 Command 对象而不是普通字典。如果节点同时返回状态更新，更新内容放在 Command 的 update 参数中；如果没有状态更新，update 可以省略。

不需要在 add_edge 中重复声明跳转。 Command(goto="node_x") 已经完整表达了路由意图，不需要再用 add_edge 声明。但 START 的出边仍然需要 add_edge 或 add_conditional_edges 声明。

子图通信使用 graph=Command.PARENT。 当子图需要将控制权交回父图时：

python

def subgraph_exit_node(state) -> Command:
    return Command(
        goto="parent_node",
        update={"result": "子图完成"},
        graph=Command.PARENT,  # 跳转目标在父图中
    )

Command 与普通返回可以混用。 同一个图中，部分节点使用 Command，部分节点返回普通字典，完全兼容。逐步迁移是可行的。