状态管理
在 LangGraph 中，状态（State）是连接所有节点的核心。有效的状态管理是构建复杂、可靠的智能体的关键。
📊 状态的核心作用
- 数据传递: 在图的不同节点之间传递数据。
- 上下文保持: 记录对话历史、中间结果和智能体上下文。
- 流程控制: 状态中的值可以用于条件分支，控制执行流程。
- 持久化: 状态可以被持久化，实现可中断和恢复的工作流。
🏗️ 定义状态
LangGraph 使用 TypedDict 来定义状态，这提供了类型提示和静态分析的好处。
python
from typing import TypedDict, List, Optional
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
# 必需字段
messages: List[BaseMessage]
# 可选字段
user_info: Optional[dict]
# 中间结果
search_results: List[str]
# 流程控制
next_action: str最佳实践
- 明确类型: 尽可能使用明确的类型，避免使用
Any。 - 可选字段: 对非必需字段使用
Optional。 - 逻辑分组: 将相关的字段组织在一起。
🔄 更新状态
节点函数通过返回一个字典来更新状态。LangGraph 会将返回的字典与当前状态合并。
python
def search_node(state: AgentState) -> dict:
query = state["messages"][-1].content
# 模拟搜索
results = [f"Result for {query}"]
# 只返回需要更新的字段
return {"search_results": results}更新机制
- 覆盖: 如果返回的键已存在，其值将被覆盖。
- 追加: 对于列表，默认是覆盖。如果需要追加，需要手动处理。
列表追加示例
python
from typing import List
from langgraph.graph import StateGraph, END
class ListState(TypedDict):
items: List[int]
def add_item(state: ListState) -> dict:
new_item = len(state["items"]) + 1
# 手动追加
return {"items": state["items"] + [new_item]}
workflow = StateGraph(ListState)
# ...📦 MessageGraph
对于聊天机器人等以消息为中心的应用，MessageGraph 简化了状态管理。
- 自动管理
messages:MessageGraph的状态隐式包含一个messages列表。 - 自动追加: 节点返回的消息会自动追加到
messages列表中。
python
from langgraph.graph import MessageGraph
from langchain_core.messages import HumanMessage, AIMessage
def chatbot_node(state: List[HumanMessage]) -> AIMessage:
# state 直接是消息列表
user_message = state[-1].content
return AIMessage(content=f"你说了: {user_message}")
graph = MessageGraph()
graph.add_node("chatbot", chatbot_node)
graph.set_entry_point("chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
app.invoke(HumanMessage("你好"))
# 返回: [HumanMessage(...), AIMessage(...)]🔗 状态与流程控制
状态中的值是实现条件路由的关键。
python
def check_results(state: AgentState) -> str:
if state["search_results"]:
return "process_results"
else:
return "ask_for_clarification"
workflow.add_conditional_edges(
"search_node",
check_results,
{
"process_results": "process_node",
"ask_for_clarification": "clarify_node",
},
)💾 持久化
LangGraph 支持状态的持久化，这对于长时间运行或可中断的工作流至关重要。
配置检查点
python
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用 SQLite 作为后端
memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)调用可恢复的图
python
# 定义会话 ID
config = {"configurable": {"thread_id": "my-session-1"}}
# 第一次调用
app.invoke({"messages": [("user", "你好")]}, config)
# 第二次调用，状态会自动恢复
app.invoke({"messages": [("user", "今天天气怎么样")]}, config)支持的后端
MemorySaver: 内存中，用于测试。SqliteSaver: SQLite 数据库。PostgresSaver: PostgreSQL 数据库。RedisSaver: Redis 缓存。
🎯 总结
TypedDict: 定义清晰、类型安全的状态。- 节点返回: 节点通过返回字典来更新状态。
MessageGraph: 简化聊天应用的状态管理。- 条件路由: 使用状态值来控制流程。
- 持久化: 使用检查点实现可恢复的工作流。
💡 下一步: 学习 工具集成，了解如何让智能体调用外部工具并管理相关状态。