LangGraph框架：构建有状态LLM应用的图结构编程指南-AI智能范式网

LangGraph框架：构建有状态LLM应用的图结构编程指南

LangGraphLLM应用图结构编程

|难度：普通|阅读约 35 分钟

1. LangGraph 框架概述

LangGraph 是由 LangChain 团队开发的开源框架（MIT 协议），专门用于构建有状态的、基于图结构的 LLM 应用。它的核心设计理念是将 AI 应用逻辑建模为有向图——节点（Node）代表具体的操作步骤，边（Edge）定义步骤之间的流转规则，状态（State）则是在节点之间传递的共享数据。

与传统的链式调用（Chain）相比，LangGraph 的最大特点是原生支持循环和条件分支。这使得它特别适合构建需要"思考-行动-观察"反复迭代的 Agent 系统。在实际应用中，LangGraph 可以独立使用，但与 LangChain 配合使用时效果更佳——LangChain 负责模型接口和工具集成，LangGraph 则专注于流程编排和控制。

1.1 核心优势解析

LangGraph 的设计带来了几个关键优势：

有状态执行：自动管理和传递状态，节点之间无需手动传参。状态对象在整个执行过程中保持持久化，每个节点都可以读取和修改状态。
循环支持：Agent 可以反复调用工具直到任务完成，不像简单 Chain 只能线性执行一次。这在需要多次迭代优化的场景中特别有用。
条件路由：根据运行时结果动态决定下一步执行的节点。这使得应用能够根据实际情况做出智能决策，而不仅是固定流程。
持久化机制：内置 checkpoint 功能，支持中断后恢复、回溯历史状态。这对于长时间运行的业务流程至关重要。
人工介入：可在关键节点暂停，等待人工审核后继续执行。这在需要人工监督的高风险操作中非常实用。
流式输出：原生支持 token 级别的流式响应，提升用户体验。前端可以实时展示 Agent 的思考过程。

1.2 典型应用场景

LangGraph 特别适合以下场景：

需要调用外部工具的对话 Agent：如客服机器人、个人助理等
多步骤的 RAG 流水线：包括检索、评估、重写、生成等环节
多 Agent 协作系统：不同 Agent 分工合作完成复杂任务
需要人工审批的自动化工作流：如内容审核、金融交易等
任何需要循环决策的 LLM 应用：如持续优化的内容生成

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 安装与环境配置

2.1 基础安装步骤

安装 LangGraph 非常简单，使用 pip 即可完成：

bash复制pip install langgraph

如果计划配合 LangChain 使用（推荐），还需要安装对应的模型集成包。以下是常见模型的安装命令：

bash复制# 使用 OpenAI 模型
pip install langchain-openai

# 使用 Anthropic 模型
pip install langchain-anthropic

# 使用 Google 模型
pip install langchain-google-genai

2.2 环境变量设置

根据使用的模型，需要设置对应的 API Key：

bash复制# OpenAI
export OPENAI_API_KEY="sk-..."

# Anthropic
export ANTHROPIC_API_KEY="sk-ant-..."

# Google
export GOOGLE_API_KEY="your-key-here"

2.3 版本要求与兼容性

LangGraph 需要 Python 3.9 及以上版本。建议使用 Python 3.11+，以获得更好的类型提示支持。截至当前版本，LangGraph 与最新版的 LangChain 完全兼容。

提示：建议定期更新 LangGraph 以获取最新功能和修复：
bash复制pip install -U langgraph

3. 核心概念深度解析

3.1 State（状态）机制

State 是整个图的核心记忆载体。它是一个 TypedDict，定义了在图中流转的所有数据字段。每个节点都能读取和更新 State 中的字段。

python复制from typing import TypedDict

class MyState(TypedDict):
    query: str           # 用户输入
    result: str          # 最终结果
    steps: list[str]     # 中间步骤记录

Reducer 机制：默认情况下，节点返回的字段值会直接覆盖 State 中的对应字段。但对于消息列表等需要"追加"而非"覆盖"的场景，LangGraph 提供了 Reducer 机制：

python复制from typing import Annotated
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]  # 自动追加消息而非替换

add_messages 是内置的 Reducer，它不仅追加消息，还能根据消息 ID 进行去重和更新。

3.2 Node（节点）设计

节点是执行具体操作的 Python 函数，接收 State 作为输入，返回一个字典表示对 State 的更新。

python复制def analyze(state: MyState) -> dict:
    query = state["query"]
    # 执行分析逻辑...
    return {"result": f"分析结果：{query}", "steps": ["完成分析"]}

节点函数只需返回需要更新的字段，不需要包含 State 的所有字段。这使得每个节点可以专注于自己的职责。

3.3 Edge（边）类型与路由

边定义了节点之间的连接关系，分为两种类型：

普通边（Direct Edge）：无条件地从一个节点流向下一个节点。

python复制graph.add_edge("node_a", "node_b")  # 执行完 node_a 后一定执行 node_b

