python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python LangGraph 用法

Python LangGraph 用法从入门到最佳实战

作者：日日大王学编程

本文系统讲解LangGraph的有向图建模思想及其在LLM应用开发中的关键能力,本文结合实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧

一、为什么需要 LangGraph

在构建基于大语言模型（LLM）的应用时，我们常常遇到这样的场景：一个问题可能需要多次调用模型才能解决；需要根据模型的输出决定下一步做什么；需要调用外部工具（搜索、数据库查询等）；某些任务需要反复执行直到满足条件。

传统的 LangChain 采用线性链式调用（Chain），其执行路径是一条直线——先做 A，再做 B，最后做 C。这种模式简单直观，但无法表达环形控制和条件分支。而真实的 Agent 运行路径是环形且带分支的：模型可能直接回答，也可能请求调用工具，工具执行完结果还要回到模型，是否结束取决于模型上一轮的输出。

LangGraph 正是为了解决这个问题而诞生的。它将 AI 应用建模为一个有向图，其中：

一句话概括：Chain 是单行道，Graph 是立交桥。

二、核心概念

2.1 State（状态）

状态是工作流的核心，用于存储和传递信息。在 LangGraph 中，状态通常是一个 Python 字典（通过 TypedDict 定义类型）。每个节点接收当前状态作为输入，返回一个字典表示对状态的更新。

from typing import TypedDict
class AgentState(TypedDict):
    messages: list          # 对话消息历史
    current_step: str       # 当前执行步骤
    result: str             # 最终结果
    error: str | None       # 错误信息

状态的设计原则：节点之间只通过状态通信，不直接调用。这保证了代码的可测试性和可维护性。

2.2 Nodes（节点）

节点是执行实际工作的函数。每个节点接收状态作为参数，返回一个字典用于更新状态。节点函数的签名如下：

def my_node(state: AgentState) -> dict:
    # 1. 从状态中读取数据
    messages = state["messages"]
    # 2. 执行某些操作（调用LLM、调用工具等）
    result = do_something(messages)
    # 3. 返回增量字典，只更新需要修改的字段
    return {
        "messages": messages + [result],
        "current_step": "done"
    }

重要约定：节点函数应返回"增量字典"（只包含需要更新的字段），而不是返回整个状态对象。这样做的好处是：后续如果有并发分支、reducer、子图拼接等操作时，状态合并更加稳定可靠。

2.3 Edges（边）

边定义节点之间的连接关系，控制流程的执行顺序。LangGraph 提供两种边：

普通边（Normal Edge）：无条件地从源节点跳转到目标节点。

from langgraph.graph import StateGraph, START, END
workflow = StateGraph(AgentState)
workflow.add_node("step1", step1_func)
workflow.add_node("step2", step2_func)
# 普通边：step1 执行完一定去 step2
workflow.add_edge(START, "step1")
workflow.add_edge("step1", "step2")
workflow.add_edge("step2", END)

条件边（Conditional Edge）：根据路由函数的返回值动态决定下一个节点。

def router(state: AgentState) -> str:
    if state["current_step"] == "error":
        return "retry"
    elif state["current_step"] == "done":
        return "end"
    return "continue"
workflow.add_conditional_edges(
    "process",
    router,
    {
        "retry": "retry_node",
        "continue": "process",
        "end": END
    }
)

2.4 Checkpointer（检查点）

检查点是 LangGraph 的持久化层，它能够在图执行过程中保存状态快照。有了检查点，你可以实现：

三、环境安装

# 安装 LangGraph 核心包
pip install langgraph
# 如果需要与 LangChain 集成
pip install langchain langchain-openai
# 如果需要 SQLite 持久化（开发环境推荐）
pip install langgraph-checkpoint-sqlite
# 如果需要 PostgreSQL 持久化（生产环境推荐）
pip install langgraph-checkpoint-postgres
# 验证安装
python -c "import langgraph; print(langgraph.__version__)"

