企业官网建设流程全解析

3.1. 状态定义

状态的定义实际上是在声明状态的Schema，后者是状态字段的完整描述。

官方推荐了三种定义Schema的方式：TypedDict、dataclass、Pydantic

3.1.1. TypedDict

from typing import TypedDict,Annotated from operator import add from langgraph.graph import StateGraph,START,END #定义状态 class OverAllStatus(TypedDict): #日志的类型是list[str]，最新的方式不是覆盖是add 追加 logs: Annotated[list[str], add] cur_id:str # 定义节点 def node_1(state: OverAllStatus) -> OverAllStatus: pre_id=state["cur_id"] return { "logs":["node_1 执行完"], "cur_id":pre_id+",node_1" } # 定义节点 def node_2(state: OverAllStatus) -> OverAllStatus: pre_id=state["cur_id"] return { "logs":["node_2 执行完"], "cur_id":pre_id+",node_2" } # 定义边 #3.1 创建图 获取建造者 builder=StateGraph(state_schema=OverAllStatus) #3.2 添加节点 builder.add_node(node_1) builder.add_node(node_2) #3.3 添加边 builder.add_edge(START,"node_1") builder.add_edge("node_1","node_2") builder.add_edge("node_2",END) #4. 编译图 graph=builder.compile() #5. 执行图 result=graph.invoke({"logs":[],"cur_id":"start"}) print(result)

输出

{'logs': ['node_1 执行完', 'node_2 执行完'], 'cur_id': 'start,node_1,node_2'}

3.1.2. dataclass

属性调用方式由['字段名']变为.字段名。

from typing import Annotated from dataclasses import dataclass from operator import add from langgraph.graph import StateGraph,START,END #定义状态 @dataclass class OverAllStatus: #日志的类型是list[str]，最新的方式不是覆盖是add 追加 logs: Annotated[list[str], add] cur_id:str # 定义节点 def node_1(state: OverAllStatus) -> OverAllStatus: pre_id=state.cur_id return OverAllStatus( ["node_1 执行完"], pre_id+",node_1" ) # 定义节点 def node_2(state: OverAllStatus) -> OverAllStatus: pre_id=state.cur_id return OverAllStatus( ["node_2 执行完"], pre_id+",node_2" ) # 定义边 #3.1 创建图 获取建造者 builder=StateGraph(state_schema=OverAllStatus) #3.2 添加节点 builder.add_node(node_1) builder.add_node(node_2) #3.3 添加边 builder.add_edge(START,"node_1") builder.add_edge("node_1","node_2") builder.add_edge("node_2",END) #4. 编译图 graph=builder.compile() #5. 执行图 result=graph.invoke({"logs":[],"cur_id":"start"}) print(result)

输出

{'logs': ['node_1 执行完', 'node_2 执行完'], 'cur_id': 'start,node_1,node_2'}

3.1.3. Pydantic

Pydantic模型的字段访问方式和dataclass相同。

from typing import Annotated from operator import add from pydantic import BaseModel from langgraph.graph import StateGraph, START, END # 定义状态 class OverAllStatus(BaseModel): # 使用 add 作为归约函数，将每个节点返回的日志追加到原列表 logs: Annotated[list[str], add] cur_id: str # 定义节点 def node_1(state: OverAllStatus) -> dict: return { "logs": ["node_1 执行完"], "cur_id": f"{state.cur_id},node_1", } def node_2(state: OverAllStatus) -> dict: return { "logs": ["node_2 执行完"], "cur_id": f"{state.cur_id},node_2", } # 创建图 builder = StateGraph(state_schema=OverAllStatus) # 添加节点，并明确指定节点名称 builder.add_node("node_1", node_1) builder.add_node("node_2", node_2) # 添加边 builder.add_edge(START, "node_1") builder.add_edge("node_1", "node_2") builder.add_edge("node_2", END) # 编译图 graph = builder.compile() # 执行图 result = graph.invoke({ "logs": [], "cur_id": "start", }) print(result)

输出

{'logs': ['node_1 执行完', 'node_2 执行完'], 'cur_id': 'start,node_1,node_2'}

3.1.4. 校验行为

学习LangChain的结构化输出时我们提到：Pydantic对格式要求最严格，如果模型返回的内容不符合结构化Schema的要求，则抛出ValidationError。而其余方式都不会对模型的返回结果进行校验，即便模型返回的内容不符合结构化要求，也会原样返回给用户。

而作为LangGraph计算图的状态时，这三种方式都要求字段名称完全一致。只是处理方式不同。具体规则如下

