这可能是LangGraph实战中最容易被忽略，但也是最核心的一环：状态（State） 与触发执行的入口（invoke）之间的微妙关系。搞懂它们，你的计算图才能真正“活”起来。

我们从最基础的状态定义开始，然后一步步拆解graph.invoke到底往图里扔了什么。接着用实际代码跑一遍，看看数据是怎么在各个节点间流动、演化的。

一、回顾：为状态Schema“画像”

在之前的文章中，我们介绍了三种主流的状态Schema定义方式：TypedDict、dataclass和Pydantic。这里，我们以最常用的TypedDict为例，快速回顾一下核心概念。

from typing import TypedDict, Annotated
from operator import add

class OverAllState(TypedDict):
    logs: Annotated[list[str], add]  # 使用add归约器，多个节点的logs自动合并
    id: str  # 普通的字符串字段

这个OverAllState定义了两种字段：

• logs：一个字符串列表，用add归约器标记，表示多个节点返回的日志会自动拼接起来。
• id：一个普通的字符串，默认行为是“覆盖”——后写的值会无情地覆盖掉先写的。

二、graph.invoke：图的总阀门

2.1 入参就是“初始状态”

先说说invoke的第一个参数——它其实就是一个初始状态字典，是你喂给图的第一口“数据”。

result = graph.invoke({"logs": ["START"], "id": "start"})

这个字典{"logs": ["START"], "id": "start"}，就是整个图的“起点数据”，它会用来初始化OverAllState。

2.2 入参字段必须“门当户对”

因为OverAllState里面定义了logs和id两个字段，所以invoke传进来的字典也得包含这两个key，而且类型要对得上。如果漏掉一个，LangGraph就会报错——它不知道拿什么值去初始化那个缺席的字段。

2.3 入参是START节点的“第一印象”

当图从START节点开始执行，第一个节点（比如node_a）看到的state，就是invoke传进来的那个字典。

来看一个完整的例子，把整个过程串起来：

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

class OverAllState(TypedDict):
    logs: Annotated[list[str], add]
    id: str

def node_a(state: OverAllState) -> OverAllState:
    print("=== node_a 接收到的状态 ===")
    for k, v in state.items():
        print(f"k: {k}, v: {v}")
    return {"logs": ["node_a 更新状态"]}

def node_b(state: OverAllState) -> OverAllState:
    print("=== node_b 接收到的状态 ===")
    print(f"logs: {state['logs']}")
    print(f"id: {state['id']}")
    return {"logs": ["node_b 更新状态"], "id": "end"}

# 构建图
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile()

# 调用图
result = graph.invoke({"logs": ["START"], "id": "start"})
print("n=== 最终结果 ===")
print(result)

输出结果：

=== node_a 接收到的状态 ===
k: logs, v: ['START']
k: id, v: start
=== node_b 接收到的状态 ===
logs: ['START', 'node_a 更新状态']
id: start
=== 最终结果 ===
{'logs': ['START', 'node_a 更新状态', 'node_b 更新状态'], 'id': 'end'}

三、Reducer（归约器）的秘密

不知道你注意到没有，logs字段的行为很特别：每次节点返回新的日志，它都会被追加到已有列表中，而不是简单地覆盖。这背后就是归约器（Reducer）在起作用。

3.1 什么是Reducer？

简单来说，当一个字段被多个节点更新时，Reducer就是那个决定“到底保留谁，或者怎么合并”的裁判。在LangGraph里，它通过Annotated类型注解来指定。

logs: Annotated[list[str], add]

这里的add就是一个Reducer，它告诉系统：别覆盖，把新值和旧值加起来（对于列表，就是拼接）。

3.2 常见的Reducer们

Reducer	作用	示例
add	拼接/相加	[1,2] + [3] = [1,2,3]
operator.add	同add	同上
自定义函数	自定义合并逻辑	例如取最大值、去重等

3.3 没有Reducer的字段

对于没有用Annotated指定Reducer的字段，比如上面的id，默认行为就是“覆盖”。后执行的节点返回的值，会直接替换掉之前的值。

所以，在最终结果里：

• logs: ['START', 'node_a 更新状态', 'node_b 更新状态'] — 三个值被拼接起来。
• id: 'end' — node_b返回的'end'覆盖了初始的'start'。

四、数据流全景：一步步拆解

我们来把上面例子中的数据流动，一步步拆解清楚：

invoke({"logs": ["START"], "id": "start"})
│
▼
LangGraph 初始化 OverAllState
├─ logs: [] + ["START"] = ["START"] ← reducer 处理（空列表 + 初始值）
└─ id: "start" ← 直接赋值
│
▼
START → node_a
│
▼
node_a 收到 state = {"logs": ["START"], "id": "start"}
│
▼
node_a 返回 {"logs": ["node_a 更新状态"]} ← 只更新logs，id不变
│
▼
LangGraph 合并：
├─ logs: ["START"] + ["node_a 更新状态"] = ["START", "node_a 更新状态"]
└─ id: "start"（没变化，保持原值）
│
▼
node_a → node_b
│
▼
node_b 收到 state = {"logs": ["START", "node_a 更新状态"], "id": "start"}
│
▼
node_b 返回 {"logs": ["node_b 更新状态"], "id": "end"} ← 更新两个字段
│
▼
LangGraph 合并：
├─ logs: ["START", "node_a 更新状态"] + ["node_b 更新状态"]
│  = ["START", "node_a 更新状态", "node_b 更新状态"]
└─ id: "end"（覆盖原来的"start"）
│
▼
node_b → END
│
▼
返回最终结果：{'logs': ['START', 'node_a 更新状态', 'node_b 更新状态'], 'id': 'end'}