四、快速入门：最小可运行示例

让我们从一个最简单的示例开始——创建一个问候节点，输入名字，输出问候语。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 1. 定义状态
class State(TypedDict):
    messages: str
# 2. 定义节点函数
def greeting_node(state: State) -> dict:
    """问候节点：拼接问候语"""
    return {"messages": "Hello! " + state["messages"]}
# 3. 构建图
graph = StateGraph(State)
# 添加节点
graph.add_node("greeting", greeting_node)
# 添加边：START -> greeting -> END
graph.add_edge(START, "greeting")
graph.add_edge("greeting", END)
# 4. 编译图
app = graph.compile()
# 5. 执行
result = app.invoke({"messages": "World"})
print(result)
# 输出: {'messages': 'Hello! World'}

这个最小示例展示了 LangGraph 应用的完整生命周期：定义状态 -> 定义节点 -> 连接边 -> 编译 -> 执行。

五、核心 API 详解

5.1 StateGraph

StateGraph 是 LangGraph 的核心类，用于构建有状态的工作流图。

from langgraph.graph import StateGraph
# 创建图实例，传入状态类型
workflow = StateGraph(State)

5.2 add_node

向图中添加一个节点。节点可以是普通函数、Lambda 表达式、LangChain Runnable 对象等。

# 方式1：普通函数
def agent_node(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}
workflow.add_node("agent", agent_node)
# 方式2：Lambda 函数（适合简单逻辑）
workflow.add_node("simple", lambda state: {"result": "done"})
# 方式3：LangChain Runnable（LCEL 管道）
from langchain_core.runnables import RunnableLambda
chain = ChatPromptTemplate.from_template("回答: {question}") | ChatOpenAI() | StrOutputParser()
workflow.add_node("qa", chain)
# 方式4：异步节点
async def async_agent(state: State) -> dict:
    response = await llm.ainvoke(state["messages"])
    return {"messages": [response]}
workflow.add_node("async_agent", async_agent)

5.3 add_edge

添加普通边，从源节点无条件跳转到目标节点。

# 线性流程：START -> step1 -> step2 -> step3 -> END
workflow.add_edge(START, "step1")
workflow.add_edge("step1", "step2")
workflow.add_edge("step2", "step3")
workflow.add_edge("step3", END)

5.4 add_conditional_edges

添加条件边，根据路由函数的返回值选择目标节点。这是实现 Agent 工具调用循环的关键。

from langgraph.graph import MessagesState
from langgraph.prebuilt import ToolNode
# 定义路由函数
def should_continue(state: MessagesState) -> str:
    last_message = state["messages"][-1]
    # 模型请求了工具调用 -> 去工具节点
    # 否则 -> 结束
    if last_message.tool_calls:
        return "tools"
    return "end"
# 添加条件边
workflow.add_conditional_edges(
    "llm_call",           # 源节点
    should_continue,      # 路由函数
    {                     # 映射字典
        "tools": "tools", # 返回"tools" -> 跳转到tools节点
        "end": END        # 返回"end" -> 结束
    }
)

5.5 compile

编译图，生成可执行的应用程序。这是所有节点和边定义完成后的最后一步。

# 基础编译
app = workflow.compile()
# 带检查点的编译（支持持久化）
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

六、完整示例：带工具调用的 Agent

这是 LangGraph 最经典的使用场景——构建一个能调用工具的对话 Agent。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
# 1. 定义工具
@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果: 关于'{query}'的搜索结果..."
@tool
def calculator(expression: str) -> str:
    """执行数学计算"""
    result = eval(expression)
    return f"计算结果: {result}"
# 2. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
# 3. 构建 Agent 循环
llm_with_tools = ChatOpenAI(model="gpt-4").bind_tools([search, calculator])
def llm_call(state: AgentState) -> dict:
    """调用 LLM"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}
def should_continue(state: AgentState) -> str:
    """条件路由：判断是否需要调用工具"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"
