用 LangGraph 构建有状态、可循环的 Python AI Agent

2026-07-13 34 预计阅读时间: 1 分钟
来源: realpython.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:7 分钟

普通的 LLM 调用通常是一条直线:提交提示词,等待模型返回结果。LangGraph 处理的是更复杂的工作流,例如规划器生成方案、审查器提出修改意见、规划器继续修订,直到满足退出条件。它在 LangChain 生态之上引入图、共享状态、条件分支和循环,使 Agent 的执行路径更明确,也更容易测试。

LangGraph 的核心不是“图”,而是状态迁移

一个 LangGraph 工作流通常包含四类元素:

  • State:节点之间共享的数据,例如消息、轮次、草稿和审批结果。
  • Node:读取状态并返回局部更新的 Python 函数。
  • Edge:规定节点之间的固定执行顺序。
  • Conditional edge:根据当前状态选择下一条路径,也就是 Agent 的控制逻辑。

与线性 Chain 相比,Graph 可以返回已经执行过的节点,因此适合“生成、检查、修订”一类闭环流程。与完全自由的 Agent 相比,显式边和退出条件又能限制执行范围,避免模型无限调用工具。

这里有一个容易忽略的细节:节点通常不需要返回完整状态,只返回它负责修改的字段。LangGraph 会把这些更新合并进共享状态;如果字段使用 reducer,还可以累积多次更新,而不是让新值覆盖旧值。

一个不依赖模型密钥的循环工作流

为了先看清图的运行机制,可以这样实践:用确定性的 Python 函数模拟“规划器”和“审查器”。示例会运行三轮,然后通过条件边退出。它不调用真实 LLM,因此可以直接复制运行。

安装依赖:

python -m pip install -U langgraph

创建 agent_graph.py

from operator import add
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph


class AgentState(TypedDict):
    topic: str
    round: int
    messages: Annotated[list[str], add]
    approved: bool


def planner(state: AgentState) -> dict:
    current_round = state["round"] + 1
    return {
        "round": current_round,
        "messages": [
            f"Planner: draft {current_round} for '{state['topic']}'"
        ],
    }


def reviewer(state: AgentState) -> dict:
    approved = state["round"] >= 3
    verdict = "approved" if approved else "revise with more detail"
    return {
        "approved": approved,
        "messages": [f"Reviewer: {verdict}"],
    }


def route_after_review(state: AgentState) -> Literal["revise", "finish"]:
    return "finish" if state["approved"] else "revise"


builder = StateGraph(AgentState)
builder.add_node("planner", planner)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "planner")
builder.add_edge("planner", "reviewer")
builder.add_conditional_edges(
    "reviewer",
    route_after_review,
    {
        "revise": "planner",
        "finish": END,
    },
)

graph = builder.compile()

result = graph.invoke(
    {
        "topic": "stateful AI agents",
        "round": 0,
        "messages": [],
        "approved": False,
    }
)

print(f"Rounds: {result['round']}")
print(f"Approved: {result['approved']}")
for message in result["messages"]:
    print(message)

运行:

python agent_graph.py

输出应显示规划器和审查器交替执行三轮。messages 字段通过 Annotated[list[str], add] 声明累积规则,所以每个节点返回的新消息都会追加到历史记录。roundapproved 没有 reducer,新值会直接替换旧值。

这个例子中的两个节点也展示了“多角色”工作流的基本形态:不同节点承担不同职责,但读写同一份受约束的状态。接入真实模型时,可以把 plannerreviewer 函数内部替换为 LangChain Chat Model 调用,同时保留图结构和路由函数。

循环必须有明确的刹车

循环是 LangGraph 相对线性工作流的重要能力,也是最需要防护的部分。生产代码不应只依赖模型返回“完成”来结束执行。更稳妥的做法是同时设置业务条件和硬限制,例如:

def route_after_review(state: AgentState) -> Literal["revise", "finish"]:
    if state["approved"] or state["round"] >= 5:
        return "finish"
    return "revise"

调用图时还可以设置递归限制,防止错误路由造成无限执行:

result = graph.invoke(
    {
        "topic": "stateful AI agents",
        "round": 0,
        "messages": [],
        "approved": False,
    },
    config={"recursion_limit": 12},
)

递归限制不是业务退出条件的替代品。它更像最后一道保险:触发时应记录状态、节点和输入,帮助定位是哪条边没有按预期收敛。

从演示图走向真实 Agent

落地 LangGraph 时,可以按下面的顺序检查设计:

  1. 把状态定义成清晰、可序列化的数据结构,不要把数据库连接或客户端对象塞进状态。
  2. 让每个节点只承担一种职责,例如调用模型、执行工具或验证输出。
  3. 对累计字段声明 reducer,并明确其他字段是覆盖还是保留。
  4. 给每个循环设置业务退出条件、最大轮次和调用预算。
  5. 单独测试路由函数,因为分支错误往往比提示词错误更难排查。
  6. 对外部工具增加超时、重试和幂等控制,避免节点重放产生重复副作用。
  7. 需要跨进程恢复或人工审批时,再引入持久化 checkpoint,而不是只依赖进程内状态。

LangGraph 的价值不在于把所有 LLM 调用画成复杂流程图,而在于把真正需要状态、循环和多个角色协作的部分变成可检查的程序控制流。对于一次调用就能完成的任务,普通函数或线性 Chain 往往更简单;当流程开始出现修订、工具反馈、人工介入和恢复执行时,图模型才会明显降低维护成本。


相关推荐