最近在学习 LangGraph 的过程中,我发现很多初学者(包括我自己)都会被 Send 和 Command 这两个概念搞混。它们看起来都能让节点跳来跳去,到底有什么区别?今天我用最直白的大白话 + 代码 + 生活类比,帮你彻底理清。
一、先搞清楚:它们是为了解决什么问题?
在普通的 StateGraph 中,节点的执行流程是固定的:
节点A → 边 → 节点B → 边 → 节点C
但很多时候,我们需要在运行时动态决定下一步去哪、给谁发任务、甚至并行执行多个任务。Send 和 Command 就是解决这种"动态调度"问题的两把钥匙。
二、Send:像一个"快递包裹"
2.1 通俗理解
想象你是一家公司的调度员。早上老板给你一张清单,上面写着 10 个客户的地址。你的工作是:给每个客户派一个快递员去送货。
Send 就是你手里的"快递单",上面写着:
- 收件人(node):哪个快递员去送
- 包裹内容(arg):送什么东西
你可以一次开出很多张快递单, LangGraph 会自动并行派出多个快递员。
2.2 代码写法
from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
topics: list[str] # 比如 ["猫", "狗", "企鹅"]
jokes: list[str]
# 调度员:给每个 topic 都派一个任务
def dispatch_jokes(state: State):
# 返回 3 个 Send,LangGraph 会并行执行 3 次 generate_joke
return [
Send("generate_joke", {"topic": topic})
for topic in state["topics"]
]
def generate_joke(state: dict):
# state 这里接收到的是 {"topic": "猫"},不是主 State!
return {"jokes": [f"关于{state['topic']}的笑话"]}
builder = StateGraph(State)
builder.add_node("generate_joke", generate_joke)
builder.add_conditional_edges(START, dispatch_jokes) # ← Send 用在边里
builder.add_edge("generate_joke", END)
graph = builder.compile()
result = graph.invoke({"topics": ["猫", "狗", "企鹅"], "jokes": []})
print(result)
# {'topics': ['猫', '狗', '企鹅'], 'jokes': ['关于猫的笑话', '关于狗的笑话', '关于企鹅的笑话']}
2.3 Send 的特点
| 用在哪 | 只能用在条件边(add_conditional_edges)里 |
| 能干什么 | 把数据发给指定节点,支持并行 |
| 传什么 | Send("节点名", {"自定义": "状态"}),第二个参数可以是任意 dict |
| 更新主状态 | ❌ 不能直接更新图的主 State |
| 经典场景 | Map-Reduce:1 个输入拆成 N 个并行任务,最后汇总 |
三、Command:像一个"遥控器"
3.1 通俗理解
Command 比 Send 更高级。如果说 Send 是一张"快递单",那 Command 就是一个万能遥控器。
这个遥控器上有很多按钮:
- update(更新):修改当前房间(图状态)的设置
- goto(跳转):直接跳到某个频道(节点)
- resume(恢复):从暂停状态继续播放(配合 interrupt 人机交互)
而且最厉害的是:这个遥控器可以由房间里的任何人(节点)直接按!
3.2 代码写法
from langgraph.types import Command
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal
class State(TypedDict):
question: str
category: str
answer: str
# 分类员节点:一边更新状态,一边决定下一步去哪
def classify(state: State) –> Command[Literal["handle_order", "handle_refund", "handle_general"]]:
question = state["question"]
if "退款" in question:
category = "refund"
next_node = "handle_refund"
elif "订单" in question:
category = "order"
next_node = "handle_order"
else:
category = "general"
next_node = "handle_general"
# 一个 Command,同时做到:
# 1. update:把 category 写进 State
# 2. goto:告诉 LangGraph 下一步去哪个节点
return Command(update={"category": category}, goto=next_node)
def handle_order(state: State):
return {"answer": f"【订单问题】已收到,分类为:{state['category']}"}
def handle_refund(state: State):
return {"answer": f"【退款问题】已收到,分类为:{state['category']}"}
def handle_general(state: State):
return {"answer": f"【一般问题】已收到,分类为:{state['category']}"}
builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("handle_order", handle_order)
builder.add_node("handle_refund", handle_refund)
builder.add_node("handle_general", handle_general)
builder.add_edge(START, "classify")
# 注意:这里不需要 add_conditional_edges!
# 因为 classify 节点自己用 Command 决定了下一步去哪
graph = builder.compile()
result = graph.invoke({"question": "我要退款!", "category": "", "answer": ""})
print(result)
# {'question': '我要退款!', 'category': 'refund', 'answer': '【退款问题】已收到,分类为:refund'}
3.3 Command 的特点
| 用在哪 | 在节点函数里直接 return |
| 能干什么 | 更新状态 + 路由跳转 + 中断恢复,三合一 |
| 更新主状态 | ✅ update={"key": "value"} 直接更新 State |
| 决定下一步 | ✅ goto="节点名" 直接决定下一个节点 |
| 配合中断 | ✅ resume="用户输入" 恢复 interrupt |
| 跨图控制 | ✅ Command(graph=Command.PARENT) 控制父图 |
| 支持 Send | ✅ goto=Send("节点", {"arg": 1}) 可以包裹 Send |
四、Send vs Command:一张图看懂
| 角色 | 调度员开的单子 | 节点自己按的按钮 |
| 使用位置 | 条件边函数里 | 节点函数内部 |
| 更新 State | ❌ 不行 | ✅ update 参数 |
| 控制路由 | ✅ 只能决定去哪个节点 | ✅ goto 决定去哪个节点 |
| 并行派发 | ✅ 返回列表 [Send(…), Send(…)] | ✅ goto=[Send(…), Send(…)] |
| 人机交互 | ❌ 不支持 | ✅ resume 恢复中断 |
| 跨图调用 | ❌ 不支持 | ✅ Command.PARENT |
| 灵活程度 | 单一功能 | 多功能合一 |
| 推荐度 | 特定场景使用 | 优先推荐 |
五、用一个例子看它们的"进化关系"
假设我们要做一个"智能客服分流"系统:
版本 1:最原始(不用 Send/Command)
def classify(state: State):
if "退款" in state["question"]:
return {"category": "refund"}
return {"category": "order"}
# 还要单独写一个条件边函数
def route(state: State):
if state["category"] == "refund":
return "handle_refund"
return "handle_order"
builder.add_node("classify", classify)
builder.add_conditional_edges("classify", route) # 边和节点分离,啰嗦
缺点:分类逻辑和路由逻辑拆成两半,维护麻烦。
版本 2:用 Command(推荐 ✅)
def classify(state: State) –> Command[Literal["handle_refund", "handle_order"]]:
if "退款" in state["question"]:
return Command(update={"category": "refund"}, goto="handle_refund")
return Command(update={"category": "order"}, goto="handle_order")
builder.add_node("classify", classify)
# 不需要 add_conditional_edges!节点自己说了算
优点:更新状态和路由决策写在一起,一眼就能看懂整个逻辑。
版本 3:如果需要并行处理多个子任务
def classify(state: State) –> Command:
# 把问题拆成 3 个子问题,并行发送给 3 个处理节点
sub_tasks = [
Send("handler", {"sub_question": q})
for q in state["sub_questions"]
]
return Command(goto=sub_tasks)
这里:Command 包裹了 Send,既有 Command 的灵活性,又有 Send 的并行能力。
六、总结:我该怎么选?
你的需求
│
├─ 只需要在条件边里做简单的动态路由?
│ └─ 用 Send(传统方式)
│
├─ 需要在节点里同时"更新状态 + 决定下一步"?
│ └─ 用 Command(推荐!)
│
├─ 需要做 Map-Reduce 并行任务?
│ └─ 在 Command.goto 里包 Send,或者条件边里用 Send
│
└─ 需要人机交互(interrupt + 恢复)?
└─ 必须用 Command(resume 参数)
一句话记住区别:
Send 是调度中心开出来的"任务单",Command 是节点自己手里的"遥控器"。
有遥控器了,谁还愿意跑调度中心填单子呢?优先学 Command,遇到并行 Map-Reduce 时把 Send 包进去。
七、常见坑
Send 不能直接在节点里 return:旧版本 LangGraph 中,节点里 return Send 是不生效的,它只能用在 add_conditional_edges 的函数里。新版虽然有所放宽,但最佳实践仍然是:节点里用 Command,边里用 Send。
Command 的 update 和传统 return 不能混:一个节点要么 return {"key": "val"},要么 return Command(update={"key": "val"}),不要搞混。
用了 Command 就不需要那条边了:如果你的节点 return 了 Command(goto=…),那么从该节点出发的 add_edge 或 add_conditional_edges 会被覆盖,可以省略。
希望这篇博客能帮到你!如果觉得有用,欢迎点赞收藏,有问题评论区交流~
网硕互联帮助中心




评论前必须登录!
注册