# 4. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("llm", llm_call)
workflow.add_node("tools", ToolNode([search, calculator]))
# 添加边
workflow.add_edge(START, "llm")
workflow.add_conditional_edges(
    "llm",
    should_continue,
    {"tools": "tools", "end": END}
)
workflow.add_edge("tools", "llm")  # 工具执行完回到模型，形成循环
# 5. 编译并执行
agent = workflow.compile()
result = agent.invoke({"messages": [("user", "帮我计算 2+2 并搜索最新的AI新闻")]})
print(result["messages"][-1].content)

这个 Agent 的工作流程如下：

START -> llm (有tool_calls) -> tools -> llm (无tool_calls) -> END
                          |
                          v
                      (继续循环...)

七、状态管理与 Reducer

7.1 默认覆盖语义

使用 TypedDict 时，节点返回的字段会覆盖状态中的旧值：

class State(TypedDict):
    counter: int
def increment(state):
    return {"counter": state["counter"] + 1}  # 覆盖旧值

7.2 归并语义（Reducer）

对于列表类型的字段，我们通常需要追加而非覆盖。LangGraph 提供了 Annotated 机制来声明归并行为：

from typing import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
    # add_messages 是一个 reducer，自动追加新消息
    messages: Annotated[list, add_messages]
    # 自定义 reducer：列表追加
    history: Annotated[list, lambda x, y: x + y]
    # 自定义 reducer：字典合并
    config: Annotated[dict, lambda x, y: {**x, **y}]

常见的内置 reducer：

Reducer行为适用场景
add_messages追加消息到列表对话历史管理
operator.add数值累加计数器
lambda x,y: x+y列表拼接日志收集
lambda x,y: {**x,**y}字典合并配置更新

7.3 消息裁剪

对话历史可能无限增长，需要通过裁剪来控制状态大小：

from langchain_core.messages import AnyMessage, RemoveMessage
from langgraph.graph import START
MAX_MESSAGES = 20
def trim_messages(state):
    """裁剪消息列表，只保留最近 N 条"""
    msgs = state["messages"]
    if len(msgs) <= MAX_MESSAGES:
        return {}  # 不需要裁剪
    # 标记要删除的消息
    to_remove = [RemoveMessage(id=m.id) for m in msgs[:-MAX_MESSAGES]]
    return {"messages": to_remove}
# 在图中添加裁剪节点
workflow.add_node("trim", trim_messages)
workflow.add_edge(START, "trim")
workflow.add_edge("trim", "agent")

八、持久化与检查点

8.1 内存检查点（开发环境）

from langgraph.checkpoint.memory import MemorySaver
# 创建内存检查点存储器
checkpointer = MemorySaver()
# 编译时传入检查点
app = workflow.compile(checkpointer=checkpointer)
# 使用线程配置执行
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [("user", "你好")]}, config=config)

8.2 SQLite 检查点（单机部署）

pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
# 方式1：文件路径
saver = SqliteSaver.from_conn_string("checkpoints.db")
# 方式2：配合 sqlite3 连接（多线程安全）
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
saver = SqliteSaver(conn)
app = workflow.compile(checkpointer=saver)

8.3 PostgreSQL 检查点（生产环境）

pip install langgraph-checkpoint-postgres

from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:pwd@localhost:5432/agentdb"
with PostgresSaver.from_conn_string(DB_URI) as saver:
    saver.setup()  # 首次运行需要建表
    app = workflow.compile(checkpointer=saver)

8.4 断点与恢复

检查点最强大的能力是支持断点续跑——在特定节点暂停执行，等待外部干预后继续：

from langgraph.types import interrupt, Command
def sensitive_operation(state):
    """需要人工审批的节点"""
    # 暂停执行，等待外部决策
    decision = interrupt({
        "action": "refund",
        "amount": 299.0,
        "user_id": state["user_id"]
    })
    if decision == "approve":
        return {"result": "refund_processed"}
    return {"result": "refund_rejected"}
