LangGraph系列教程：基于状态构建上下文感知的AI系统

本文深入探讨LangGraph中的“状态”概念及其在AI工作流中的核心作用。通过基础状态（如计数器）和复杂状态（含消息历史）的定义，结合代码示例，演示如何通过函数式编程实现状态的不可变修改。然后进一步解析了如何构建简单和复杂的工作流图（Workflow），并以实际案例（如设备预测性维护、对话机器人）说明状态管理的实际应用。附带的完整Python代码提供了可落地的操作模板，适合开发者快速掌握LangGraph的状态管理机制。

LangGraph的状态是什么？

LangGraph 已成为创建循环代理式人工智能工作流的强大工具。LangGraph 中的一个关键概念是“状态”这一理念——它是基本的构建模块，使我们的 AI 系统能够在整个流程中保持和处理信息。

在AI系统处理数据时，LangGraph中的状态是一种维护和跟踪信息的方法。可以把它想象成系统的内存，允许它在工作流或图形的不同阶段移动时记住和更新信息。

基本状态定义

让我们从LangGraph中最简单的状态形式开始：

from typing import TypedDictclass BasicState(TypedDict):count: int

这个基本状态定义创建了一个简单的结构，它可以保存单个整数值，也可以是任何值。虽然看起来很简单，但这种类型的状态在许多情况下都非常有用，例如：

• 跟踪对话中的回合数
• 统计特定事件的发生次数
• 维持一个简单的分数、指标或AI输出

复杂状态结构

当我们进入更现实的应用程序时，我们通常需要更复杂的状态结构。LangGraph允许我们创建可以保存各种类型信息的复杂状态：

from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessageclass ComplexState(TypedDict):count: intmessages: Annotated[list[HumanMessage | AIMessage], add_messages]

这个复杂的状态不仅跟踪计数，而且还维护消息列表。Annotated类型提供了LangGraph用于特殊处理消息列表（元组）的附加元数据。这个结构特别适用于：

• 需要记住对话历史的聊天机器人
• 在多个交互中维护上下文的人工智能助手
• 需要跟踪定量（计数）和定性（消息）数据的系统

状态修改函数

一旦我们定义了状态结构，我们就需要修改它们的方法。在LangGraph中，我们通常创建新的状态对象，而不是修改现有的状态对象，并遵循不变性原则：

def increment_count(state: BasicState) -> BasicState:return BasicState(count=state["count"] + 1)def add_message(state: ComplexState, message: str, is_human: bool = True) -> ComplexState:new_message = HumanMessage(content=message) if is_human else AIMessage(content=message)return ComplexState(count=state["count"],messages=state["messages"] + [new_message])

这些函数演示了我们如何：

• 在我们的基本状态增加计数器
• 向我们的复杂状态添加新消息，区分人类和AI消息
• 创建反映这些变化的新状态对象

简单图状态

现在我们已经了解了state的基础知识以及如何修改它，让我们看看如何在LangGraph中使用state：

from langgraph.graph import StateGraph, ENDdef create_simple_graph():workflow = StateGraph(BasicState)def increment_node(state: BasicState):return {"count": state["count"] + 1}workflow.add_node("increment", increment_node)workflow.set_entry_point("increment")workflow.add_edge("increment", END)return workflow.compile()

这个简单的图表展示了LangGraph工作流的基本结构：

1. 我们使用BasicState创建一个StateGraph。
2. 我们定义了一个节点函数（increment_node）来修改状态。
3. 我们将这个节点添加到图中，并将其设置为入口点。
4. 我们从我们的节点到图的END创建一条边。
5. 最后，我们编译我们的工作流。

虽然这个图看起来很基本，但它可以作为更复杂工作流的基础。

复杂图状态

基于我们对简单图形的理解，让我们创建一个更高级的工作流：

def create_complex_graph():workflow = StateGraph(ComplexState)def process_message(state: ComplexState):last_message = state["messages"][-1].content if state["messages"] else "No messages yet"response = f"Received: {last_message}. Count is now {state['count'] + 1}"return {"count": state["count"] + 1,"messages": state["messages"] + [AIMessage(content=response)]}workflow.add_node("process", process_message)workflow.set_entry_point("process")workflow.add_edge("process", END)return workflow.compile()