3.1.4.1. 输入字段不匹配

TypedDict将输入字段视为字典的Key，不匹配时抛出KeyError异常。

dataclass将输入字段视为类的属性，不匹配时抛出TypeError（类型错误）异常

Pydantic对输入字段进行校验，不匹配时抛出ValidationError异常。

3.1.4.2. 节点返回字段不匹配

图节点返回的是对于状态的更新，如果返回字段和状态字段不匹配，上述三种Schema定义方式的行为是统一的：状态更新会被忽略

3.1.5. 推荐用法

在实际使用中，推荐优先使用TypedDict定义 LangGraph 状态图的 State Schema。

大多数官方案例也采用TypedDict方式定义状态Schema。这种方式写法简洁、结构清晰，能够直接描述状态中包含哪些字段，以及每个字段对应的数据类型，非常适合用于定义图运行过程中的共享状态。

相比普通dict，TypedDict可以提供更明确的字段约束和类型提示；相比dataclass，TypedDict更贴近 LangGraph 中状态的更新方式，因为节点通常返回的是表示“部分状态更新”的字典，而不是完整对象；相比Pydantic BaseModel，它又更加轻量，不会引入额外的数据校验开销。因此，在没有复杂校验需求的情况下，TypedDict是定义 LangGraph State Schema 的首选方式。

3.2. State Reducer

3.2.1. 什么是State Reducer

State Reducer 是 LangGraph 中用于合并状态更新的核心机制。在 LangGraph 的StateGraph中，每个节点可以读取和写入共享状态，而 Reducer 定义了如何将多个节点对同一状态键的更新合并

Reducer 的核心特征：

• 函数签名：(Value, Value) -> Value，接收当前值和更新值，返回合并后的新值
• 注解定义：通过Annotated[Type, reducer_function]为状态键指定 Reducer
• 默认行为：未指定 Reducer 的状态键使用覆盖策略（Last-Write-Wins）

3.2.2. 如何定义Reducer

3.2.2.1. 定义Reducer函数

Reducer 本质上是一个二元合并函数，用于定义当同一个字段产生多个更新值时，LangGraph 应该如何将这些值合并为一个最终结果。

函数签名：(Value, Value) -> Value

示例代码如下：

# left 从最开始的位置合并到当前节点的值 # right 当前节点的值 # 返回值 合并后的值 def my_reducer(left: list[str], right: list[str]) -> list[str]: return left+right # 1.node1 运行之后的值 left_state=["start","node_1 执行完"] right_state=["node_2 执行完"] merged=my_reducer(left_state,right_state) print(merged)

其中，my_reducer用于处理list[str]类型的数据。它接收两个列表参数：

• left：当前已累计的状态值；
• right：本次待合并的新值。

函数内部通过left + right将两个列表合并，并返回合并后的结果。

因此，该 Reducer 的作用是：当某个状态字段存在多次列表更新时，将这些列表内容追加合并，而不是直接覆盖原值。

运行结果如下

['start', 'node_1 执行完', 'node_2 执行完']

3.2.2.2. 将Reducer和状态字段关联

在 LangGraph 中，Reducer 通常通过 Python 的typing.Annotated与状态字段进行关联。

Annotated[]是 Python 提供的一种类型注解扩展机制，用于在原始类型之外附加额外的元数据信息。需要注意的是，Annotated[]本身并不规定这些元数据的具体含义，它只负责在类型注解中保留这些信息。

严格来说，Annotated的第一个参数是被注解的原始类型，后续参数是附加的元数据。至于这些元数据表示什么、如何解析，则由使用它的框架或工具自行决定。

在 LangGraph 中，框架利用这一机制，将状态字段的类型和 Reducer 规则同时声明在字段定义中。其基本形式如下：

Annotated[Type, reducer_function]

其中：

• Type：表示状态字段的数据类型；
• reducer_function：表示该字段对应的 Reducer 函数。

示例代码如下：

from typing import TypedDict, Annotated class OverAllState(TypedDict): logs: Annotated[list[str], my_reducer] cur_id: str

在上述代码中：

• logs字段的类型是list[str]；
• my_reducer是与logs字段关联的 Reducer 函数；
• 当多个节点同时更新logs字段时，LangGraph 会使用my_reducer将多个列表合并；
• cur_id字段没有指定 Reducer，因此采用默认更新规则。

3.2.2.3. 常用内置Reducer函数

1. operator.add

operator.add是 Python 内置的加法操作函数，底层由 C 实现

