LangGraph 系列 10：LangGraph Workflow 的 State、Reducer 与路由

LangGraph 系列 1—9 已经完成 Agent、Tool、Runtime Context、AgentState、短期记忆和长期记忆。前九篇主要围绕模型驱动的 Agent 展开，但 Agent 只是 LangGraph 的一种应用方式。

从本文开始进入 LangGraph 系列的 Workflow 阶段。面对执行步骤明确、分支规则固定的业务，通常不需要让大模型决定每一步，而是由程序员把执行路径直接编排成 Workflow。

本篇先不调用大模型，使用普通 Python 函数把 Workflow 中最重要的 State、Reducer、Node 和 Edge 逐一验证清楚。这样可以排除模型输出的不确定性，直接观察状态究竟怎样在图中流转。

1. Workflow 与 Agent 的区别

Workflow 和 Agent 都可以组织复杂任务，但两者决定执行路径的方式不同。

Workflow 的路径由代码提前定义。例如一个出行建议流程可以固定为：

清理输入
-> 判断温度
-> 根据温度选择建议
-> 整理最终结果

Agent 则会让模型根据消息和工具结果决定下一步动作。模型可能调用工具，也可能直接回答，还可能连续调用多个工具。

因此，两者的核心区别是：

• Workflow 由程序员编排流程。
• Agent 由模型参与决定流程。

它们并不互斥。一个 Agent 可以成为 Workflow 中的某个节点，多个 Agent 也可以由一张更大的 Workflow 统一编排。

2. 一张工作流由什么组成

LangGraph 使用图来描述 Workflow，最基础的组成部分有三个：

1. State：保存当前执行过程中的共享数据。
2. Node：读取 State，执行一段逻辑，再返回状态更新。
3. Edge：决定当前节点完成后进入哪个节点。

可以把 State 理解为一张不断变化的数据表。节点负责产生更新，Reducer 负责决定更新如何写回，边负责决定接下来执行谁。

3. 使用 TypedDict 定义 State

工作流不会自动提供一个满足业务需要的 State。创建 StateGraph 时，需要明确传入 State Schema：

from typing_extensions import TypedDict


class TravelState(TypedDict):
    city: str
    temperature: int
    category: str
    advice: str
    result: str

本文使用 TypedDict，因为它的代码最直接，也不会在运行时创建额外对象。LangGraph 1.1.3 也支持 dataclass 和 Pydantic：

• dataclass 适合需要默认值的状态。
• Pydantic 适合需要运行时输入校验的状态。
• TypedDict 适合结构明确、追求简单和低开销的状态。

State 中应该保存流程真正需要共享的数据，不要因为某个局部函数临时使用一个变量，就把所有局部变量都放进 State。

4. 理解 State 的三条规则

学习 Workflow State 时，可以先记住三条规则。

4.1 节点和路由函数可以读取当前 State

节点函数的第一个参数通常是 State：

def normalize_input(state: TravelState) -> dict:
    return {"city": state["city"].strip()}

路由函数同样可以读取 State，并根据其中的数据选择下一条路径。

4.2 节点只返回需要更新的字段

节点不需要返回完整 State。如果一个节点只修改城市名称，就只返回 city：

return {"city": state["city"].strip()}

没有出现在返回值中的字段会保持不变。

4.3 Reducer 决定更新如何写回

State 中的每个字段都有独立的更新规则。没有显式指定 Reducer 时，LangGraph 使用默认覆盖行为。

Reducer 可以理解为一个二元合并函数：左侧参数是 State 中已经存在的字段值，右侧参数是节点本次返回的字段更新，函数返回合并后的新值。

new_value = reducer(current_value, update_value)

不同字段可以使用不同的 Reducer。例如，score 使用默认覆盖，steps 使用 operator.add 追加列表。节点只负责返回局部更新，不需要自己把旧值和新值拼接起来。

如果一个节点返回：

{"score": 2}

下一个节点又返回：

{"score": 3}

最终 score 是 3，不是 5，因为这个字段没有配置累加 Reducer。

5. Reducer 的三种常见行为

5.1 默认覆盖

普通字段不写 Reducer时，新值覆盖旧值：

class WorkflowState(TypedDict):
    score: int

默认覆盖适合流程阶段、最终结果、分类结果等只需要保留最新值的数据。

5.2 使用 operator.add 追加普通列表

如果多个节点都要向列表追加内容，可以使用 Annotated 绑定 Reducer：

import operator
from typing import Annotated


class WorkflowState(TypedDict):
    steps: Annotated[list[str], operator.add]

当节点分别返回 [‘first_step’] 和 [‘second_step’] 时，最终列表会保留两项，而不是让后一项覆盖前一项。