这个更复杂的图表展示了我们如何：

• 使用我们的ComplexState来维护计数和消息历史
• 处理传入消息并生成响应
• 在每一步用新的信息更新我们的状态

这种类型的图形可以构成一个简单的聊天机器人或人工智能助手的基础，展示了LangGraph中状态的强大和灵活性。

在LangGraph中理解和有效地使用状态为创建复杂的、上下文感知的人工智能系统开辟了一个可能性的世界。从简单的计数器到复杂的会话代理，状态的概念允许我们构建可以轻松处理多步骤流程的人工智能工作流。

完整实战案例

改示例是使用 LangGraph实现状态管理的完整教学示例，包含 状态定义、状态修改函数 和 图（workflow）构建 的核心逻辑。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage# Step 1: Basic State Definition
class BasicState(TypedDict):count: int# Step 2: More Complex State
class ComplexState(TypedDict):count: intmessages: Annotated[list[HumanMessage | AIMessage], add_messages]# Step 3: State Modification Functions
def increment_count(state: BasicState) -> BasicState:return BasicState(count=state["count"] + 1)def add_message(state: ComplexState, message: str, is_human: bool = True) -> ComplexState:new_message = HumanMessage(content=message) if is_human else AIMessage(content=message)return ComplexState(count=state["count"],messages=state["messages"] + [new_message])# Step 4: Simple Graph with State
def create_simple_graph():workflow = StateGraph(BasicState)def increment_node(state: BasicState):return {"count": state["count"] + 1}workflow.add_node("increment", increment_node)workflow.set_entry_point("increment")workflow.add_edge("increment", END)return workflow.compile()# Step 5: More Complex Graph with State
def create_complex_graph():workflow = StateGraph(ComplexState)def process_message(state: ComplexState):last_message = state["messages"][-1].content if state["messages"] else "No messages yet"response = f"Received: {last_message}. Count is now {state['count'] + 1}"return {"count": state["count"] + 1,"messages": state["messages"] + [AIMessage(content=response)]}workflow.add_node("process", process_message)workflow.set_entry_point("process")workflow.add_edge("process", END)return workflow.compile()# Interactive Session
def run_interactive_session():print("Welcome to the Interactive LangGraph State Lesson!")print("\nStep 1: Basic State")basic_state = BasicState(count=0)print(f"Initial basic state: {basic_state}")print("\nStep 2: More Complex State")complex_state = ComplexState(count=0, messages=[])print(f"Initial complex state: {complex_state}")print("\nStep 3: State Modification")modified_basic = increment_count(basic_state)print(f"Modified basic state: {modified_basic}")modified_complex = add_message(complex_state, "Hello, LangGraph!")print(f"Modified complex state: {modified_complex}")print("\nStep 4: Simple Graph with State")simple_graph = create_simple_graph()result = simple_graph.invoke(BasicState(count=0))print(f"Simple graph result: {result}")print("\nStep 5: Complex Graph with State")complex_graph = create_complex_graph()initial_state = ComplexState(count=0, messages=[HumanMessage(content="Hello, LangGraph!")])result = complex_graph.invoke(initial_state)print(f"Complex graph result: {result}")if __name__ == "__main__":run_interactive_session()

这段代码是使用 LangGraph（一个开源框架）实现状态管理的完整教学示例，包含 状态定义、状态修改函数 和 图（workflow）构建 的核心逻辑。以下是逐段详细解释：

1. 导入必要的模块

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

• TypedDict: 定义结构化的字典类型（类似数据类）。
• Annotated: 类型注解工具，用于增强类型约束（如 messages 字段必须包含特定类型的消息）。
• StateGraph, END: LangGraph 核心组件，用于构建状态图。
• add_messages: 装饰器，标记可添加消息的字段。
• HumanMessage, AIMessage: 表示人类用户消息和 AI 生成的响应消息。

2. 定义基本状态（BasicState）

class BasicState(TypedDict):count: int

• 简单状态类，仅包含一个整数 count，用于演示基础状态管理。

3. 定义复杂状态（ComplexState）

class ComplexState(TypedDict):count: intmessages: Annotated[list[HumanMessage | AIMessage], add_messages]

