随着大语言模型（LLM）技术的快速发展，AI应用的需求已经从简单的问答扩展到了复杂的多步骤任务处理、多智能体协作等场景。传统的线性链式开发方式在面对这些复杂需求时显得力不从心。LangGraph应运而生，它是一个用于构建有状态、多智能体AI应用的开源框架，采用图（Graph）结构来编排复杂工作流，为开发者提供了更强大和灵活的工具。
本文将基于LangGraph的官方文档和实践经验，从基础概念到核心机制，再到实际应用案例，全面深入地解析LangGraph的技术原理和应用方法，帮助开发者掌握这一强大的AI应用开发框架。
一、LangGraph基础概览
1.1 核心理念：图驱动的工作流
LangGraph将AI应用的逻辑抽象为一张由"节点"（Nodes）和"边"（Edges）构成的状态图。下图清晰展示了这种结构与执行方式：
在LangGraph的图结构中，每个组件都有明确的职责：
| 组件 | 定义 | 功能 | 示例 |
|---|---|---|---|
| 节点（Nodes） | 独立的执行单元 | 处理数据，执行操作 | 调用LLM、执行工具函数、查询数据库 |
| 边（Edges） | 流程控制逻辑 | 决定下一步执行哪个节点 | 固定连接、条件分支、动态路由 |
| 状态（State） | 共享的数据结构 | 保存上下文信息 | 消息历史、查询结果、中间状态 |
1.2 主要特点与优势
LangGraph相比传统的AI应用开发框架，具有以下显著特点：
| 特性 | 说明 | 应用场景 |
|---|---|---|
| 循环与条件分支 | 支持智能体的"思考-行动"循环，动态改变执行路径 | 需要多轮决策的任务 |
| 内置状态管理 | 通过检查点自动持久化状态，支持断点续执行 | 长时间运行的应用 |
| 人机协同 | 可在流程中暂停并等待人工输入或审批 | 高风险或需要把关的场景 |
| 多智能体协作 | 原生支持定义多个智能体，编排协同工作 | 复杂的团队协作任务 |
| 可观测性 | 与LangSmith集成，提供可视化调试 | 开发和调试阶段 |
1.3 生态系统与使用方式
LangGraph提供了灵活的集成路径，开发者可以根据需求选择不同的使用方式：
| 使用方式 | 特点 | 适用场景 |
|---|---|---|
| 独立使用 | 作为底层框架，自定义智能体系统 | 有特殊需求的开发者 |
| 与LangChain集成 | 无缝兼容LangChain生态中的模型、工具 | 已熟悉LangChain的开发者 |
| 高级组件 | 使用预构建的智能体模式 | 快速开发常见应用 |
| 商业平台 | 提供部署、托管、监控等生产级功能 | 企业用户 |
1.4 LangGraph与LangChain的区别
| 特性 | LangChain | LangGraph |
|---|---|---|
| 核心抽象 | 链（Chain），线性编排 | 状态图（Stateful Graph），带循环和分支 |
| 适用场景 | RAG、文档处理、一次性问答 | 智能体、多轮对话、自动化流程、多智能体系统 |
| 状态管理 | 通过Memory组件传递上下文 | 内置的、中心化的状态管理 |
二、LangGraph图的关键概念
2.1 智能体（Agent）
智能体是LangGraph中最基础也是最重要的概念之一。
工作原理
| 特性 | 说明 |
|---|---|
| 核心机制 | LLM基于环境反馈循环使用工具 |
| 实现复杂度 | 相对简单，但需要清晰的工具集设计 |
| 适用场景 | 开放性问题，无法预测所需步骤数量 |
| 关键要求 | 需要良好的工具集和文档设计 |
2.2 工作流（Workflow）
2.2.1 并行化（Parallelization）
并行化是指让多个LLMs同时处理一个任务，并通过编程方式对它们的输出结果进行聚合。
两种主要变体：
| 类型 | 说明 | 示例 |
|---|---|---|
| 分段（Sectioning） | 将任务分解成多个相互独立的子任务 | 生成报告的不同部分（摘要、方法、结果） |
| 投票（Voting） | 对同一个任务运行多次，综合多个结果 | 多个LLMs进行情感分析，投票确定结果 |
并行化的适用场景：
| 场景 | 优势 |
|---|---|
| 提高速度 | 显著缩短处理时间 |
| 提高置信度 | 综合多个结果，提高可靠性 |
| 复杂任务处理 | 每个LLM专注于特定方面 |
2.2.2 路由（Router）
路由的工作是将输入进行分类，并将其导向到相应的后续任务。
| 路由的作用 | 说明 |
|---|---|
| 输入分类与任务导向 | 根据输入内容导向合适的处理流程 |
| 分离关注点 | 每个任务专注于特定领域 |
| 避免性能受损 | 不同输入类型分配给最适合的处理任务 |
2.2.3 协调者（Orchestrator）
在协调者-工作者模式中，一个协调者将任务分解并将每个子任务分配给工作者。
| 组件 | 职责 |
|---|---|
| 协调者 | 动态分解任务，分配子任务，综合结果 |
| 工作者 | 处理分配的特定子任务 |
与并行化模式的区别：
| 特性 | 并行化模式 | 协调者-工作者模式 |
|---|---|---|
| 子任务定义 | 预先定义好的 | 动态生成的 |
| 灵活性 | 相对固定 | 高度灵活 |
| 适用场景 | 已知结构的任务 | 无法预测子任务的复杂任务 |
2.2.4 评估者-优化器（Evaluator-optimizer）
| 适用场景 | 说明 |
|---|---|
| 明确的评估标准 | 存在清晰的质量衡量标准 |
| 迭代改进的价值 | 每次调整都能显著提升结果 |
| 人类反馈有效 | LLM能够理解并应用人类反馈 |
| LLM反馈能力 | LLM能够提供有效的评估和改进建议 |
三、深度理解LangGraph核心：Graph
3.1 Graph的基本组成
Graph是LangGraph的基本构建模块，它是一个有向无环图（DAG），用于描述任务之间的依赖关系。
| 元素 | 定义 | 说明 |
|---|---|---|
| State（状态） | 共享的数据结构 | 在整个应用当中共享，包含所有节点的状态 |
| Node（节点） | 处理数据的单元 | Python函数，以State为输入，返回更新后的State |
| Edge（边） | 依赖关系 | Python函数，根据当前State决定下一步执行哪个Node |
3.2 State状态
3.2.1 State的定义方式
使用TypedDict定义：
from typing import TypedDict class OverallState(TypedDict): foo: str user_input: str graph_output: str使用Pydantic BaseModel定义：
from pydantic import BaseModel class OverallState(BaseModel): a: str| 定义方式 | 特点 | 适用场景 |
|---|---|---|
| TypedDict | 轻量级，类型提示 | 简单状态结构 |
| Pydantic BaseModel | 支持验证、序列化 | 需要数据验证的场景 |
3.2.2 State的更新机制
| 更新策略 | 说明 | 示例 |
|---|---|---|
| 默认替换 | 直接替换字段值 | return {"a": "goodbye"} |
| 合并操作 | 使用add_messages、add等操作 | messages: Annotated[list[AnyMessage], add_messages] |
MessagesState便捷使用：
from langgraph.graph import MessagesState # 直接使用，无需手动定义 state = MessagesState()3.3 Node节点
3.3.1 Node的基本定义
def node_1(state: InputState) -> OverallState: return {"foo": state["user_input"] + "> 长沙市"}| 特性 | 说明 |
|---|---|
| 输入 | State对象（必选）+ config（可选） |
| 输出 | 更新后的State对象 |
| 命名 | 唯一字符串，未指定时使用函数名 |
3.3.2 Node的高级特性
| 特性 | 说明 | 示例 |
|---|---|---|
| 缓存机制 | 相同输入优先从缓存获取结果 | cache_policy=CachePolicy(ttl=5) |
| 重试机制 | 失败时自动重试 | retry=RetryPolicy(max_attempts=4) |
缓存示例：
from langgraph.types import CachePolicy from langgraph.cache.memory import InMemoryCache builder.add_node("node1", node_1, cache_policy=CachePolicy(ttl=5)) graph = builder.compile(cache=InMemoryCache()) # 第一次调用：执行节点逻辑 graph.invoke({"number": 5}, config={"configurable": {"user_id": "123"}}) # 第二次调用：从缓存获取 graph.invoke({"number": 5}, config={"configurable": {"user_id": "456"}})3.4 Edge边
3.4.1 边的类型
| 边类型 | 说明 | 示例 |
|---|---|---|
| 普通边 | 固定连接两个节点 | builder.add_edge("node_1", "node_2") |
| 条件边 | 根据状态动态选择下一个节点 | builder.add_conditional_edges(START, routing_func) |
| Send动态路由 | 一个节点同时路由到多个节点 | Send("node1", {"msg": message}) |
3.4.2 普通边和EntryPoint
from langgraph.constants import START, END builder.add_edge(START, "node_1") builder.add_edge("node_1", "node_2") builder.add_edge("node_2", END)3.4.3 条件边
def routing_func(state: State) -> str: if state["number"] > 5: return "node1" else: return END builder.add_conditional_edges(START, routing_func)使用映射的方式：
def routing_func(state: State) -> bool: return state["number"] > 5 builder.add_conditional_edges( START, routing_func, {True: "node_a", False: "node_b"} )3.4.4 Send动态路由
from langgraph.types import Send def routing_func(state: State): result = [] for message in state["messages"]: result.append(Send("node1", {"msg": message})) return result builder.add_conditional_edges(START, routing_func)3.5 子图的使用
创建子图：
# 子图构建 subgraph_builder = StateGraph(State) subgraph_builder.add_node("sub_node", sub_node_function) subgraph_builder.add_edge(START, "sub_node") subgraph_builder.add_edge("sub_node", END) subgraph = subgraph_builder.compile() # 父图构建 builder = StateGraph(State) builder.add_node("subgraph_node", subgraph) builder.add_edge(START, "subgraph_node") builder.add_edge("subgraph_node", END) graph = builder.compile()3.6 图的Stream支持
| Stream模式 | 说明 | 适用场景 |
|---|---|---|
| values | 流式传输状态的完整值 | 需要查看每步的完整状态 |
| updates | 流式传输更新内容 | 只关心状态变化 |
| custom | 流式传输自定义数据 | 调试和监控 |
| messages | 流式传输LLM的Token | 实时显示LLM响应 |
| debug | 传输尽可能多的信息 | 深度调试 |
Custom Stream示例：
from langgraph.config import get_stream_writer def node(state: State): writer = get_stream_writer() writer({"自定义key": "在节点内返回自定义信息"}) return {"answer": "some data"} for chunk in graph.stream(inputs, stream_mode="custom"): print(chunk)四、LangGraph实战：SQLAgent实现
4.1 SQL应用场景
| 场景 | 说明 |
|---|---|
| 自然语言查询 | 用户用自然语言查询数据库 |
| 数据分析 | 自动生成SQL并执行分析 |
| 报表生成 | 根据需求自动生成报表 |
4.2 langchain_community的使用
4.2.1 安装
pip install langchain-community pip install langchain-community[sql]
4.2.2 基本使用
from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit db = SQLDatabase.from_uri("sqlite:///Chinook.db") toolkit = SQLDatabaseToolkit(db=db, llm=llm) tools = toolkit.get_tools()4.2.3 主要功能模块
| 模块 | 功能 |
|---|---|
| utilities | 实用工具类，如SQLDatabase |
| agent_toolkits | 智能体工具包，如SQLDatabaseToolkit |
| llms | 各种LLM集成 |
| chat_models | 聊天模型集成 |
4.3 SQL智能体的创建
完整示例：
import asyncio from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit llm = init_chat_model("deepseek:deepseek-chat") db = SQLDatabase.from_uri("sqlite:///Chinook.db") toolkit = SQLDatabaseToolkit(db=db, llm=llm) tools = toolkit.get_tools() system_prompt = """ 你是SQL数据库智能体，根据问题生成正确的{dialect}查询语句。 限制查询结果最多{top_k}条，只查询相关列。 严禁执行INSERT、UPDATE、DELETE等操作。 开始时必须查看表结构。 """.format(dialect=db.dialect, top_k=10) agent = create_agent(llm, tools, system_prompt=system_prompt) async def main(): async for step in agent.astream( {"messages": [{"role": "user", "content": "哪种音乐类型的曲目平均时长最长？"}]}, stream_mode="values" ): step["messages"][-1].pretty_print() asyncio.run(main())4.4 SQL工具介绍
| 工具 | 功能 |
|---|---|
| sql_db_query | 执行SQL查询语句 |
| sql_db_schema | 获取数据库表结构信息 |
| sql_db_list_tables | 获取所有可用表名 |
| sql_db_query_checker | 安全验证SQL查询 |
4.5 MCP Server Chart集成
集成示例：
import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient mcp_client = MultiServerMCPClient({ "mcp-server-chart": { "command": "npx", "args": ["-y", "@antv/mcp-server-chart"], "transport": "stdio" } }) mcp_tools = asyncio.run(mcp_client.get_tools()) agent = create_agent(llm, tools + mcp_tools, system_prompt=system_prompt)MCP特点：
| 特性 | 说明 |
|---|---|
| MCP协议支持 | 提供MCP适配器功能 |
| 多服务器连接 | 支持连接多个MCP服务器 |
| 工具集成 | 将外部工具集成到智能体 |
| 图表生成 | 支持数据可视化 |
五、LangGraph本地服务器与UI界面
5.1 配置langgraph.json
{ "$schema": "https://langgra.ph/schema.json", "dependencies": ["."], "graphs": { "agent": "./main.py:agent", "sql_agent": "./example/pro2/sql_agent.py:agent" }, "env": ".env", "image_distro": "wolfi" }| 配置项 | 说明 |
|---|---|
| $schema | Schema定义 |
| dependencies | 依赖关系 |
| graphs | 图的入口点 |
| env | 环境变量文件 |
| image_distro | 镜像发行版 |
5.2 启动服务器
langgraph dev
5.3 UI界面功能
| 功能 | 说明 |
|---|---|
| 实时聊天交互 | 与智能体实时对话 |
| 工具调用可视化 | 显示工具调用过程 |
| 状态追踪 | 追踪状态变化 |
| 调试功能 | 详细的调试信息 |
