AI创想

标题: LangGraph 状态管理全解析：从定义到高级应用 [打印本页]
作者: 创想小编    时间: 昨天 08:16
标题: LangGraph 状态管理全解析：从定义到高级应用
作者：佑瞻
在构建复杂的语言模型应用时，状态管理往往是容易被忽视却又至关重要的一环。当我们使用 LangGraph 搭建智能交互系统时，如何高效定义、更新和管理状态直接影响着系统的灵活性和可维护性。今天我们就来深入探讨 LangGraph 中状态管理的核心机制，从基础定义到高级特性，带大家全面掌握这一关键技术。
一、状态的基础定义与更新机制

1.1 状态结构的定义方式

在 LangGraph 中，状态可以通过多种方式定义，包括 TypedDict、Pydantic 模型或数据类。我们先以 TypedDict 为例，看看如何构建一个基础的状态结构：
python
运行
  1. from langchain_core.messages import AnyMessage
  2. from typing_extensions import TypedDict
  4. class State(TypedDict):
  5.     messages: list[AnyMessage]
  6.     extra_field: int
复制代码
这个状态结构包含两个字段：一个消息列表和一个整数类型的额外字段。默认情况下，图的输入和输出模式由状态决定，这为大多数 LLM 应用提供了灵活的表述方式。
1.2 状态更新的核心原则

当我们需要更新状态时，节点函数应返回新的状态更新，而不是直接修改原始状态。来看一个简单的例子：
python
运行
  1. from langchain_core.messages import AIMessage
  3. def node(state: State):
  4.     messages = state["messages"]
  5.     new_message = AIMessage("Hello!")
  6.     return {"messages": messages + [new_message], "extra_field": 10}
复制代码
这个节点函数向消息列表中添加了一条新消息，并设置了额外字段的值。接下来我们需要将这个节点添加到状态图中：
python
运行
  1. from langgraph.graph import StateGraph
  3. builder = StateGraph(State)
  4. builder.add_node(node)
  5. builder.set_entry_point("node")
  6. graph = builder.compile()
复制代码
调用这个图时，我们可以传入初始状态并获取更新后的完整状态：
python
运行
  1. from langchain_core.messages import HumanMessage
  3. result = graph.invoke({"messages": [HumanMessage("Hi")]})
复制代码
输出结果会包含我们添加的两条消息和设置的额外字段值。这里需要注意的是，我们只需更新状态的部分键，调用结果会返回完整的状态。
二、使用 Reducers 优化状态更新

2.1 Reducers 的基本概念

每个状态键都可以有独立的 reducer 函数，用于控制节点更新的应用方式。如果没有显式指定 reducer，默认会覆盖该键的值。对于 TypedDict 状态，我们可以通过注解为字段指定 reducer 函数：
python
运行
  1. from typing_extensions import Annotated
  2. from operator import add  # 也可以自定义add函数
  4. class State(TypedDict):
  5.     messages: Annotated[list[AnyMessage], add]
  6.     extra_field: int
复制代码
这样一来，节点函数可以简化为：
python
运行
  1. def node(state: State):
  2.     new_message = AIMessage("Hello!")
  3.     return {"messages": [new_message], "extra_field": 10}
复制代码
2.2 消息状态的特殊处理

在实际应用中，消息列表的更新有更多需要考虑的地方，比如更新现有消息或接受简化的消息格式。LangGraph 提供了内置的 add_messages reducer 来处理这些情况：
python
运行
  1. from langgraph.graph.message import add_messages
  3. class State(TypedDict):
  4.     messages: Annotated[list[AnyMessage], add_messages]
  5.     extra_field: int
复制代码
甚至可以直接使用预构建的 MessagesState：
python
运行
  1. from langgraph.graph import MessagesState
  3. class State(MessagesState):
  4.     extra_field: int
复制代码
三、高级状态管理特性

3.1 定义独立的输入输出模式

默认情况下，StateGraph 使用单一模式，但我们也可以定义独立的输入和输出模式：
python
运行
  1. from langgraph.graph import StateGraph, START, END
  3. # 输入模式
  4. class InputState(TypedDict):
  5.     question: str
  6. # 输出模式
  7. class OutputState(TypedDict):
  8.     answer: str
  9. # 整体模式
  10. class OverallState(InputState, OutputState):
  11.     pass
  13. def answer_node(state: InputState):
  14.     return {"answer": "bye", "question": state["question"]}
  16. builder = StateGraph(OverallState, input=InputState, output=OutputState)
  17. builder.add_node(answer_node)
  18. builder.add_edge(START, "answer_node")
  19. builder.add_edge("answer_node", END)
  20. graph = builder.compile()
  22. print(graph.invoke({"question": "hi"}))  # 输出: {'answer': 'bye'}