条件边（Conditional Edge）：根据函数的返回值动态决定执行路径。

python复制def route(state: MyState) -> str:
    if "错误" in state["result"]:
        return "retry"     # 走 retry 节点
    return "finish"        # 走 finish 节点

graph.add_conditional_edges("check", route, ["retry", "finish"])

3.4 StateGraph 构建流程

StateGraph 是将上述概念组合在一起的构建器。典型构建流程如下：

python复制from langgraph.graph import StateGraph, START, END

# 1. 创建图实例
graph = StateGraph(MyState)

# 2. 添加节点
graph.add_node("analyze", analyze)

# 3. 添加边
graph.add_edge(START, "analyze")  # START 是内置起点
graph.add_edge("analyze", END)    # END 是内置终点

# 4. 编译为可执行应用
app = graph.compile()

# 5. 调用执行
result = app.invoke({"query": "你好", "result": "", "steps": []})

4. 从零构建第一个应用

4.1 文本处理示例

下面通过一个完整的文本处理示例展示 LangGraph 的基本用法：

python复制from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义 State 结构
class State(TypedDict):
    text: str
    word_count: int

# 2. 定义节点函数
def process_text(state: State) -> dict:
    """文本预处理"""
    text = state["text"]
    cleaned = text.strip().lower()
    return {"text": cleaned}

def count_words(state: State) -> dict:
    """统计词数"""
    words = state["text"].split()
    return {"word_count": len(words)}

# 3. 构建图
graph = StateGraph(State)
graph.add_node("process", process_text)
graph.add_node("count", count_words)

# 4. 定义执行流程
graph.add_edge(START, "process")
graph.add_edge("process", "count")
graph.add_edge("count", END)

# 5. 编译并执行
app = graph.compile()
result = app.invoke({"text": "  Hello World LangGraph  ", "word_count": 0})

print(result)
# 输出: {'text': 'hello world langgraph', 'word_count': 3}

这个示例清晰地展示了 LangGraph 的执行流程：START → process（清洗文本）→ count（统计字数）→ END。State 在节点之间自动传递和更新。

4.2 条件分支实现

条件边是 LangGraph 的核心特性之一。下面是一个根据查询类型路由到不同处理节点的示例：

python复制from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    category: str
    response: str

def classify(state: State) -> dict:
    """查询分类"""
    query = state["query"].lower()
    if "价格" in query or "多少钱" in query:
        return {"category": "pricing"}
    elif "故障" in query or "报错" in query:
        return {"category": "technical"}
    return {"category": "general"}

def handle_pricing(state: State) -> dict:
    """处理价格查询"""
    return {"response": "关于价格问题，请访问我们的定价页面或联系销售团队。"}

def handle_technical(state: State) -> dict:
    """处理技术问题"""
    return {"response": "关于技术问题，请提供错误日志，我们的工程师会帮您排查。"}

def handle_general(state: State) -> dict:
    """处理一般查询"""
    return {"response": "感谢您的咨询，请问还有什么可以帮助您的？"}

def route_query(state: State) -> Literal["pricing", "technical", "general"]:
    """路由函数"""
    return state["category"]

# 构建图
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("pricing", handle_pricing)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)

# 定义流程
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_query, ["pricing", "technical", "general"])
graph.add_edge("pricing", END)
graph.add_edge("technical", END)
graph.add_edge("general", END)

# 编译并执行
app = graph.compile()
result = app.invoke({"query": "这个产品多少钱？", "category": "", "response": ""})
print(result["response"])
# 输出: 关于价格问题，请访问我们的定价页面或联系销售团队。

5. 构建工具调用 Agent

5.1 定义工具集

首先定义 Agent 可以调用的工具：

python复制from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    return f"搜索结果：关于'{query}'的最新信息..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)  # 生产环境应使用更安全的计算方式
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"

tools = [search_web, calculate]

@tool 装饰器来自 LangChain，它会自动将函数转换为 LLM 能理解的工具描述格式。

5.2 初始化模型

python复制from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)  # 将工具绑定到模型

5.3 定义 Agent 节点

python复制from typing import Annotated
from langgraph.graph import MessagesState

