LangGraph 05：Send 与 Command

发表于 2026-08-19 更新于 2026-08-20 分类于 ai

用 Send 做动态扇出、Command 合并状态更新与跳转，并演示扇入汇合与 MapReduce 词频统计。

前言

固定边只能表达「谁连谁」；真正复杂的图往往要在运行时决定「派生几个任务、跳到哪个节点」。
LangGraph 1.x 用 Send 做动态扇出，用 Command 把状态更新与 goto 跳转写在同一返回值里。
本文覆盖动态 Send、Command(update, goto)、扇入汇合，以及 MapReduce 词频示例。
聊天模型对接 火山方舟 Coding Plan（ark-code-latest）；MapReduce 与扇入示例可不调 LLM。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明依赖（版本请按项目实际调整）。

uv init langgraph-send-command
cd langgraph-send-command
uv venv --python 3.12
uv add "langchain>=1.0,<2.0" langchain-openai langchain-core langgraph python-dotenv rich

在项目根目录创建 .env，写入 Coding Plan 的 Key 与专用 Base URL。
不要把 .env 提交进 Git。

1 2	OPENAI_API_KEY=你的火山方舟 API Key OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

请勿改用普通方舟 .../api/v3，否则无法消耗 Coding Plan 套餐额度。
后续示例复制到项目根后执行 uv run python xxx.py。

实现

Send 动态扇出

Send(node_name, payload) 告诉运行时：向指定节点投递一份私有输入。
条件边返回 list[Send] 时，同一节点可被并行调用多次，各自拿到不同 payload。

下面用一个 worker 按体裁写诗、词、笑话；路由函数为每种体裁发出一条 Send。

import os
from typing import Literal, Sequence, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.4,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

CONTENT_TYPES = ["poem", "joke", "ci_poem"]


class OverAllState(TypedDict):
    topic: str
    poem: str
    ci_poem: str
    joke: str


class WorkerState(TypedDict):
    content_type: Literal["poem", "joke", "ci_poem"]
    prompt: str


class InputState(TypedDict):
    topic: str


class OutputState(TypedDict):
    poem: str
    ci_poem: str
    joke: str


def worker_node(state: WorkerState) -> OutputState:
    content_type = state["content_type"]
    content = model.invoke([HumanMessage(state["prompt"])]).content
    return {content_type: content}


def router(state: InputState) -> Sequence[Send]:
    labels = {"poem": "七言绝句", "joke": "笑话", "ci_poem": "词"}
    topic = state["topic"]
    return [
        Send(
            "worker_node",
            {
                "content_type": content_type,
                "prompt": f"请生成关于{topic}的{labels[content_type]}，只输出正文。",
            },
        )
        for content_type in CONTENT_TYPES
    ]


builder = StateGraph(
    state_schema=OverAllState,
    input_schema=InputState,
    output_schema=OutputState,
)
builder.add_node("worker_node", worker_node)
builder.add_conditional_edges(START, router, path_map=["worker_node"])
builder.add_edge("worker_node", END)

graph = builder.compile()
rprint(graph.invoke({"topic": "莲花"}))

要点：path_map 仍要声明可能到达的节点名；真正「发几份、发什么」由 Send 列表决定。
worker 返回的字段名与 content_type 一致，才能写回全局状态的对应键。

Command 路由

Command 可同时携带 update（写入状态）与 goto（下一节点）。
节点返回类型写成 Command[Literal[...]]，编译器才能识别可达目标。

下面用 Command 替代「路由函数 + 条件边」的经典写法。

import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.4,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class OverAllState(TypedDict):
    topic: str
    content_type: Literal["poem", "joke"]
    content_chinese: str
    poem: str
    joke: str


def router(
    state: OverAllState,
) -> Command[Literal["poem_node", "joke_node", "__end__"]]:
    if state["content_type"] == "poem":
        return Command(update={"content_chinese": "一首诗"}, goto="poem_node")
    if state["content_type"] == "joke":
        return Command(update={"content_chinese": "一个笑话"}, goto="joke_node")
    return Command(goto=END)


def poem_node(state: OverAllState) -> OverAllState:
    text = model.invoke(
        f"写一首关于{state['topic']}主题的{state['content_chinese']}，只输出正文。"
    ).content
    return {"poem": text}


def joke_node(state: OverAllState) -> OverAllState:
    text = model.invoke(
        f"写一个关于{state['topic']}主题的{state['content_chinese']}，只输出正文。"
    ).content
    return {"joke": text}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("router", router)