5.3 使用 add_messages 处理消息列表

消息不能只按普通列表处理。LangGraph 提供了 add_messages：

from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages


class WorkflowState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

它除了追加新消息，还会跟踪消息 ID：

• 新 ID 会追加到列表。
• 相同 ID 会替换原消息。
• 配合 RemoveMessage 可以删除指定消息。
• 字典消息可以转换成 LangChain 消息对象。

如果只需要消息列表，也可以直接使用预置的 MessagesState：

from langgraph.graph import MessagesState


class CustomState(MessagesState):
    user_name: str

MessagesState 已经定义了使用 add_messages 的 messages 字段。

6. 验证 State 与 Reducer

示例代码放在：

llm_learning/langgraph/p10_workflow_basics/

继续使用现有 Python 3.12 环境：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate
python -m pip install -r langgraph/p10_workflow_basics/requirements.txt

本文固定使用：

langgraph==1.1.3
langchain-core==1.2.22
typing-extensions==4.15.0

6.1 完整代码

01_state_and_reducers.py：

"""演示 Workflow State 中默认覆盖、列表追加和消息合并三种 Reducer 行为。"""

import operator
from typing import Annotated

from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict


class WorkflowState(TypedDict):
    """定义所有节点共享的工作流状态。"""

    # 没有声明 Reducer，节点返回新值时会覆盖旧值。
    score: int
    # 普通列表使用 operator.add，把每次返回的列表追加到已有列表。
    steps: Annotated[list[str], operator.add]
    # 消息列表使用 add_messages，可以按消息 ID 追加或替换消息。
    messages: Annotated[list[AnyMessage], add_messages]


def first_step(state: WorkflowState) -> dict:
    """第一次更新分数、执行记录和回答消息。"""

    return {
        "score": 2,
        "steps": ["first_step"],
        "messages": [AIMessage(content="初始建议", id="answer-1")],
    }


def second_step(state: WorkflowState) -> dict:
    """再次覆盖分数，并用相同 ID 修正上一条 AI 消息。"""

    return {
        "score": 3,
        "steps": ["second_step"],
        "messages": [AIMessage(content="修正后的建议", id="answer-1")],
    }


# StateGraph 是图的构建器，需要先添加节点和边。
builder = StateGraph(WorkflowState)
builder.add_node(first_step)
builder.add_node(second_step)
builder.add_edge(START, "first_step")
builder.add_edge("first_step", "second_step")
builder.add_edge("second_step", END)

# compile() 返回可以 invoke() 和 stream() 的 CompiledStateGraph。
graph = builder.compile()

result = graph.invoke(
    {
        "score": 1,
        "steps": ["input"],
        "messages": [HumanMessage(content="请给我一条建议。", id="user-1")],
    }
)

print("score:", result["score"])
print("steps:", result["steps"])
print("message_count:", len(result["messages"]))

for message in result["messages"]:
    print(type(message).__name__, message.id, message.content)

运行：

python langgraph/p10_workflow_basics/01_state_and_reducers.py

真实输出：

score: 3
steps: ['input', 'first_step', 'second_step']
message_count: 2
HumanMessage user-1 请给我一条建议。
AIMessage answer-1 修正后的建议

从结果可以确认：

1. score 从 1 依次被覆盖为 2 和 3。
2. steps 使用 operator.add，三个阶段都被保留。
3. 两个节点返回的 AI 消息 ID 都是 answer-1，因此最终只保留修正后的消息。

7. Node 是真正执行工作的步骤

Node 通常是同步或异步 Python 函数，也可以是实现了 Runnable 接口的对象。

最简单的节点只接收 State：

def classify_temperature(state: TravelState) -> dict:
    return {"category": "hot"}

根据需要，节点还可以接收：

• config：读取 thread_id、tags、metadata 等运行配置。
• runtime：读取 Runtime Context、Store 和流式写入器等运行时能力。

这些内容前面已经介绍过，本篇只保留 state 参数，集中观察工作流本身。

节点返回的字典表示“状态更新”，不是新的完整 State。尤其是配置了列表 Reducer 后，如果节点把整个旧列表和新内容一起返回，旧内容会被重复追加。

8. 普通边、START 与 END

普通边表示固定执行路径：

builder.add_edge(START, "normalize_input")
builder.add_edge("normalize_input", "classify_temperature")
builder.add_edge("format_result", END)

• START 是虚拟开始节点，负责把图输入发送给第一个节点。
• END 是虚拟结束节点，表示后面不再执行其他操作。
• add_edge(‘a’, ‘b’) 表示节点 a 完成后固定进入节点 b。