def agent_node(state: MessagesState) -> dict:
    """调用 LLM 决定下一步"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

5.4 定义工具执行节点

python复制from langchain_core.messages import ToolMessage

tools_by_name = {t.name: t for t in tools}  # 工具名称到工具对象的映射

def tool_node(state: MessagesState) -> dict:
    """执行工具调用"""
    results = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        results.append(
            ToolMessage(content=str(observation), tool_call_id=tool_call["id"])
        )
    return {"messages": results}

5.5 定义路由逻辑

python复制from typing import Literal
from langgraph.graph import END

def should_continue(state: MessagesState) -> Literal["tools", END]:
    """判断是否继续调用工具"""
    last_message = state["messages"][-1]
    return "tools" if last_message.tool_calls else END

5.6 组装完整 Agent

python复制from langgraph.graph import StateGraph, START

graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")  # 工具执行后回到 agent，形成循环

agent = graph.compile()

5.7 运行 Agent

python复制from langchain_core.messages import HumanMessage

result = agent.invoke({
    "messages": [HumanMessage(content="帮我算一下 (15 + 27) * 3")]
})

for msg in result["messages"]:
    print(msg.content)

典型执行流程是：

用户发送消息
Agent 节点分析后决定调用工具
Tools 节点执行工具并返回结果
Agent 节点处理结果并决定下一步
循环直到任务完成

6. 高级特性与应用

6.1 持久化会话

使用 Checkpointer 实现会话记忆：

python复制from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")  # 内存数据库
agent = graph.compile(checkpointer=memory)

# 使用 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user-123"}}

# 第一轮对话
result1 = agent.invoke(
    {"messages": [HumanMessage(content="我叫小明")]},
    config=config
)

# 第二轮对话 - Agent 记得之前的内容
result2 = agent.invoke(
    {"messages": [HumanMessage(content="你还记得我叫什么吗？")]},
    config=config
)

6.2 人工介入机制

在关键节点加入人工审核：

python复制agent = graph.compile(
    checkpointer=memory,
    interrupt_before=["tools"]  # 在执行工具前暂停
)

# 第一次调用会在 tools 节点前暂停
result = agent.invoke(
    {"messages": [HumanMessage(content="帮我删除所有数据")]},
    config=config
)

# 查看 Agent 打算做什么
pending_tool_calls = agent.get_state(config).values["messages"][-1].tool_calls

# 人工确认后继续执行
result = agent.invoke(None, config=config)

6.3 流式输出

支持多种粒度的流式响应：

python复制# Token 级别流式
for msg, _ in agent.stream(
    {"messages": [HumanMessage(content="写一首关于 AI 的诗")]},
    config=config,
    stream_mode="messages"
):
    if msg.content:
        print(msg.content, end="", flush=True)

7. 实战案例：RAG 质量控制系统

下面是一个带质量评估的 RAG 系统实现：

python复制from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

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

class RAGState(TypedDict):
    question: str
    context: str
    answer: str
    quality_score: float
    retry_count: int

def retrieve(state: RAGState) -> dict:
    """文档检索"""
    question = state["question"]
    context = f"检索到的与'{question}'相关的文档内容..."
    return {"context": context}

def evaluate_quality(state: RAGState) -> dict:
    """质量评估"""
    score = 0.8  # 模拟评估分数
    return {"quality_score": score}

def generate_answer(state: RAGState) -> dict:
    """生成回答"""
    prompt = f"根据上下文回答问题：\n上下文：{state['context']}\n问题：{state['question']}"
    response = llm.invoke(prompt)
    return {"answer": response.content}

def rewrite_query(state: RAGState) -> dict:
    """重写查询"""
    prompt = f"重写以下问题以改善检索效果：\n{state['question']}"
    response = llm.invoke(prompt)
    return {"question": response.content, "retry_count": state["retry_count"] + 1}

def quality_check(state: RAGState) -> Literal["generate", "rewrite", "force_generate"]:
    """质量检查路由"""
    if state["quality_score"] >= 0.7:
        return "generate"
    if state["retry_count"] >= 2:
        return "force_generate"  # 超过重试次数限制
    return "rewrite"

# 构建图
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("evaluate", evaluate_quality)
graph.add_node("generate", generate_answer)
graph.add_node("rewrite", rewrite_query)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "evaluate")
graph.add_conditional_edges(
    "evaluate",
    quality_check,
    {"generate": "generate", "rewrite": "rewrite", "force_generate": "generate"}
)
graph.add_edge("rewrite", "retrieve")  # 重写后重新检索
graph.add_edge("generate", END)

rag_agent = graph.compile()

这个系统会在检索后评估质量，如果质量不足会重写查询重新检索，最多重试2次。

8. 调试与最佳实践

8.1 可视化调试

python复制# 生成 Mermaid 图（需要安装 graphviz）
print(app.get_graph().draw_mermaid())

8.2 错误处理建议

python复制def safe_node(state: State) -> dict:
    try:
        # 节点逻辑...
        return {"result": "成功"}
    except Exception as e:
        return {"error": str(e), "result": "失败"}

8.3 生产环境建议

使用 PostgreSQL 等持久化 Checkpointer
关键操作加入人工审核
配置合理的超时和重试机制
接入 LangSmith 进行监控
对工具调用做权限控制

9. 框架对比与选型

LangGraph vs LangChain Agents：

LangChain Agents 是高层抽象，适合标准 ReAct 模式
LangGraph 提供更底层的流程控制能力

LangGraph vs CrewAI/AutoGen：

CrewAI/AutoGen 侧重多 Agent 对话，开箱即用
LangGraph 更灵活，适合需要精细控制的场景

选择建议：

需要快速实现标准 Agent → LangChain Agents
需要复杂流程控制 → LangGraph
需要多 Agent 对话 → CrewAI/AutoGen