LangGraph 中 add_node 与 add_edge 的 state 参数详解

一、add_node 与 state 参数详解

在 StateGraph 中，你需要首先定义一个 State 类型（通常为 TypedDict 或 dataclass）。每个节点函数必须接受且仅接受一个参数——当前 State，并返回一个 字典，字典的键是 State 中的字段名，值是对该字段的更新。LangGraph 会自动将返回的字典与当前 State 合并（浅合并，除非使用了 Annotated 或 reducer）。

from typing import TypedDict
from langgraph.graph import StateGraph

class MyState(TypedDict):
    count: int
    messages: list

def node_func(state: MyState) -> dict:
    # 可以读取 state 中的任意字段
    new_count = state["count"] + 1
    # 返回需要更新的字段（其他字段保持不变）
    return {"count": new_count}

builder = StateGraph(MyState)
builder.add_node("increment", node_func)

重点：

• 节点函数不能直接修改传入的 state 对象，必须通过返回值描述改动。
• 返回的字典支持部分更新：未提及的字段会保留原值。
• 如果某个字段需要更复杂的合并逻辑（如向列表追加而非覆盖），可以使用 Annotated + reducer 函数（例如 operator.add）。

二、add_edge 与 state 的关系

add_edge 用于连接两个节点（或节点与终点）。State 在边上传递时不做任何修改——下游节点接收到的 State 是上游节点返回并合并后的完整 State。条件边 add_conditional_edges 也是同样的机制，只是路由依据 State 的值来决定走哪条边。

builder.add_edge("node_a", "node_b")  # node_a 执行后，状态自动流入 node_b

常见模式：

• 线性流程：start -> node1 -> node2 -> end
• 分支流程：通过条件边根据 State 字段选择不同节点
• 循环：条件边可以指回之前的节点，State 在其中持续累积

三、实用例程

例程 1：计数器（线性传递）

演示 State 的基本读写与边连接。

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class CounterState(TypedDict):
    total: int
    logs: Annotated[list, operator.add]   # 自动合并列表

def add_one(state: CounterState) -> dict:
    return {"total": state["total"] + 1, "logs": ["加1"]}

def double(state: CounterState) -> dict:
    return {"total": state["total"] * 2, "logs": ["翻倍"]}

def reset(state: CounterState) -> dict:
    return {"total": 0, "logs": ["重置"]}

builder = StateGraph(CounterState)
builder.add_node("add", add_one)
builder.add_node("mul", double)
builder.add_node("clear", reset)
builder.add_edge(START, "add")
builder.add_edge("add", "mul")
builder.add_edge("mul", "clear")
builder.add_edge("clear", END)
graph = builder.compile()

# 运行
result = graph.invoke({"total": 5, "logs": []})
print(result)  # {'total': 0, 'logs': ['加1', '翻倍', '重置']}

例程 2：条件分支（根据 State 决策）

模拟一个简单质检智能体：根据内容长度决定是直接回复还是转人工。

from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END

class QAAgentState(TypedDict):
    query: str
    length: int
    action: str

def check_length(state: QAAgentState) -> dict:
    length = len(state["query"])
    return {"length": length}

def auto_reply(state: QAAgentState) -> dict:
    return {"action": f"自动回复：你说了 {state['length']} 个字"}

def human_transfer(state: QAAgentState) -> dict:
    return {"action": "转人工客服"}

def route_by_length(state: QAAgentState) -> Literal["auto_reply", "human_transfer"]:
    if state["length"] < 20:
        return "auto_reply"
    return "human_transfer"

builder = StateGraph(QAAgentState)
builder.add_node("check", check_length)
builder.add_node("auto_reply", auto_reply)
builder.add_node("human_transfer", human_transfer)
builder.add_edge(START, "check")
builder.add_conditional_edges("check", route_by_length)
builder.add_edge("auto_reply", END)
builder.add_edge("human_transfer", END)

graph = builder.compile()
print(graph.invoke({"query": "Hi", "length": 0, "action": ""}))
# {'query': 'Hi', 'length': 2, 'action': '自动回复：你说了 2 个字'}