五、那些容易踩的坑与注意事项

5.1 入参可以只传部分字段吗？

除非字段有默认值，否则不行。如果Schema里定义了字段，但没有默认值，invoke时必须把所有字段都传进去，否则就会报错。

如果是Pydantic或dataclass，可以为字段设置默认值，比如：

from pydantic import BaseModel

class OverAllState(BaseModel):
    logs: Annotated[list[str], add] = []  # 默认空列表
    id: str = "default_id"  # 默认值

这样调用时就可以省略某些字段，比如：

result = graph.invoke({"logs": ["START"]})  # id使用默认值"default_id"

5.2 入参中的Reducer是怎么处理的？

当invoke接收到初始值时，LangGraph会先把Reducer应用到初始值上。对于addReducer，它会将初始值与一个空值进行合并。比如列表，就是空列表[] + 初始值。

5.3 节点可以不返回所有字段吗？

当然可以。节点只需要返回它想要更新的字段就好，其他字段会自动保持不变。这是LangGraph里一个很重要的设计原则——节点返回的是“部分更新”，而不是“完整状态”。

5.4 如果节点返回了Schema中没有的字段会怎样？

这个字段会被静默忽略。既不会报错，也不会添加到状态中。这可能导致一些非常隐蔽的bug，比如你拼错了字段名，程序依然正常运行，但你的更新意图完全没有生效。

六、一句话总结

概念	要点
graph.invoke入参	初始状态，必须包含所有必需字段（除非有默认值）
字段匹配	入参的key必须与Schema定义一致
Reducer处理	入参经过Reducer处理后进入第一个节点
节点返回值	部分更新，只需返回要修改的字段
无Reducer的字段	默认覆盖行为
多余字段	静默忽略，小心埋坑

一句话概括：graph.invoke(initial_state)的入参是图的起点数据，它必须包含State定义的所有字段（除非有默认值），这些数据会经过Reducer处理后进入第一个节点，随后在每个节点间按Reducer规则不断演化，最终得到图的输出结果。

七、面试题精选

面试题1：graph.invoke传入的初始状态和节点返回的状态有什么区别？

参考答案：

graph.invoke传入的是完整的初始状态，它包含了所有必需的字段。而节点返回的是部分状态更新，只需要包含该节点想要修改的字段即可。LangGraph会自动将节点返回的更新字典与当前状态进行合并，生成新的状态传递给下一个节点。

这种设计使得每个节点只需要关心自己需要修改的数据，而不必维护整个状态对象，降低了节点之间的耦合。

面试题2：如果一个节点返回的字段名拼写错误（比如把logs写成logss），会发生什么？

参考答案：

这个错误的字段会被静默忽略。由于logss不在状态Schema中，LangGraph不会将其合并到状态中，也不会报错。这会导致该节点的更新意图没有生效，而开发者可能很难发现这个问题。

解决方案：

1. 使用类型注解和IDE的自动补全功能避免拼写错误
2. 编写单元测试验证状态更新的正确性
3. 在开发阶段打印中间状态以便调试

面试题3：如何让某个字段在节点不更新时保留上一次的值？

参考答案：

这正是LangGraph的默认行为。节点只需要返回它想要更新的字段，其他字段会自动保持不变。例如：

def node_a(state):
    # 只更新logs，id保持不变
    return {"logs": ["node_a executed"]}

即使后面的节点也没有更新id，id的值会一直保留初始值或最近一次更新的值。

如果需要更精细的控制，可以使用自定义Reducer来实现更复杂的合并逻辑。

面试题4：addReducer对于不同类型的字段表现有何不同？

参考答案：

addReducer实际上使用的是Python的+运算符，因此其行为取决于操作数的类型：

类型	行为	示例
list	拼接	[1,2] + [3] = [1,2,3]
int	加法	5 + 3 = 8
str	拼接	"ab" + "cd" = "abcd"
set	不支持	会报错

对于集合类型的字段，可以使用自定义Reducer实现并集操作：

def union_reducer(a: set, b: set) -> set:
    return a | b

class MyState(TypedDict):
    tags: Annotated[set[str], union_reducer]

面试题5：如何在invoke时传入动态生成的初始状态？

参考答案：

可以在调用invoke之前动态构建初始状态字典。例如：

# 从数据库或API获取数据
user_data = fetch_user_from_db(user_id)
session_info = get_session_info()

# 动态构建初始状态
initial_state = {
    "logs": [f"Session started for user {user_data['name']"}],
    "id": session_info["session_id"],
    "user": user_data,
    "config": load_config()
}

# 调用图
result = graph.invoke(initial_state)

这种方法非常灵活，可以根据不同的业务场景生成不同的初始状态，使同一个图能够处理多样化的输入数据。

面试题6：graph.invoke除了初始状态外还有哪些参数？

参考答案：

graph.invoke的完整签名大致如下：

graph.invoke(
    input: dict,  # 初始状态
    config: Optional[RunnableConfig] = None,  # 运行时配置
    *args,  # 其他参数
    **kwargs  # 其他关键字参数
)

其中config参数可以用来设置：

• recursion_limit：图的最大递归深度，防止无限循环
• tags：用于追踪和监控的标签
• metadata：附加的元数据
• callbacks：回调函数，用于事件监听

例如：

result = graph.invoke(
    {"logs": [], "id": "start"},
    config={"recursion_limit": 100}
)

如果这篇文章帮你理清了头绪，或者，直接去试试吧。遇到卡壳的地方，欢迎随时交流。