LangGraph：构建有状态的图驱动 AI 代理

FreeGuideOnline 最新 2026-06-14

LangGraph 状态图代理：构建有状态的图驱动 AI 代理

什么是 LangGraph？

LangGraph 是一个专为构建有状态、多参与者的 AI 代理而设计的框架。它基于图的概念，将代理的行为、决策流程以及状态管理以节点和边的形式进行建模。与传统的“单步推理”不同，LangGraph 让你的代理能够记住对话历史、维护内部状态，并在复杂的、多步骤的任务中做出连贯的决策。

简单来说，LangGraph 帮助你把一个简单的语言模型调用，转变为能够自主规划、执行和记忆的图驱动代理。

为什么需要状态图代理？

大多数基于大型语言模型（LLM）的代理都存在一个痛点：状态丢失。每次调用时，代理都需要重新理解上下文，无法自然地记住之前的步骤或用户偏好。LangGraph 通过以下方式解决这个问题：

  • 显式状态管理：图中的每个节点都可以读取和更新一个共享的状态对象，所有信息都存储在图中。
  • 循环与分支：图结构天然支持循环（例如：自我修正）和条件分支（例如：根据中间结果决定下一步）。
  • 可持久化：状态可以被序列化并存储，支持长时间运行的工作流或人机交互中的暂停与恢复。

核心概念：节点、边与状态图

LangGraph 代理的本质是一个状态图（StateGraph）。要理解和创建它，需要掌握三个核心概念。

节点（Node）

节点是执行的逻辑单元，通常是一个 Python 函数。它接收当前状态，处理后返回一个状态更新。常见的节点包括：

  • 代理节点：调用 LLM 进行决策或生成。
  • 工具节点：执行具体操作，如搜索、计算或访问数据库。
  • 条件判断节点：评估当前状态，决定下一步走向。

边（Edge）

边定义了节点之间的流转规则。LangGraph 支持三种边：

  • 正常边（Normal Edge）：从一个节点直接连接到另一个节点，无分支。
  • 条件边（Conditional Edge）：根据当前状态动态选择下一个节点。需要一个路由函数，返回下一个节点的名称。
  • 入口点与结束点：分别用 set_entry_point 和 finish_point 标记。

状态（State）

状态是一个可序列化的 Pydantic 模型或 TypedDict，它承载了代理工作过程中所有需要持久化的数据。典型的代理状态包含：

  • 消息历史（用户消息、助手消息、工具消息）
  • 工具调用记录
  • 中间计算结果
  • 自定义标志位（如 next_step）

状态在节点间传递时，LangGraph 采用合并更新策略：每个节点返回的状态字典会与当前状态合并，而不是完全覆盖。这使你可以在不同节点中逐步构建状态。

环境准备

开始之前，请确保安装必要的库。推荐使用 Python 3.11+。

pip install langgraph langchain-openai langchain-community

若需使用其他模型提供商，可安装对应的 LangChain 集成包。本教程以 OpenAI 为例，你需要设置 API 密钥：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"

构建第一个简单的状态图代理

我们将构建一个具有记忆能力的“旅行助手”代理。它能回答旅行问题，也可以调用一个天气查询工具辅助决策。代理会自主决定是否调用工具，并基于结果生成最终回答。

步骤 1：定义状态

使用 TypedDict 定义状态，包含 messages 列表（LangChain 的消息对象）。

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

这里 Annotated[Sequence[BaseMessage], operator.add] 表示当节点返回包含 messages 的更新时，会执行追加（add）而不是替换，从而保留完整历史。

步骤 2：创建工具节点

定义一个模拟天气查询工具的函数。

from langchain_core.messages import ToolMessage
import json

def get_weather(city: str) -> str:
    """获取指定城市的天气信息（模拟）"""
    # 实际应用中可调用真实 API
    weather_data = {
        "北京": "晴朗，22°C",
        "上海": "多云，25°C",
        "纽约": "小雨，18°C"
    }
    return weather_data.get(city, "未知城市")

# 工具节点函数：根据上一个助手消息中的工具调用执行工具
def weather_tool_node(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    # 从助手消息中提取工具调用
    tool_calls = last_message.tool_calls

    # 通常会有多个工具调用，这里简化处理
    tool_results = []
    for tool_call in tool_calls:
        args = json.loads(tool_call["args"])
        result = get_weather(args["city"])
        tool_results.append(
            ToolMessage(content=result, tool_call_id=tool_call["id"])
        )
    return {"messages": tool_results}

步骤 3：配置 LLM 并绑定工具

使用 LangChain 的 OpenAI 聊天模型，并绑定工具定义。

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

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

# 定义工具列表，遵循 OpenAI 的函数调用格式
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如北京"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

llm_with_tools = llm.bind_tools(tools)

步骤 4：定义代理节点

代理节点调用 LLM，接收消息历史并返回更新后的消息。

from langchain_core.messages import AIMessage

def agent_node(state: AgentState) -> AgentState:
    # 添加系统提示，指导代理行为
    messages = [SystemMessage(content="你是一个旅行助手，当需要查询天气时请使用工具。")] + list(state["messages"])
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

步骤 5：定义路由逻辑

根据 LLM 的返回决定下一步：如果消息中包含工具调用，则进入工具节点；否则结束流程。

def should_use_tool(state: AgentState) -> str:
    last_message = state["messages"][-1]
    # 如果助手消息有 tool_calls，则需要调用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "weather_tool"
    # 否则，直接结束
    return "end"

步骤 6：构建图

使用 StateGraph 将所有组件组合在一起。

from langgraph.graph import StateGraph, END

# 初始化图，指定状态类型
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("weather_tool", weather_tool_node)

# 设置入口点
workflow.set_entry_point("agent")

# 添加条件边：从 agent 出发，根据路由函数决定下一站
workflow.add_conditional_edges(
    "agent",
    should_use_tool,
    {
        "weather_tool": "weather_tool",
        "end": END
    }
)

# 添加正常边：工具调用后返回 agent 继续处理
workflow.add_edge("weather_tool", "agent")

# 编译图
app = workflow.compile()

此时，图的结构如下：

  1. 从 agent 节点开始。
  2. 若 LLM 选择调用工具，则走向 weather_tool，执行完毕后回到 agent。
  3. 若 LLM 直接回复，则走向 END，流程结束。

这个循环保证了代理可以多次调用工具，直到给出最终答案。

运行代理

我们可以通过 invoke 方法调用编译好的图，并传入初始状态。

from langchain_core.messages import HumanMessage

inputs = {"messages": [HumanMessage(content="北京明天天气怎么样？适合出行吗？")]}

# 流式输出中间状态（可选）
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"节点 '{key}':")
        # 节点内可能产生多条消息，这里只打印最后一条
        if "messages" in value:
            print(value["messages"][-1])
        print("---")