例程 3：循环计数器（带退出条件）

State 在循环中不断累加，直到满足条件才退出。

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

class LoopState(TypedDict):
    count: int
    max: int

def increment(state: LoopState) -> dict:
    return {"count": state["count"] + 1}

def should_continue(state: LoopState) -> str:
    if state["count"] >= state["max"]:
        return "exit"
    return "loop"

builder = StateGraph(LoopState)
builder.add_node("inc", increment)
builder.add_edge(START, "inc")
builder.add_conditional_edges("inc", should_continue, {"loop": "inc", "exit": END})
graph = builder.compile()

print(graph.invoke({"count": 0, "max": 5}))
# {'count': 5, 'max': 5}

例程 4：LLM 简单对话（累积消息）

使用 Annotated[list, operator.add] 自动合并消息历史。

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage

class ChatState(TypedDict):
    messages: Annotated[list, operator.add]

def chatbot(state: ChatState) -> dict:
    # 模拟调用 LLM（实际可用 langchain 模型）
    last_msg = state["messages"][-1].content
    reply = f"Echo: {last_msg}"
    return {"messages": [AIMessage(content=reply)]}

builder = StateGraph(ChatState)
builder.add_node("chat", chatbot)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

graph = builder.compile()
state = graph.invoke({"messages": [HumanMessage(content="Hello")]})
print(state["messages"])
# [HumanMessage(content='Hello'), AIMessage(content='Echo: Hello')]

四、实用 FAQ

Q1：节点函数可以返回 None 吗？
不可以。 必须返回一个字典（可以是空字典 {}，表示不更新任何字段）。返回 None 会引发运行时错误。

Q2：如何实现字段数值的累加（而非覆盖）？
使用 Annotated[T, reducer]。例如 total: Annotated[int, lambda a,b: a+b] 或使用 operator.add。LangGraph 会自动合并多次返回的同一字段。

Q3：State 是一个普通字典，我可以修改它然后返回吗？
强烈不建议。 虽然技术上你可以 state["count"] += 1; return {}，但这会产生数据不一致的风险（LangGraph 内部会缓存和检查）。始终通过返回显式更新字段。

Q4：add_edge 影响 State 的内容吗？
不影响。State 沿边原样传递。只有节点函数的返回值会触发合并更新。

Q5：如何初始化 State？
在调用 graph.invoke() 时传入初始字典。也可以在图中加入一个起始节点（前置节点）专门负责设置默认值。

Q6：可以在一个节点中返回嵌套字典的部分更新吗？
可以，但返回值必须与顶层字段匹配。如需深度合并，可以自定义 reducer 或返回完整子字典。

Q7：条件边 add_conditional_edges 的路由函数能接受多个参数吗？
不能。路由函数只能接受 state 一个参数，返回字符串或字符串列表（支持并行分支）。

Q8：State 中能放 Pydantic 模型吗？
可以，但推荐使用 TypedDict 获得更好的 IDE 提示。如果使用 Pydantic 模型，需确保可序列化（LangGraph 内部依赖 JSON 兼容数据）。

Q9：如何调试 State 的变化？
可以添加一个“打印节点”：

def debug_node(state: MyState) -> dict:
    print(state)
    return {}
builder.add_node("debug", debug_node)

将 debug 插入到任何位置观察 State 快照。

Q10：多个节点同时写入同一字段，最终结果如何？
取决于字段的 reducer。默认是覆盖（最后一个节点生效）。如果使用 operator.add 则按顺序合并。

总结

add_node 和 add_edge 构成了 LangGraph 的执行图，而 state 参数是节点之间通信的唯一载体。牢记“读取当前 state，返回修改字典”这一原则，再配合 Annotated 实现灵活合并，就能构建出强大的状态机、多步骤 Agent 或复杂工作流。通过条件边和循环，你可以让流程具备真正的智能决策能力。希望本文的例程和 FAQ 能帮助你快速上手，避免踩坑。