LangGraph之State的定义

在 LangGraph（LangChain 生态中的一个用于构建状态机和有向无环图工作流的库）中，State（状态）是整个工作流的核心数据结构。它用于在节点（Node）之间传递信息、维护上下文，并驱动整个图的执行流程。

LangGraph 中 State 的定义主要有以下几种方式，每种适用于不同的使用场景和复杂度需求：

一、TypedDict（推荐方式）

这是 LangGraph 官方最推荐的方式，尤其适用于 Python 3.8+（需安装typing_extensions）。

特点：

• 类型安全：通过类型注解明确每个字段的数据类型。
• 可读性强：清晰地表达状态结构。
• 支持部分更新（Partial Update）：在节点函数中只需返回需要更新的字段，LangGraph 会自动合并到全局状态中。
• 与 Pydantic 兼容性好（虽然不是必须用 Pydantic）。

示例：

fromtypingimportTypedDict,ListclassAgentState(TypedDict):messages:List[str]current_step:struser_query:strresult:str

然后在构建图时：

fromlanggraph.graphimportStateGraph graph=StateGraph(AgentState)

⚠️ 注意：所有字段都必须有类型注解，否则 LangGraph 无法正确处理状态更新。

为什么推荐使用 TypedDict？

LangGraph 需要知道：

• 状态有哪些字段？
• 每个字段的类型是什么？（用于调试、序列化、工具调用等）
• 哪些字段可以被节点更新？

而TypedDict的__annotations__属性正好提供了这些元信息：

print(AgentState.__annotations__)# 输出：# {'messages': list[str], 'user_query': str, 'current_step': str, 'result': typing.Optional[str]}

LangGraph 在构建图时会读取这些注解，从而：

• 验证节点返回的更新是否合法（不能返回未定义的字段）
• 支持 IDE 自动补全和类型检查（如 PyCharm、VS Code + Pylance）
• 实现安全的部分状态合并（partial update）

✅ 所有字段都必须有类型注解！否则 LangGraph 会报错或忽略该字段。

messages字段的更新机制

假设我们这样定义状态：

fromtypingimportTypedDict,Annotated,Sequencefromlangchain_core.messagesimportBaseMessageclassState(TypedDict):messages:Annotated[Sequence[BaseMessage],...]# 注意这里用了 Annotated

💡 实际项目中，messages通常用list[BaseMessage]或Sequence[BaseMessage]，并配合Annotated来定制更新行为。

更新机制 1：覆盖式更新（Replace）

这是默认行为：节点返回一个新列表，完全替换原有messages。

defadd_greeting(state:State)->dict:return{"messages":[HumanMessage(content="Hi!")]}

✅ 效果：原 messages 被清空，只保留新列表。
⚠️ 风险：容易丢失历史消息！

更新机制 2：追加式更新（Append）—— 推荐！

LangGraph 支持通过operator.add或自定义 reducer 实现“追加”而非“覆盖”。

方法 A：使用Annotated+operator.add

fromoperatorimportaddfromtypingimportAnnotated,Sequencefromlangchain_core.messagesimportBaseMessageclassState(TypedDict):messages:Annotated[Sequence[BaseMessage],add]

现在，当节点返回{"messages": [new_msg]}时，LangGraph 会执行：

state["messages"]=add(state["messages"],[new_msg])# 等价于：state["messages"] + [new_msg]

✅ 效果：新消息被追加到末尾，历史保留！
✅ 这是构建多轮对话 Agent 的标准做法。

方法 B：自定义 Reducer 函数

可以定义更复杂的合并逻辑：

defmerge_messages(old:list,new:list)->list:# 例如：去重、截断、过滤等combined=old+newreturncombined[-10:]# 只保留最近10条classState(TypedDict):messages:Annotated[list[BaseMessage],merge_messages]

更新机制 3：部分更新 + 追加组合

你可以在一个节点中同时更新多个字段，其中messages追加，其他字段覆盖：

