AI Agent 的状态管理：工作流与图结构

引言：为什么状态管理是 Agent 的基石

在传统的 AI 应用中，模型通常执行一次性的推理：输入问题，输出答案，任务即告结束。然而，AI Agent 作为能够自主规划、执行多步骤任务的智能系统，其内部状态会随着任务的推进而不断演化。一个 Agent 在执行过程中，可能需要调用外部工具、检索知识库、等待用户反馈，甚至在多个步骤之间回溯重试。如果缺乏良好的状态管理机制，Agent 的复杂性将迅速失控，导致调试困难、行为不可预测。 想象一下，一个支持多轮对话的电商 Agent，需要经历「理解意图 → 检索商品 → 询问偏好 → 确认下单」多个阶段。在每个阶段之间，Agent 必须记住用户的偏好、已检索的商品列表、当前对话的上下文。如果没有结构化的状态管理，这种复杂度将呈指数级增长。

从有限状态机到图结构

有限状态机（FSM）的启示

有限状态机（Finite State Machine, FSM）是计算机科学中经典的控制流模型。它将系统行为抽象为若干状态（State）和状态转移（Transition），每个状态转移由特定的事件触发。FSM 的优势在于： - 状态清晰：系统的每一个时刻都明确处于某个状态，便于追踪和调试。 - 行为可控：状态转移是预定义、可预测的，避免了无限制的随机行为。 - 易于验证：由于状态数量有限，可以通过穷举或形式化方法验证系统的正确性。 对于简单 Agent（如单轮问答或顺序工具调用），FSM 足以胜任。但面对复杂任务，如多分支条件判断、循环回溯、并行执行时，传统 FSM 的线性结构会显得力不从心。此时，图结构（Graph Structure） 应运而生。

图结构：超越线性的状态表达

图结构将 Agent 的每个执行步骤视为节点（Node），将步骤之间的依赖关系视为边（Edge）。相比于 FSM 的线性转移，图结构天然支持： | 能力 | 图结构支持 | 传统 FSM 支持 | |------|------------|---------------| | 分支条件判断 | ✅ 多条出边 | ⚠️ 需扩展 | | 循环与回溯 | ✅ 有向环 | ⚠️ 需特殊处理 | | 并行执行 | ✅ 多分支同步 | ❌ 不支持 | | 动态路由 | ✅ 运行时决定下一路径 | ⚠️ 受限于预定义转移 | LangChain 团队的 LangGraph 正是将图结构理念引入 AI Agent 状态管理的代表性框架。它允许开发者将 Agent 的逻辑以图的方式显式建模，每一步的状态变化都能被追踪和干预。

LangGraph 框架核心概念

LangGraph 是构建在 LangChain 之上的状态驱动 Agent 框架，核心设计围绕三个概念展开。

1. State（状态）

State 是贯穿图执行全生命周期的数据容器。在 LangGraph 中，State 通常是一个 TypedDict，定义了每个节点共享和传递的数据结构：

from typing import TypedDict, List
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: List[BaseMessage]  # 对话历史
    next_node: str               # 下一个执行节点
    tool_calls: List[dict]       # 已调用的工具记录
    iteration: int               # 迭代次数，用于防循环

State 在节点之间传递时，每个节点可以读取并修改它，但只返回需要变更的部分字段。这种不可变快照的设计，使得状态回溯和调试变得简单。

2. Node（节点）

Node 是图中的执行单元，代表一个可调用函数。它接收当前 State，执行特定逻辑，返回更新后的 State 字段。在 Agent 中，常见的节点包括： - LLM 节点：调用大模型生成回复或决策 - 工具节点：调用外部 API（如搜索、数据库、计算器） - 路由节点：根据 State 条件决定下一步走向 - 条件节点：检查终止条件（如达到最大迭代次数、任务完成）

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def llm_node(state: AgentState) -> dict:
    """LLM 推理节点：根据当前消息生成响应"""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    """工具节点：模拟调用外部工具"""
    last_message = state["messages"][-1]
    # 假设工具返回计算结果
    tool_result = f"工具执行结果：{last_message.content}"
    from langchain_core.messages import AIMessage
    return {"messages": [AIMessage(content=tool_result)]}

3. Edge（边）与条件路由

Edge 定义了节点之间的流转关系。LangGraph 支持两种边： - 普通边：执行完节点 A 后，无条件进入节点 B - 条件边（Conditional Edge）：根据当前 State 动态决定下一个节点，实现灵活的分支控制

from langgraph.graph import StateGraph, END

条件路由函数：判断是否需要调用工具

def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] # 如果 LLM 要求调用工具，则路由到工具节点 if "tool" in last_message.content.lower(): return "tool_node" # 否则任务完成 return END

构建图

workflow = StateGraph(AgentState) workflow.add_node("llm", llm_node) workflow.add_node("tool", tool_node) workflow.set_entry_point("llm") workflow.add_conditional_edges( "llm", should_continue, {"tool_node": "tool", END: END} ) workflow.add_edge("tool", "llm") # 工具执行后返回 LLM

构建多步骤 Agent 工作流：实战示例

以下是一个完整的 LangGraph 多步骤 Agent 示例，演示如何构建一个「带工具调用的推理循环」：

from typing import TypedDict, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

========== 1. 定义状态结构 ==========

class AgentState(TypedDict): messages: List[BaseMessage] iteration: int

========== 2. 初始化 LLM ==========

llm = ChatOpenAI(model="gpt-4o", temperature=0)

========== 3. 定义节点 ==========

def agent_node(state: AgentState) -> dict: """Agent 推理节点""" response = llm.invoke(state["messages"]) return {"messages": [response], "iteration": state["iteration"] + 1} def search_tool(state: AgentState) -> dict: """模拟搜索工具""" last_msg = state["messages"][-1].content # 实际项目中这里调用搜索引擎 API result = f"[搜索结果] 关于 '{last_msg}' 的相关信息已获取。" return {"messages": [AIMessage(content=result)]} def calculator_tool(state: AgentState) -> dict: """模拟计算工具""" last_msg = state["messages"][-1].content # 实际项目中这里解析数学表达式并计算 result = f"[计算结果] 表达式 '{last_msg}' 的计算结果：42" return {"messages": [AIMessage(content=result)]}

========== 4. 条件路由逻辑 ==========

def r