builder.add_node("poem_node", poem_node)
builder.add_node("joke_node", joke_node)
builder.add_edge(START, "router")
builder.add_edge("poem_node", END)
builder.add_edge("joke_node", END)

graph = builder.compile()
rprint(graph.invoke({"topic": "莲花", "content_type": "poem"}))
rprint(graph.invoke({"topic": "猫咪", "content_type": "joke"}))
rprint(graph.invoke({"topic": "猫咪", "content_type": "xxx"}))  # type: ignore[arg-type]

未知 content_type 时直接 goto=END，图会干净结束且不调用生成节点。
Command 适合「改一点状态再跳转」；纯分支仍可用条件边。

扇入汇合

并行分支汇入同一节点时，边的写法决定是「或」还是「与」。
分条 add_edge(a, e) / add_edge(b, e) 偏或汇合；add_edge([a, b], e) 表示与汇合（两侧都完成后再进 e）。

下面用空状态与日志观察超步，确认汇合语义。

from typing import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class EmptyState(TypedDict):
    pass


def make_node(name: str):
    def _node(state: EmptyState, config: RunnableConfig) -> EmptyState:
        step = config["metadata"]["langgraph_step"]
        rprint(f"{name} @ step={step}")
        return {}

    return _node


builder = StateGraph(state_schema=EmptyState)
for name in ["node_a", "node_b", "node_c", "node_d", "node_e"]:
    builder.add_node(name, make_node(name))

builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_a", "node_c")
builder.add_edge("node_b", "node_d")
# 或汇合：c 或 d 到达即可触发 e
builder.add_edge("node_c", "node_e")
builder.add_edge("node_d", "node_e")
# 与汇合可改为：builder.add_edge(["node_c", "node_d"], "node_e")
builder.add_edge("node_e", END)

graph = builder.compile()
graph.invoke({})

需要严格「两侧都完成再汇总」时，改用列表形式的与边，避免过早进入汇合节点。
与 Send 扇出搭配时，下游 reducer 字段通常要加 Annotated[..., add] 之类归约器。

MapReduce

经典 MapReduce：分发 → map → reduce。
用 Send 把每条输入发给 mapper_node，用 Annotated[list, add] 合并中间结果，再由 reducer_node 聚合。

下面统计多句英文里的词频，全程不调用 LLM。

from operator import add
from typing import Annotated, Sequence, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from rich import print as rprint


class OverAllState(TypedDict):
    input_values: list[str]
    entries: Annotated[list[tuple[str, int]], add]
    word_counts: dict[str, int]


class MapperInputState(TypedDict):
    input_value: str


def router_node(state: OverAllState) -> Sequence[Send]:
    return [
        Send("mapper_node", {"input_value": value})
        for value in state["input_values"]
    ]


def mapper_node(state: MapperInputState) -> OverAllState:
    entries = [(word, 1) for word in state["input_value"].split()]
    return {"entries": entries}


def reducer_node(state: OverAllState) -> OverAllState:
    buckets: dict[str, list[int]] = {}
    for key, value in state["entries"]:
        buckets.setdefault(key, []).append(value)
    return {"word_counts": {k: sum(vs) for k, vs in buckets.items()}}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("mapper_node", mapper_node)
builder.add_node("reducer_node", reducer_node)
builder.add_conditional_edges(START, router_node, path_map=["mapper_node"])
builder.add_edge("mapper_node", "reducer_node")
builder.add_edge("reducer_node", END)

graph = builder.compile()
rprint(
    graph.invoke(
        {"input_values": ["hello world", "hello atguigu", "hello llm"]}
    )
)

entries 必须声明归约器，否则并行 map 会互相覆盖而不是追加。
reducer_node 只在所有 map 结果汇齐后执行一次，适合做最终聚合。

验证

运行 Send 示例：输出应同时包含 poem / joke / ci_poem 三个键。
运行 Command 示例：content_type=poem 只填 poem；非法类型直接结束。
运行 MapReduce：word_counts["hello"] 应为 3。

总结

Send：运行时动态扇出，一份图定义可派生 N 个同构任务。
Command：一次返回完成 update + goto，适合有状态的路由节点。
扇入：分条边与列表边分别对应或/与汇合，选型影响下游触发时机。
MapReduce：Send + Annotated[..., add] + reduce 节点即可落地词频类批处理。