START 和 END 不会执行普通 Python 业务逻辑，它们只是图结构中的特殊标记。

9. 条件边与路由函数

如果下一步取决于 State，就需要条件边：

builder.add_conditional_edges(
    "classify_temperature",
    route_temperature,
)

路由函数读取当前 State，然后返回下一个节点名称：

def route_temperature(
    state: TravelState,
) -> Literal["hot_advice", "cold_advice", "comfortable_advice"]:
    routes = {
        "hot": "hot_advice",
        "cold": "cold_advice",
        "comfortable": "comfortable_advice",
    }
    return routes[state["category"]]

Literal 不只是类型提示。它也让 LangGraph 知道路由函数可能到达哪些节点，生成的图结构会更准确。

如果路由返回值不是实际节点名称，可以提供 path_map：

builder.add_conditional_edges(
    "check_result",
    route_result,
    {True: "success", False: "retry"},
)

此时路由函数可以返回 True 或 False，再由映射表转换成节点名称。

10. 完整的固定边与条件边案例

这个案例根据温度选择出行建议：

START
-> normalize_input
-> classify_temperature
-> hot_advice / cold_advice / comfortable_advice
-> format_result
-> END

10.1 完整代码

02_fixed_and_conditional_workflow.py：

"""使用温度出行建议案例演示固定边、条件边、invoke 和 stream。"""

import operator
from typing import Annotated, Literal

from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class TravelState(TypedDict):
    """保存输入、判断结果、最终建议和节点执行记录。"""

    city: str
    temperature: int
    category: str
    advice: str
    result: str
    steps: Annotated[list[str], operator.add]


def normalize_input(state: TravelState) -> dict:
    """清理城市名称两侧的空格。"""

    return {
        "city": state["city"].strip(),
        "steps": ["normalize_input"],
    }


def classify_temperature(state: TravelState) -> dict:
    """根据温度写入分类结果。"""

    if state["temperature"] >= 30:
        category = "hot"
    elif state["temperature"] <= 10:
        category = "cold"
    else:
        category = "comfortable"

    return {
        "category": category,
        "steps": ["classify_temperature"],
    }


def route_temperature(
    state: TravelState,
) -> Literal["hot_advice", "cold_advice", "comfortable_advice"]:
    """读取分类结果，返回下一步需要执行的节点名称。"""

    routes = {
        "hot": "hot_advice",
        "cold": "cold_advice",
        "comfortable": "comfortable_advice",
    }
    return routes[state["category"]]


def hot_advice(state: TravelState) -> dict:
    """生成高温天气建议。"""

    return {
        "advice": "注意防晒并及时补水。",
        "steps": ["hot_advice"],
    }


def cold_advice(state: TravelState) -> dict:
    """生成低温天气建议。"""

    return {
        "advice": "注意保暖，外出时带好外套。",
        "steps": ["cold_advice"],
    }


def comfortable_advice(state: TravelState) -> dict:
    """生成舒适天气建议。"""

    return {
        "advice": "天气舒适，适合户外散步。",
        "steps": ["comfortable_advice"],
    }


def format_result(state: TravelState) -> dict:
    """把城市与建议整理成最终结果。"""

    return {
        "result": f'{state["city"]}：{state["advice"]}',
        "steps": ["format_result"],
    }


builder = StateGraph(TravelState)
builder.add_node(normalize_input)
builder.add_node(classify_temperature)
builder.add_node(hot_advice)
builder.add_node(cold_advice)
builder.add_node(comfortable_advice)
builder.add_node(format_result)

# START 到分类节点之间是执行路径固定的普通边。
builder.add_edge(START, "normalize_input")
builder.add_edge("normalize_input", "classify_temperature")

# 分类完成后调用路由函数，只选择三个建议节点中的一个。
builder.add_conditional_edges("classify_temperature", route_temperature)

# 三个分支最终都会进入同一个格式化节点。
builder.add_edge("hot_advice", "format_result")
builder.add_edge("cold_advice", "format_result")
builder.add_edge("comfortable_advice", "format_result")
builder.add_edge("format_result", END)

graph = builder.compile()


def make_input(city: str, temperature: int) -> TravelState:
    """创建字段完整的测试输入。"""

    return {
        "city": city,
        "temperature": temperature,
        "category": "",
        "advice": "",
        "result": "",
        "steps": [],
    }


# 三组输入分别验证高温、低温和舒适三个分支。
for city, temperature in [(" 杭州 ", 32), ("哈尔滨", 5), ("昆明", 22)]:
    result = graph.invoke(make_input(city, temperature))
    print(result["result"])
    print("steps:", result["steps"])

