随着大语言模型（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界面功能

功能	说明
实时聊天交互	与智能体实时对话
工具调用可视化	显示工具调用过程
状态追踪	追踪状态变化
调试功能	详细的调试信息