# 最终状态
final_state = app.invoke(inputs)
print("\n最终回复:", final_state["messages"][-1].content)

运行后，你会看到代理先进入 agent 节点，产生一个工具调用；然后进入 weather_tool 节点，返回天气结果；再次进入 agent 节点，基于结果生成出行建议；最后结束。

增加持久化与记忆

默认情况下，状态图仅在单次调用期间存在。要实现跨会话的记忆，只需在编译时提供一个 checkpointer（检查点器）。LangGraph 内置了内存检查点器和 SQLite 检查点器。

使用内存检查点

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app_with_memory = workflow.compile(checkpointer=memory)

调用时需提供一个 thread_id，同一线程的调用会共享状态：

config = {"configurable": {"thread_id": "user-123"}}

# 第一次对话
output1 = app_with_memory.invoke(
    {"messages": [HumanMessage(content="我叫小明，我喜欢历史。")]},
    config=config
)
# 第二次对话（可以引用之前的信息）
output2 = app_with_memory.invoke(
    {"messages": [HumanMessage(content="根据我的喜好，推荐一个北京景点。")]},
    config=config
)
print(output2["messages"][-1].content)
# 代理将记住用户名叫小明，喜欢历史，从而推荐故宫等景点。

使用 SQLite 永久存储

from langgraph.checkpoint.sqlite import SqliteSaver

db_path = "agent_sessions.db"
with SqliteSaver.from_conn_string(db_path) as checkpointer:
    app = workflow.compile(checkpointer=checkpointer)
    # 后续调用与上面类似，状态会持久化到文件

有了检查点，代理不仅可以记忆跨对话，还支持人为干预：你可以在等待状态下暂停执行，修改状态后继续。

进阶模式：人机协作回环

在某些敏感操作前，你可能希望人类审批。这可以通过在图中插入一个“等待审批”节点，并利用检查点实现。

简化示例：在调用工具前，增加一个人工审批步骤。

def human_approval_node(state: AgentState) -> AgentState:
    # 实际场景中，这里会将图暂停并等待外部信号
    # 教程简化：直接返回批准状态
    print("请求人工审批：即将调用工具。")
    # 这里可添加标志位通知前端
    return state

def should_call_tool(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        # 在实际系统中，这里会根据外部审批结果决定
        return "weather_tool"
    return "end"

# 构建图时，在 agent 和 weather_tool 之间加入 human_approval 节点
workflow2 = StateGraph(AgentState)
workflow2.add_node("agent", agent_node)
workflow2.add_node("approval", human_approval_node)
workflow2.add_node("weather_tool", weather_tool_node)

workflow2.set_entry_point("agent")
workflow2.add_conditional_edges("agent", should_call_tool, {"weather_tool": "approval", "end": END})
workflow2.add_edge("approval", "weather_tool")
workflow2.add_edge("weather_tool", "agent")

结合检查点，外部应用可以监听状态并调用 update_state 来继续执行。

可视化图结构

在开发复杂代理时，直观地查看图非常有用。LangGraph 支持将图导出为 Mermaid 格式。

print(app.get_graph().draw_mermaid())

将输出的 Mermaid 代码复制到支持 Mermaid 的工具中（如 GitHub Markdown、Mermaid Live Editor），即可得到流程图。

常见问题与最佳实践

  • 状态设计：保持状态扁平化，避免嵌套过深。使用 TypedDict 时合理利用 Annotated 定义合并策略。
  • 节点函数纯净化：尽量让节点返回状态更新，避免在节点内产生副作用。副作用（如数据库写入）应在独立节点或有监督的模式下进行。
  • 错误处理：为工具节点添加异常捕获，可将错误信息作为 ToolMessage 返回，让 LLM 自行修正或决定下一步。
  • 超长对话：利用状态中的消息修剪功能，避免上下文超过模型令牌限制。
  • 自循环与递归：小心设计条件边，确保图有明确的终止条件，防止无限循环。

总结

通过状态图代理，LangGraph 赋予了 AI 代理真正的状态感知能力和复杂决策流程。你学会了：

  • 用节点、边和状态构建图代理。
  • 实现工具调用循环。
  • 添加持久化记忆。
  • 设计带有人工干预的流程。

这是一个强大的起点。你可以继续扩展，集成更多工具、引入多代理协作、甚至构建自主执行长期任务的完整系统。LangGraph 的图结构让代理的行为透明、可控且易于调试，是下一代 AI 应用程序的坚实基础。