langgraph python guide中文入门教程详解

在大语言模型（LLM）应用从单次问答走向复杂任务编排的过程中，传统的无环流水线（如简单的提示词链）很难支撑诸如多轮工具调用、反思重试、状态回溯等复杂业务场景。LangGraph 是 LangChain 团队推出的 Python 状态图编排框架，专为解决“带状态的循环执行流”而设计。本文作为一份严谨、直接的中文实操指南，将系统拆解 LangGraph 的状态机制、节点与边编排逻辑、检查点持久化以及常见问题排障，帮助开发者在 Python 环境中快速建立规范的智能体图工作流。

一、核心模型：状态图（StateGraph）的运行原理

LangGraph 的底层结构是一张图（Graph），其本质是一个带状态的离散状态机。与传统 DAG（有向无环图）工作流框架不同，LangGraph 允许图的边形成闭环（Cycles），这对于需要模型根据工具执行结果进行多轮自我修正的 Agent 系统至关重要。

状态图主要由三个要素构成：

• State（状态容器）：整个图运行时共享的数据上下文。通常使用 Python 的 TypedDict 或 Pydantic 模型声明。状态中的各个键值支持通过 Reducer（聚合函数，如 add_messages）实现增量追加，而不是粗暴地覆盖历史。
• Nodes（节点）：图中的计算单元。每个节点就是一个普通的 Python 函数，接收当前状态字典，经过运算后返回一个部分字典来更新状态。
• Edges（边）：决定执行流程走向的通道，分为无条件边（普通边）与条件边（Conditional Edges）。条件边通过决策函数返回目标节点标识，实现动态路由。

二、环境准备与依赖安装

LangGraph 依赖 Python 3.10 或更高版本。在隔离的虚拟环境中安装基础包：

# 创建并激活虚拟环境
python3 -m venv venv
source venv/bin/activate  # macOS / Linux
# 或在 Windows 执行: venv\Scripts\activate

# 安装核心依赖包
pip install --upgrade langgraph langchain-core langchain-openai

安装完成后，可通过以下单行命令验证核心模块是否可用：

python3 -c "import langgraph; import langchain_core; print('LangGraph 版本:', langgraph.__version__)"

若打印出版本号（通常为 0.2.x 或更高版本），说明基础环境已配置完成。此外，如需调用外部模型，需将对应平台的 API 密钥导出为环境变量（例如 export OPENAI_API_KEY="your-api-key"）。

三、实操开发流程：四步构建标准循环图

在 LangGraph 中开发应用遵循高度规范的四步法：定义状态结构、编写执行节点、编排连线与条件、编译图并运行。以下步骤展现了从数据结构到图实例的完整实现过程。

完整代码实现清单

下面是一个包含自定义计算节点、模拟状态流转与检查点机制的完整可执行代码示例（demo_graph.py）：

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

# 1. 定义状态结构
class AgentState(TypedDict):
    # add_messages 会将新产生的消息追加到列表中，而非直接替换原有列表
    messages: Annotated[list[BaseMessage], add_messages]
    step_count: int

# 2. 定义处理节点
def reasoning_node(state: AgentState) -> dict:
    current_step = state.get("step_count", 0) + 1
    last_msg = state["messages"][-1].content
    
    # 模拟模型判断逻辑：若包含计算指令则进入工具，否则直接给出结论
    if "计算" in last_msg and current_step == 1:
        reply = AIMessage(content="[ACTION_CALCULATE] 需要执行数学计算")
    else:
        reply = AIMessage(content=f"已完成分析，最终解答：收到问题 '{last_msg}'")
    
    return {
        "messages": [reply],
        "step_count": current_step
    }

def calculation_tool_node(state: AgentState) -> dict:
    # 执行具体工具运算，并将工具响应回传给状态
    tool_result = AIMessage(content="[TOOL_RESULT] 经计算，核心指标值为 42")
    return {
        "messages": [tool_result]
    }

# 3. 定义路由决策函数
def route_decision(state: AgentState) -> str:
    last_message = state["messages"][-1]
    # 若上一条消息要求执行工具，则进入工具节点；否则流向 END
    if "[ACTION_CALCULATE]" in last_message.content:
        return "tool_node"
    return END

# 4. 构建并编排状态图
builder = StateGraph(AgentState)