# 编译时指定在哪些节点前/后暂停
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_operation"]
)
# 第一次调用：执行到 interrupt 处暂停
config = {"configurable": {"thread_id": "order-888"}}
result = app.invoke(
    {"messages": [("user", "我要退款")], "user_id": "u123"},
    config=config
)
# 此时 result 中包含 __interrupt__ 信息
# 恢复执行：传入 resume 命令
result = app.invoke(
    Command(resume="approve"),
    config=config
)

8.5 状态历史与时间旅行

# 获取当前状态
snapshot = app.get_state(config)
print(snapshot.values)    # 当前状态值
print(snapshot.next)      # 下一步要执行的节点
# 遍历状态历史
for snap in app.get_state_history(config):
    print(snap.config["configurable"]["checkpoint_id"], snap.next)
# 时间旅行：回到过去的某个快照
history = list(app.get_state_history(config))
old_snapshot = history[2]  # 回到第3个快照
# 从旧快照分叉出新分支
forked_config = {
    "configurable": {
        "thread_id": "user-123",
        "checkpoint_id": old_snapshot.config["configurable"]["checkpoint_id"]
    }
}
# 在新分支上继续执行，不影响原分支
result = app.invoke(
    {"messages": [("user", "换个思路重新分析")]},
    config=forked_config
)
# 手动更新状态
app.update_state(config, {"answer": "手动修正后的答案"})

九、流式传输

LangGraph 支持流式输出，可以在节点执行过程中逐步返回结果，显著提升用户体验：

# 方式1：values 模式 - 返回每个节点的完整状态
for event in app.stream(input, stream_mode="values"):
    print(event)
# 方式2：updates 模式 - 返回每个节点的增量更新
for event in app.stream(input, stream_mode="updates"):
    print(event)
# 方式3：messages 模式 - 只流式返回消息内容
for event in app.stream(input, stream_mode="messages"):
    print(event)
# 方式4：custom 模式 - 使用自定义 writer
from langgraph.types import StreamWriter
def my_node(state, writer: StreamWriter):
    writer({"type": "progress", "step": "processing"})
    result = do_something(state)
    writer({"type": "result", "data": result})
    return {"result": result}

十、子图与嵌套

LangGraph 支持将子图嵌入父图，实现模块化设计：

from typing import TypedDict
# 1. 定义子图状态
class SubgraphState(TypedDict):
    input_text: str
    processed_text: str
# 2. 定义子图节点
def preprocess(state: SubgraphState) -> dict:
    return {"processed_text": state["input_text"].strip().lower()}
def analyze(state: SubgraphState) -> dict:
    return {"processed_text": state["processed_text"] + " [analyzed]"}
# 3. 构建子图
subgraph = StateGraph(SubgraphState)
subgraph.add_node("preprocess", preprocess)
subgraph.add_node("analyze", analyze)
subgraph.add_edge(START, "preprocess")
subgraph.add_edge("preprocess", "analyze")
subgraph.add_edge("analyze", END)
compiled_subgraph = subgraph.compile()
# 4. 在父图中使用子图
class ParentState(TypedDict):
    text: str
    result: str
def parent_node(state: ParentState) -> dict:
    # 调用子图
    sub_result = compiled_subgraph.invoke({"input_text": state["text"]})
    return {"result": sub_result["processed_text"]}
parent_graph = StateGraph(ParentState)
parent_graph.add_node("process", parent_node)
parent_graph.add_edge(START, "process")
parent_graph.add_edge("process", END)
app = parent_graph.compile()
result = app.invoke({"text": "  Hello World  "})
print(result)
# 输出: {'result': 'hello world [analyzed]'}

子图状态传播规则：

十一、并行执行