defprocess_step(state:State)->dict:new_msg=AIMessage(content="I'm processing your request.")return{"messages":[new_msg],# 因为用了 Annotated[add]，所以是追加"current_step":"processing",# 覆盖"attempts":state.get("attempts",0)+1# 自增}

LangGraph 会分别对每个字段应用其对应的 reducer（默认是覆盖，除非用Annotated指定）。

更新机制 4：空更新或不更新

如果节点不返回messages，则保持原样：

deflog_step(state:State)->dict:print("Current step:",state["current_step"])return{}# messages 不变

最佳实践建议

✅ 1. 始终用Annotated[Sequence[BaseMessage], operator.add]定义 messages

fromtypingimportTypedDict,Annotated,Sequencefromlangchain_core.messagesimportBaseMessagefromoperatorimportaddclassAgentState(TypedDict):messages:Annotated[Sequence[BaseMessage],add]user_input:strstep:str

✅ 2. 节点函数只返回“变化的部分”

不要手动拼接整个 messages 列表，让 LangGraph 处理合并：

# ❌ 不推荐：手动拼接defbad_node(state):return{"messages":state["messages"]+[new_msg]}# ✅ 推荐：只返回新增消息defgood_node(state):return{"messages":[new_msg]}

✅ 3. 使用BaseMessage子类（如HumanMessage,AIMessage）

这能让你的消息携带角色信息，便于 LLM 理解上下文：

fromlangchain_core.messagesimportHumanMessage,AIMessage# 在节点中return{"messages":[AIMessage(content="Hello! I'm an AI.")]}

完整示例：带消息追加的简单 Agent

fromtypingimportTypedDict,Annotated,Sequencefromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessagefromlanggraph.graphimportStateGraph,START,ENDfromoperatorimportaddclassState(TypedDict):messages:Annotated[Sequence[BaseMessage],add]defrespond(state:State)->dict:last_msg=state["messages"][-1].content reply=f"You said:{last_msg}"return{"messages":[AIMessage(content=reply)]}graph=StateGraph(State)graph.add_edge(START,"respond")graph.add_node("respond",respond)graph.add_edge("respond",END)app=graph.compile()# 运行result=app.invoke({"messages":[HumanMessage(content="Hi")]})print(result["messages"])# 输出包含 HumanMessage("Hi") 和 AIMessage("You said: Hi")

总结

二、Pydantic BaseModel（实验性支持）

虽然 LangGraph 最初是围绕 TypedDict 设计的，但社区和官方也在逐步支持 Pydantic 模型。

特点：

• 更强大的验证能力（如字段校验、默认值、嵌套模型等）。
• 支持序列化/反序列化（对持久化状态很有用）。
• 目前在 LangGraph 中的支持不如 TypedDict 成熟，可能存在兼容性问题（截至 LangGraph v0.1.x）。

示例：

frompydanticimportBaseModelfromtypingimportListclassAgentState(BaseModel):messages:List[str]=[]current_step:str="start"user_query:strresult:str=""

🔍 注意：使用 Pydantic 模型时，某些 LangGraph 功能（如 partial update）可能不会按预期工作，因为 LangGraph 内部依赖dict.update()行为，而 Pydantic 对象不是普通 dict。

建议：除非你有强验证或序列化需求，否则优先使用 TypedDict。

三、普通字典（不推荐）

理论上你可以直接用dict作为状态，但 LangGraph 要求状态类必须可被 introspect（内省），以便知道有哪些字段可以更新。

为什么不推荐？

• 没有类型提示，容易出错。
• 无法利用 LangGraph 的自动合并机制（因为不知道哪些是合法字段）。
• 在大型项目中难以维护。

LangGraph 在内部会检查状态类是否为TypedDict或具有__annotations__，普通 dict 通常会导致运行时错误。

四、自定义类（带__annotations__）

如果你不想用 TypedDict，也可以定义一个普通类并手动添加类型注解：

classAgentState:messages:list[str]current_step:struser_query:strresult:str

这种方式在技术上可行，因为 LangGraph 会读取AgentState.__annotations__来获取字段信息。

但存在风险：

• 缺少 TypedDict 的语义保证（TypedDict 是专为结构化字典设计的）。
• 无法直接实例化为字典，可能在节点函数中造成混淆。
• 不符合官方最佳实践。

关键机制：Partial State Updates（部分状态更新）

无论使用哪种方式，LangGraph 的核心优势之一是允许节点只返回状态的一部分，系统会自动 merge 到全局状态中。

例如：

defstep_1(state:AgentState)->dict:return{"current_step":"processed","result":"hello"}

LangGraph 会将返回的 dict 合并到当前 state 中，其他字段保持不变。

✅ 这要求状态定义必须是“字段已知”的结构（如 TypedDict），否则无法安全合并。

总结对比表

自此，本文分享到此结束！！！