普通的 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] 声明累积规则,所以每个节点返回的新消息都会追加到历史记录。round 和 approved 没有 reducer,新值会直接替换旧值。
这个例子中的两个节点也展示了“多角色”工作流的基本形态:不同节点承担不同职责,但读写同一份受约束的状态。接入真实模型时,可以把 planner 和 reviewer 函数内部替换为 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 时,可以按下面的顺序检查设计:
- 把状态定义成清晰、可序列化的数据结构,不要把数据库连接或客户端对象塞进状态。
- 让每个节点只承担一种职责,例如调用模型、执行工具或验证输出。
- 对累计字段声明 reducer,并明确其他字段是覆盖还是保留。
- 给每个循环设置业务退出条件、最大轮次和调用预算。
- 单独测试路由函数,因为分支错误往往比提示词错误更难排查。
- 对外部工具增加超时、重试和幂等控制,避免节点重放产生重复副作用。
- 需要跨进程恢复或人工审批时,再引入持久化 checkpoint,而不是只依赖进程内状态。
LangGraph 的价值不在于把所有 LLM 调用画成复杂流程图,而在于把真正需要状态、循环和多个角色协作的部分变成可检查的程序控制流。对于一次调用就能完成的任务,普通函数或线性 Chain 往往更简单;当流程开始出现修订、工具反馈、人工介入和恢复执行时,图模型才会明显降低维护成本。