LangGraph 支持扇出（Fan-out）和扇入（Fan-in）模式，实现多路并行执行：

from typing import TypedDict, List
class ParallelState(TypedDict):
    query: str
    results: List[str]
def search_web(state: ParallelState) -> dict:
    return {"results": [f"Web搜索: {state['query']}"]}
def search_database(state: ParallelState) -> dict:
    return {"results": [f"数据库查询: {state['query']}"]}
def search_knowledge_base(state: ParallelState) -> dict:
    return {"results": [f"知识库检索: {state['query']}"]}
def merge_results(state: ParallelState) -> dict:
    # 合并所有并行结果
    all_results = state["results"]
    return {"results": all_results + ["合并完成"]}
# 构建并行图
workflow = StateGraph(ParallelState)
workflow.add_node("web", search_web)
workflow.add_node("db", search_database)
workflow.add_node("kb", search_knowledge_base)
workflow.add_node("merge", merge_results)
# 扇出：从START同时到三个搜索节点
workflow.add_edge(START, "web")
workflow.add_edge(START, "db")
workflow.add_edge(START, "kb")
# 扇入：三个搜索节点都到合并节点
workflow.add_edge("web", "merge")
workflow.add_edge("db", "merge")
workflow.add_edge("kb", "merge")
workflow.add_edge("merge", END)
app = workflow.compile()
result = app.invoke({"query": "LangGraph教程", "results": []})
print(result)

十二、最佳实践与常见陷阱

12.1 状态设计原则

  1. 最小化状态：只保留节点间通信必需的字段
  2. 不可变更新：节点返回增量字典，不要直接修改状态对象
  3. 类型安全：始终使用 TypedDict 定义状态结构
  4. 归并语义：列表字段使用 Annotated 声明追加行为

12.2 常见错误与排查

问题现象可能原因解决方案
多轮对话丢失记忆每次 thread_id 不同使用稳定的 thread_id（如用户ID）
服务重启后记忆丢失使用了 InMemorySaver切换到 SQLite 或 PostgreSQL
resume 报找不到线程thread_id 不一致确保两次调用使用相同的 thread_id
状态越来越大、越来越慢消息无限累积定期裁剪或摘要历史消息
PostgreSQL 报缺表未调用 setup()首次运行执行 saver.setup()
SQLite 报线程错误多线程共享连接设置 check_same_thread=False
节点函数返回整个 state反模式返回增量字典即可

12.3 性能优化建议

  1. 异步优先：对于 I/O 密集型节点，使用 async def 定义节点
  2. 连接池：生产环境使用连接池管理数据库连接
  3. 消息裁剪：定期裁剪对话历史，控制状态大小
  4. 批量处理：利用并行扇出同时执行多个独立任务
  5. 缓存策略：对重复查询结果做缓存，减少 LLM 调用次数

12.4 调试技巧

# 1. 可视化图结构
from langchain_community.graphs import NetworkxEntityGraph
graph = app.get_graph()
graph.draw_mermaid_png("workflow.png")
# 2. 查看执行历史
for event in app.stream(input, stream_mode="updates"):
    print(event)
# 3. 使用 LangSmith 追踪
import langsmith
langsmith.trace("my_agent", graph=app)

十三、总结

LangGraph 通过将 AI 应用建模为有向图，解决了传统链式调用无法表达复杂控制流的问题。其核心抽象——状态（State）、节点（Node）、边（Edge）——提供了清晰且强大的编程模型。

本文涵盖了 LangGraph 的核心用法：

掌握这些核心概念后，你可以构建从简单的对话机器人到复杂的多 Agent 协作系统等各种 AI 应用。建议从最小可运行示例开始，逐步添加条件和循环逻辑，最终构建出生产级别的工作流应用。

到此这篇关于Python LangGraph 用法从入门到最佳实战的文章就介绍到这了,更多相关Python LangGraph 用法内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

阅读全文