print("stream updates:")

# updates 模式只返回每个节点本次提交的局部状态更新。
for update in graph.stream(make_input(" 杭州 ", 32), stream_mode="updates"):
    print(update)

10.2 真实输出

杭州：注意防晒并及时补水。
steps: ['normalize_input', 'classify_temperature', 'hot_advice', 'format_result']
哈尔滨：注意保暖，外出时带好外套。
steps: ['normalize_input', 'classify_temperature', 'cold_advice', 'format_result']
昆明：天气舒适，适合户外散步。
steps: ['normalize_input', 'classify_temperature', 'comfortable_advice', 'format_result']
stream updates:
{'normalize_input': {'city': '杭州', 'steps': ['normalize_input']}}
{'classify_temperature': {'category': 'hot', 'steps': ['classify_temperature']}}
{'hot_advice': {'advice': '注意防晒并及时补水。', 'steps': ['hot_advice']}}
{'format_result': {'result': '杭州：注意防晒并及时补水。', 'steps': ['format_result']}}

三组输入分别走了三个不同分支。stream_mode=’updates’ 返回的不是每一步完整 State，而是节点本次提交的局部更新，因此特别适合观察哪个节点修改了哪些字段。

11. StateGraph 为什么需要 compile

StateGraph 是图结构的 Builder。它只负责注册状态、节点和边，不能直接执行：

builder = StateGraph(TravelState)

完成图结构后，需要编译：

graph = builder.compile()

compile() 会检查图结构，并返回 CompiledStateGraph。真正提供 invoke()、stream()、ainvoke() 和 astream() 的是编译后的图。

常见调用方式包括：

# 返回执行完成后的最终 State。
result = graph.invoke(input_state)

# 逐步返回每个节点提交的局部状态更新。
for update in graph.stream(input_state, stream_mode="updates"):
    print(update)

# 逐步返回每个阶段执行后的完整 State 快照。
for state in graph.stream(input_state, stream_mode="values"):
    print(state)

这些执行方式和 StateGraph 的完整接口可以继续参考 LangGraph Graph API。

12. 并行分支与 Fan-in

路由函数不只能返回一个节点名称，也可以返回多个节点：

def choose_tasks(state: ParallelState) -> list[str]:
    return ["walking_advice", "water_advice"]

这两个节点会在下一个 Super-step 中执行。它们都更新 suggestions，所以这个字段必须配置 Reducer：

suggestions: Annotated[list[str], operator.add]

等待多个节点都完成后再进入汇总节点，可以使用列表形式的起点：

builder.add_edge(
    ["walking_advice", "water_advice"],
    "summarize",
)

这就是 Fan-out 和 Fan-in：先从一个位置分发到多个并行节点，再等待它们完成后汇聚到一个节点。

12.1 完整代码

03_parallel_branches.py：

"""演示路由到多个并行节点，并在两个分支完成后执行 Fan-in 汇总。"""

import operator
from typing import Annotated

from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class ParallelState(TypedDict):
    """并行分支共享的状态。"""

    city: str
    # 两个并行节点都会更新 suggestions，因此必须声明 Reducer。
    suggestions: Annotated[list[str], operator.add]
    steps: Annotated[list[str], operator.add]
    summary: str


def prepare(state: ParallelState) -> dict:
    """记录并行任务开始。"""

    return {"steps": ["prepare"]}


def choose_tasks(state: ParallelState) -> list[str]:
    """一次返回两个节点名称，让它们在下一个 Super-step 执行。"""

    return ["walking_advice", "water_advice"]


def walking_advice(state: ParallelState) -> dict:
    """生成散步建议。"""

    return {
        "suggestions": [f'在{state["city"]}散步 30 分钟'],
        "steps": ["walking_advice"],
    }


def water_advice(state: ParallelState) -> dict:
    """生成饮水建议。"""

    return {
        "suggestions": ["随身携带饮用水"],
        "steps": ["water_advice"],
    }


def summarize(state: ParallelState) -> dict:
    """等待两个分支完成后汇总建议。"""

    # 排序后再连接，使教学输出保持稳定。
    suggestions = sorted(state["suggestions"])
    return {
        "summary": "；".join(suggestions),
        "steps": ["summarize"],
    }


builder = StateGraph(ParallelState)
builder.add_node(prepare)
builder.add_node(walking_advice)
builder.add_node(water_advice)
builder.add_node(summarize)

builder.add_edge(START, "prepare")

# path_map 列出路由函数可能返回的目标节点，便于校验和绘图。
builder.add_conditional_edges(
    "prepare",
    choose_tasks,
    ["walking_advice", "water_advice"],
)