它接收两个参数，等价于a（第一个参数）+b（第二个参数）

代码如下

from operator import add print(f"{add(1,2) = }") print(f"{add([1,2], [3,4]) = }") print(f"{add(['a','b'], ['c']) = }")

输出如下

add(1,2) = 3 add([1,2], [3,4]) = [1, 2, 3, 4] add(['a','b'], ['c']) = ['a', 'b', 'c']

2. langgraph.graph.message.add_messages

add_messages是 LangGraph 中专用于合并消息列表的 Reducer 函数，常用于维护对话历史类的状态字段。其函数签名如下

add_messages是 LangGraph 中专用于合并消息列表的 Reducer 函数，常用于维护对话历史类的状态字段。其函数签名如下：

def add_messages( left: Messages, right: Messages, *, format: Literal["langchain-openai"] | None = None, ) -> Messages: ... return merged

参数说明：

• left：状态中已有的消息列表；
• right：当前节点返回的消息更新值；
• format：可选参数，用于指定返回消息的格式，通常无需手动设置。

left与right的类型均为Messages。Messages可以理解为 LangChain 消息对象的列表，其中每个元素都是BaseMessage或其子类的实例，常见子类包括：

• HumanMessage：用户的输入消息；
• AIMessage：AI 的回复消息；
• SystemMessage：系统提示消息；
• ToolMessage：工具调用的结果消息。

add_messages处理的是对话消息序列，而非普通的字符串列表。

BaseMessage包含一个可选的id属性，用于唯一标识一条消息。add_messages在合并left与right时，不是简单地执行列表拼接，而是依据消息的id进行合并：

• 若right中的某条消息的id在left中不存在，则将该消息追加到结果列表末尾；
• 若right中的某条消息的id与left中已有消息的id相同，则使用right中的新消息替换left中的旧消息。

因此，add_messages的作用可以概括为：在保留历史消息的基础上追加新消息，并允许通过相同的消息id覆盖已有消息。

需要特别说明，add_messages并非简单地对left与right求“并集”。更准确地说，它是一个基于消息id的消息列表合并函数：既支持追加新消息，也支持更新已有消息。

可以理解为：

merged = left + right

但若right中存在与left相同id的消息，则最终结果中不会出现重复消息，而是用right中的消息覆盖left中对应的旧消息。

示例代码如下

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langgraph.graph import add_messages left_state=[ SystemMessage(content="你是一个专业的翻译",id="1"), HumanMessage(content="你好",id="2"), AIMessage(content="你好，我是AI助手",id="3"), ] right_state=[ HumanMessage(content="我是老王，你是小王",id="2"), AIMessage(content="好的，我记住了",id="3"), HumanMessage(content="你是谁？",id="4"), AIMessage(content="我是小王",id="5"), ] merged=add_messages(left_state,right_state) print(merged)

输出如下

[SystemMessage(content='你是一个专业的翻译', additional_kwargs={}, response_metadata={}, id='1'), HumanMessage(content='我是老王，你是小王', additional_kwargs={}, response_metadata={}, id='2'), AIMessage(content='好的，我记住了', additional_kwargs={}, response_metadata={}, id='3', tool_calls=[], invalid_tool_calls=[]), HumanMessage(content='你是谁？', additional_kwargs={}, response_metadata={}, id='4'), AIMessage(content='我是小王', additional_kwargs={}, response_metadata={}, id='5', tool_calls=[], invalid_tool_calls=[])]

3.2.3. 默认行为

如果某个 State 字段没有显式定义Reducer，LangGraph 会使用默认的状态更新行为：后一次更新值会覆盖该字段原有的状态值。

换句话说，当节点返回的更新结果中包含某个字段时，如果该字段没有配置 Reducer，LangGraph 不会对新旧值进行追加、合并或累加，而是直接使用本次返回的新值替换原来的旧值。

示例代码如下：

from langgraph.graph import StateGraph, START, END from typing import TypedDict class OverAllState(TypedDict): logs: list[str] id: str def node_a(state: OverAllState): return { "logs": ["node_a"], "id": "node_a" } def node_b(state: OverAllState): return { "logs": ["node_b"], "id": "node_b" } 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('=' * 30, '-> result <-', '=' * 30) print(result)

输出如下

============================== -> result <- ============================== {'logs': ['node_b'], 'id': 'node_b'}

可以看到，logs字段和id字段都没有定义Reducer，因此，节点返回的新值会覆盖初始状态中的旧值，图运行结果中的状态值和最后一次更新保持一致。