复制代码
这里输出结果只包含定义的输出模式字段，实现了数据的有效过滤。
3.2 节点间传递私有状态

有时我们需要节点间交换中间数据，这些数据不需要作为图的主要模式部分：
python
运行
  1. from langgraph.graph import StateGraph, START, END
  3. # 公共状态
  4. class OverallState(TypedDict):
  5.     a: str
  6. # 节点1的输出（包含私有数据）
  7. class Node1Output(TypedDict):
  8.     private_data: str
  10. def node_1(state: OverallState) -> Node1Output:
  11.     return {"private_data": "set by node_1"}
  13. # 节点2的输入（只需要私有数据）
  14. class Node2Input(TypedDict):
  15.     private_data: str
  17. def node_2(state: Node2Input) -> OverallState:
  18.     return {"a": "set by node_2"}
  20. def node_3(state: OverallState) -> OverallState:
  21.     return {"a": "set by node_3"}
  23. builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3])
  24. builder.add_edge(START, "node_1")
  25. graph = builder.compile()
  27. response = graph.invoke({"a": "set at start"})
  28. print(f"Output: {response}")  # 输出: {'a': 'set by node_3'}
复制代码
这个例子中，私有数据只在节点 1 和节点 2 之间传递，节点 3 只能访问公共状态。
3.3 使用 Pydantic 模型进行状态验证

Pydantic 模型可以为状态添加运行时验证：
python
运行
  1. from pydantic import BaseModel
  2. from langgraph.graph import StateGraph, START, END
  4. class OverallState(BaseModel):
  5.     a: str
  7. def node(state: OverallState):
  8.     return {"a": "goodbye"}
  10. builder = StateGraph(OverallState)
  11. builder.add_node(node)
  12. builder.add_edge(START, "node")
  13. builder.add_edge("node", END)
  14. graph = builder.compile()
  16. graph.invoke({"a": "hello"})  # 正常运行
  17. try:
  18.     graph.invoke({"a": 123})  # 会抛出异常
  19. except Exception as e:
  20.     print("异常信息:", e)
复制代码
这里需要注意，目前图的输出不会是 Pydantic 模型实例，运行时验证只针对节点输入，且错误跟踪不显示具体节点。
3.4 添加运行时配置

我们可以在调用图时配置参数，而不是将参数混入状态中：
python
运行
  1. from langchain_core.runnables import RunnableConfig
  2. from langgraph.graph import StateGraph, START, END
  4. # 配置模式
  5. class ConfigSchema(TypedDict):
  6.     my_runtime_value: str
  8. # 状态模式
  9. class State(TypedDict):
  10.     my_state_value: str
  12. def node(state: State, config: RunnableConfig):
  13.     value = config["configurable"]["my_runtime_value"]
  14.     return {"my_state_value": 1 if value == "a" else 2}
  16. builder = StateGraph(State, config_schema=ConfigSchema)
  17. builder.add_node(node)
  18. builder.add_edge(START, "node")
  19. builder.add_edge("node", END)
  20. graph = builder.compile()
  22. print(graph.invoke({}, {"configurable": {"my_runtime_value": "a"}}))  # 输出: {'my_state_value': 1}
  23. print(graph.invoke({}, {"configurable": {"my_runtime_value": "b"}}))  # 输出: {'my_state_value': 2}
复制代码
3.5 节点重试策略

对于调用 API、数据库或 LLM 等可能失败的操作，我们可以为节点添加重试策略：
python
运行
  1. from langgraph.pregel import RetryPolicy
  3. builder.add_node(
  4.     "node_name",
  5.     node_function,
  6.     retry=RetryPolicy(),  # 使用默认重试策略
  7. )
复制代码
默认情况下，会重试除特定异常外的所有错误，对于 HTTP 请求库的异常，只重试 5xx 状态码。
3.6 节点缓存机制

为避免重复执行昂贵操作，我们可以为节点添加缓存：
python
运行
  1. from langgraph.types import CachePolicy
  2. from langgraph.cache.memory import InMemoryCache
  4. builder.add_node(
  5.     "node_name",
  6.     node_function,
  7.     cache_policy=CachePolicy(ttl=120),  # 缓存有效期120秒
  8. )
  10. graph = builder.compile(cache=InMemoryCache())  # 启用内存缓存
复制代码
也可以使用 SqliteCache 实现持久化缓存。
总结与实践建议

通过本文，我们全面了解了 LangGraph 中状态管理的各个方面，从基础的状态定义和更新，到使用 reducers 优化更新过程，再到高级的输入输出模式定义、私有状态传递、运行时配置、重试策略和节点缓存等特性。这些功能为构建复杂的语言模型应用提供了强大的支持。
如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148798756



欢迎光临 AI创想 (https://www.llms-ai.com/) Powered by Discuz! X3.4