# 列表形式的起点表示：两个并行节点都完成后，才执行 summarize。
builder.add_edge(["walking_advice", "water_advice"], "summarize")
builder.add_edge("summarize", END)

graph = builder.compile()

result = graph.invoke(
    {
        "city": "杭州",
        "suggestions": [],
        "steps": [],
        "summary": "",
    }
)

print("suggestions:", result["suggestions"])
print("steps:", result["steps"])
print("summary:", result["summary"])

12.2 真实输出

suggestions: ['在杭州散步 30 分钟', '随身携带饮用水']
steps: ['prepare', 'walking_advice', 'water_advice', 'summarize']
summary: 在杭州散步 30 分钟；随身携带饮用水

summarize 只出现一次，并且执行前已经拿到两个分支的结果，说明 Fan-in 生效。

两个分支位于同一个 Super-step，不应把合并后的列表顺序理解为业务执行顺序保证。示例在 summarize 中先对建议排序，再生成稳定输出；真实业务如果依赖顺序，应在数据中保存明确的序号或优先级，然后在汇总节点排序。

13. 为什么并行更新必须配置 Reducer

如果两个并行节点同时更新同一个字段，而该字段没有 Reducer，LangGraph 无法判断应该保留哪个值。

例如下面的字段只有默认覆盖行为：

class State(TypedDict):
    values: list[str]

两个并行节点分别返回 [‘left’] 和 [‘right’] 时，实际会产生：

InvalidUpdateError
At key 'values': Can receive only one value per step. Use an Annotated key to handle multiple values.

上面保留了本次测试中最关键的异常信息，省略了调用栈。

这不是随机选择一个结果，也不是后执行的节点覆盖前一个节点。LangGraph 会直接拒绝存在歧义的并行更新。

如果业务目标是合并两个列表，应显式写成：

values: Annotated[list[str], operator.add]

14. 常见问题

14.1 节点返回完整 State

节点只应返回需要修改的字段。对于使用 operator.add 的列表，如果把旧列表和新值一起返回，Reducer 会再次追加旧内容，造成重复数据。

14.2 列表没有配置 Reducer

没有 Reducer 的列表仍然使用覆盖行为。前一个节点返回的列表会被后一个节点替换。

14.3 并行节点更新同一个普通字段

同一 Super-step 中存在多个值时，默认字段会触发 InvalidUpdateError。需要合并就配置 Reducer；不需要合并就让不同节点更新不同字段。

14.4 路由函数返回不存在的节点

路由返回值必须是已经注册的节点名称，或者能够通过 path_map 映射到真实节点。

建议为路由函数增加 Literal 返回类型，减少拼写错误，也让图结构可视化更准确。

14.5 同一节点同时配置普通边和条件边

普通边不会因为条件边选择了其他路径而自动停止。如果从同一个节点同时配置普通边和条件边，两边都可能执行。

一个节点通常只选择一种路由方式：路径固定时使用普通边，需要判断时使用条件边。

14.6 循环没有结束条件

Workflow 可以形成循环，但必须存在能够进入 END 的条件。如果始终返回循环节点，最终会触发 GraphRecursionError。

不要通过盲目提高 recursion_limit 掩盖错误，应先检查路由条件是否真的能够结束。

15. 总结

知识点	作用	本文验证结果
Workflow	由程序员提前编排执行路径	温度建议流程按固定规则运行
StateGraph	注册 State、Node 和 Edge 的构建器	编译后才能执行
State	保存工作流当前数据快照	所有节点读取同一份状态结构
默认 Reducer	使用新值覆盖旧值	score 最终为 3
operator.add	合并普通列表	节点执行记录和并行建议被保留
add_messages	按消息 ID 合并消息	相同 ID 的 AI 消息被替换
Node	读取 State 并返回局部更新	每个节点只修改需要的字段
普通边	连接固定的执行步骤	输入清理后固定进入温度分类
条件边	根据路由函数选择下一步	三种温度进入不同建议节点
START	表示图的输入入口	输入首先进入清理节点
END	表示工作流结束	格式化后终止执行
compile()	生成可执行图	返回 CompiledStateGraph
invoke()	返回最终 State	得到完整出行建议
stream()	逐步返回执行结果	updates 展示节点局部更新
Fan-out	从一个位置进入多个节点	同时生成散步与饮水建议
Fan-in	等待多个分支后汇总	汇总节点只执行一次

本文已经把 Workflow 的基本结构跑通。下一篇将在这些概念之上加入本地 Qwen3，使用评估器优化器构建“生成内容、评估结果、不合格时携带反馈重新生成”的循环工作流。