• 复杂状态类，包含：
  • count: 整数计数器。
  • messages: 一个消息列表，存储交互记录（人类消息或 AI 响应）。
  • Annotated[list[...], add_messages]: 类型注解，强制 messages 字段遵循 LangGraph 的消息规范。

4. 状态修改函数

(1) increment_count 函数

def increment_count(state: BasicState) -> BasicState:return BasicState(count=state["count"] + 1)

• 输入一个 BasicState，将 count 加 1 并返回新状态。

(2) add_message 函数

def add_message(state: ComplexState, message: str, is_human: bool = True) -> ComplexState:new_message = HumanMessage(content=message) if is_human else AIMessage(content=message)return ComplexState(count=state["count"],messages=state["messages"] + [new_message])

• 输入 ComplexState 和消息内容，根据 is_human 参数判断消息类型（人类或 AI），将新消息追加到 messages 列表中。

5. 构建简单状态图（create_simple_graph）

def create_simple_graph():workflow = StateGraph(BasicState)def increment_node(state: BasicState):return {"count": state["count"] + 1}workflow.add_node("increment", increment_node)workflow.set_entry_point("increment")workflow.add_edge("increment", END)return workflow.compile()

• 目标: 创建一个仅包含 count 增加的简单流程。
• 节点 (increment_node): 处理输入状态，将 count +1。
• 入口点: 默认从 increment 节点开始。
• 边: 直接连接到结束节点 END。
• 编译: 将流程图转换为可执行的逻辑。

6. 构建复杂状态图（create_complex_graph）

def create_complex_graph():workflow = StateGraph(ComplexState)def process_message(state: ComplexState):last_message = state["messages"][-1].content if state["messages"] else "No messages yet"response = f"Received: {last_message}. Count is now {state['count'] + 1}"return {"count": state["count"] + 1,"messages": state["messages"] + [AIMessage(content=response)]}workflow.add_node("process", process_message)workflow.set_entry_point("process")workflow.add_edge("process", END)return workflow.compile()

• 目标: 创建处理消息并维护对话历史的复杂流程。
• 节点 (process_message):
  • 读取当前最新消息（last_message）。
  • 生成响应（包含 count +1 的结果）。
  • 更新 count 和追加 AI 响应到 messages。
• 流程: 每次调用 process 节点后直接结束。

7. 交互式会话运行

def run_interactive_session():print("Welcome to the Interactive LangGraph State Lesson!")# 分步骤演示 BasicState 和 ComplexState 的操作# ...（省略中间打印语句）# 调用简单图和复杂图simple_result = simple_graph.invoke(BasicState(count=0))complex_result = complex_graph.invoke(ComplexState(count=0, messages=[HumanMessage(content="Hello, LangGraph!")]))print(f"Simple graph result: {simple_result}")print(f"Complex graph result: {complex_result}")

• 功能: 运行教学示例，展示不同状态和流程的实际效果。
• 简单图调用: 初始状态为 count=0，执行后结果为 count=1。
• 复杂图调用: 初始包含一条人类消息，处理后生成 AI 响应并更新 count。

最后总结

LangGraph的状态管理是其构建上下文感知AI系统的核心机制。状态作为系统“内存”，贯穿工作流始终，确保信息在不同阶段间连贯传递。通过TypedDict定义结构化状态（如BasicState追踪计数、ComplexState维护消息历史），开发者能够灵活处理简单数值统计与多模态交互场景。状态的修改遵循函数式编程原则——每次操作均生成新状态对象而非直接修改旧状态，保障数据一致性与可追溯性。

实际应用中，开发者可通过构建简单图（如计数器递增）验证基础逻辑，再进阶到复杂图（如消息处理与动态响应），模拟对话机器人或自动化流程。例如，制造业中的设备预测性维护场景中，状态可整合设备运行数据与历史故障记录，驱动智能决策；而政务服务领域的政策咨询机器人则依赖状态维护用户提问轨迹与政策匹配结果。代码实践部分通过完整的Python示例（包括基础/复杂状态定义、工作流构建与交互式运行），为开发者提供了可直接复用的模板，降低了从理论到落地的门槛。

LangGraph的状态管理不仅解决了多步骤流程中的上下文丢失问题，还通过类型注解、不可变状态等设计，增强了系统的可靠性与扩展性，是构建高效AI代理的关键支撑。