# 注册节点
builder.add_node("reasoning", reasoning_node)
builder.add_node("tool_node", calculation_tool_node)

# 编排边流转
builder.add_edge(START, "reasoning")
builder.add_conditional_edges(
    "reasoning",
    route_decision,
    {
        "tool_node": "tool_node",
        END: END
    }
)
# 工具执行完毕后，回环连接到 reasoning 节点形成闭环
builder.add_edge("tool_node", "reasoning")

# 挂载内存检查点
memory = MemorySaver()
app = builder.compile(checkpointer=memory)

# 5. 执行测试验证
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "session-1001"}}
    input_payload = {
        "messages": [HumanMessage(content="请帮我计算业务指标")],
        "step_count": 0
    }
    
    print("--- 开始图执行 ---")
    for event in app.stream(input_payload, config=config):
        for node_name, output in event.items():
            print(f"节点 [{node_name}] 输出:")
            for msg in output.get("messages", []):
                print(f"  {type(msg).__name__}: {msg.content}")
            if "step_count" in output:
                print(f"  当前步骤计数: {output['step_count']}")
    print("--- 图执行完毕 ---")

执行该脚本后，可以看到完整的状态流转过程：START 首先进入 reasoning 节点，路由函数判定需要计算后转向 tool_node，工具执行产生结果后回环至 reasoning 节点，最终由 reasoning 输出最终解答并路由至 END 退出。控制台示例输出如下：

--- 开始图执行 ---
节点 [reasoning] 输出:
  AIMessage: [ACTION_CALCULATE] 需要执行数学计算
  当前步骤计数: 1
节点 [tool_node] 输出:
  AIMessage: [TOOL_RESULT] 经计算，核心指标值为 42
节点 [reasoning] 输出:
  AIMessage: 已完成分析，最终解答：收到问题 '[TOOL_RESULT] 经计算，核心指标值为 42'
  当前步骤计数: 2
--- 图执行完毕 ---

四、条件边与路由机制（Conditional Edges）详解

在真实业务中，条件边承担着“大脑决策分流”的关键角色。理解条件边的映射关系与边界防线，是写出健壮工作流的前提。

编写条件边时，有三个核心规范必须遵守：

1. 显式提供映射字典（Path Map）：虽然 add_conditional_edges 允许仅传递决策函数名，但在生产环境中务必传入第三个参数映射表（如 {"tools": "tool_node", END: END}）。显式声明可以让编译阶段静态校验所有可能分支的合法性，防止运行时跳转到未注册节点。
2. 无死循环保障：如果模型在工具返回结果后依然反复输出相同的工具调用指令，图将无限循环。除了优化 Prompt 之外，必须在调用 invoke 或 stream 时在 config 中配置 recursion_limit（默认为 25 次），触发上限时抛出 GraphRecursionError。
3. 使用官方预置路由：LangGraph 提供了 langgraph.prebuilt.tools_condition，它能够自动判断消息对象中的 tool_calls 列表是否存在，开箱即用，避免自行编写冗余的条件判断代码。

五、会话持久化与状态检查点（Checkpointer）

生产级对话系统的核心挑战在于跨请求的状态保存与多会话隔离。LangGraph 通过 Checkpointer（检查点系统）原生实现了这一能力。

挂载检查点后，所有状态变更都会在每个超级步（Superstep）自动写入快照。开发者可以通过以下两个核心 API 与状态进行交互：

# 1. 查询当前线程的最新状态
current_state = app.get_state(config)
print("当前最后一条消息:", current_state.values["messages"][-1].content)
print("下一待执行节点:", current_state.next)

# 2. 手动干预/修改状态（支持人机协作场景）
app.update_state(
    config,
    {"messages": [HumanMessage(content="用户在审批环节手动修正了输入")]},
    as_node="human_feedback"
)

六、常见问题与调试排障指南

在编写 LangGraph 应用时，以下几个高频问题最容易导致运行时异常或状态丢失：

七、总结与核心概念全景速查

掌握 LangGraph 的本质，就是建立起“状态驱动、节点计算、路由连线、检查点持久”的架构认知。在实际工程项目中，建议从最小的状态字典与纯函数节点起步，显式指定条件边的路径映射，并在图编译时挂载检查点，从而构建出兼具灵活性、容错力与可观测性的 Python 大模型智能体应用。