AI Agent 开发实战(11):第一个 LangGraph(用代码把工作流画成图)
目标
第 09 篇你用 while 循环硬撑出一个 Agent:模型思考 → 调工具 → 看结果 → 再思考,直到收敛。它能跑,但两个问题很明显:
- 控制流散在 Python 逻辑里,加一个分支就要改循环体,越写越乱;
- 状态靠一个
messages列表手动维护,没有"中途存档、断点续跑"的能力。
本篇用 LangGraph 把同样的逻辑"画"成一张图:节点是步骤,边是流转,循环用一条"指回自己"的边表达。完成后你会理解——Agent 的本质是一张可循环的图,而 LangGraph 只是把这个结构显式化了。

环境准备
- Python 3.10+(LangGraph 要求)
- 一个干净项目目录,下文以
D:\你的用户名\项目\langgraph-demo为例 - 第 03 篇申请的 OpenRouter Key(环境变量
OPENROUTER_API_KEY,复用 09 篇底座,不用新申请) .env文件(放 Key,不要硬编码)
装依赖:
pip install -U langgraph openai python-dotenv
步骤
1. 定义"图的状态"
LangGraph 的核心是一份贯穿全图的共享状态。每个节点读它、改它。我们先定义一个最简单的状态:只有 messages 一个字段,并用 add_messages 归约器让新消息追加而不是覆盖。
import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from openai import OpenAI
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY"),
)
class State(TypedDict):
# Annotated[list, add_messages]:节点返回消息时,追加进列表而非替换
messages: Annotated[list, add_messages]
⚠️ 归约器是 LangGraph 最容易踩的坑(见踩坑 1)。不写
Annotated,节点返回的消息会覆盖整段历史,多轮对话直接废掉。
2. 写一个工具和一个工具清单
和 09 篇一样,先有个能调的工具:
def get_current_time() -> str:
"""返回当前本地时间,格式 YYYY-MM-DD HH:MM:SS。用于回答与时间相关的问题。"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
tools = [{
"type": "function",
"function": {
"name": "get_current_time",
"description": "返回当前本地时间。当用户询问现在几点、今天日期时使用。",
"parameters": {"type": "object", "properties": {}, "required": []},
},
}]
3. 写两个节点(agent / tools)
节点就是普通函数,第一个参数是 state,返回值是"对状态的更新"。
def agent(state: State):
"""调用模型,决定直接回答还是请求调工具。"""
resp = client.chat.completions.create(
model="deepseek/deepseek-v4-pro",
messages=state["messages"],
tools=tools,
)
# 把模型的回复(可能是文本,也可能是 tool_calls)追加进状态
return {"messages": [resp.choices[0].message]}
def tools_node(state: State):
"""执行模型请求的工具调用,把结果写回状态。"""
last = state["messages"][-1]
results = []
for tc in last.tool_calls:
if tc.function.name == "get_current_time":
content = get_current_time()
results.append({
"role": "tool",
"tool_call_id": tc.id,
"content": content,
})
return {"messages": results}
对比 09 篇:09 篇里"调模型 + 解析 tool_calls + 循环"全挤在一个 run_agent 函数里;这里被拆成两个职责单一的节点,图的边负责把它们串起来。
4. 写路由函数(决定要不要继续调工具)
def should_continue(state: State):
"""路由:模型最后一条消息带了 tool_calls 就走 tools,否则结束。"""
last = state["messages"][-1]
if getattr(last, "tool_calls", None):
return "tools"
return END
这个函数就是 09 篇 while 循环里"模型是否还在调工具"的判断,只不过现在它变成图上的一条条件边。
5. 把节点和边拼成图
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", tools_node)
builder.add_edge(START, "agent") # 入口
builder.add_conditional_edges( # 条件边:根据 should_continue 分流
"agent", should_continue,
{"tools": "tools", END: END}
)
builder.add_edge("tools", "agent") # 工具跑完,回到 agent 形成循环
graph = builder.compile()
把这段和 09 篇对照,你会一眼看出:add_edge("tools", "agent") 就是那个 while 循环体,而 should_continue 就是循环的终止条件。agent → tools → agent → ... → END 这条回路,就是 ReAct 循环的可视化。
6. 跑起来
result = graph.invoke({
"messages": [{"role": "user", "content": "现在几点了?"}]
})
print(result["messages"][-1].content)
想看整张图的样子,可以导出 Mermaid(需 pip install langgraph 自带支持):
from langgraph.graph import StateGraph # get_graph().draw_mermaid()
print(graph.get_graph().draw_mermaid())
验证
三项全过 = 你的第一个 LangGraph Agent 成立:
- 程序输出当前时间(说明
agent调了模型 → 触发get_current_time→tools执行 → 回到agent汇总) - 把问题换成"讲个笑话",模型不再调工具,直接回答并走到
END(说明should_continue路由正确) draw_mermaid()能画出START → agent →(tools)→ tools → agent → END的结构
第 2 项最关键:它证明同一条边能根据情况走不同分支,这正是手写 while 难以直观表达的。
设计权衡:LangGraph vs 第 09 篇手写循环
| 维度 | 09 篇(手写 while) | 11 篇(LangGraph) |
|---|---|---|
| 控制流 | 藏在循环体里,靠读代码才懂 | 显式成图,节点/边一目了然 |
| 状态管理 | 自己维护 messages 列表 | 框架管,归约器控制合并 |
| 循环/分支 | while + if 硬编码 | add_edge 回指 + add_conditional_edges |
| 持久化 | 没有,断了重来 | 可加 checkpointer,断点续跑 |
| 可读性 | 小脚本直观 | 复杂流程更清晰 |
| 代价 | 无依赖、零学习成本 | 引入框架、概念多、小项目偏重 |
诚实地说:Demo 级别 LangGraph 不比手写循环强多少,甚至更啰嗦。它的价值在"流程变复杂后"——多 Agent 协作、人工介入(human-in-the-loop)、状态回溯、可视化调试,这些手写要自己造轮子,LangGraph 开箱就有。所以别为了"显得高级"在小项目上用它,这点和你"反过度工程"的判断一致。
踩坑记录
1. 忘了 Annotated 归约器,历史被覆盖
class State(TypedDict): messages: list 这样写,节点返回的 {"messages": [...]} 会整体替换旧历史,模型第二轮的上下文里看不到第一轮。必须 Annotated[list, add_messages]。
2. tool_calls 属性不存在报错
should_continue 里直接 last.tool_calls 在普通 dict 消息上会 AttributeError。用 getattr(last, "tool_calls", None),或确保状态里流动的是 SDK 消息对象(本例 add_messages 会保留对象形态)。
3. 条件边 mapping 漏写 END
add_conditional_edges("agent", should_continue, {"tools": "tools"}) 没给 END 分支,模型不调工具时图不知道往哪走,会报错或卡死。路由函数返回的所有可能值都要在 mapping 里出现。
4. base_url 端点写错
连 OpenRouter 必须用 https://openrouter.ai/api/v1(v1 接口),不是 03 篇 Claude Code 用的 /anthropic 兼容端点。两者是不同协议,别混。
5. 模型名斜杠格式
用 deepseek/deepseek-v4-pro(厂商/模型),写错直接 404(和 09 篇踩坑 6 同源)。
6. 版本依赖
LangGraph API 迭代较快,pip install -U langgraph 拿到新版;老教程里的 graph.add_edge("__start__", ...) 已过时,用 START/END 常量。
7. 循环没出口 = 死循环
如果 should_continue 永远返回 "tools"(比如工具实现有 bug 一直不收敛),图会无限循环。add_conditional_edges 务必有走到 END 的路径。
8. checkpointer 不是默认开启
想"断点续跑"要显式传 compile(checkpointer=...) 并带 thread_id 调用,否则每次 invoke 都是全新状态,不报错但也没记忆。
下一步
工具标准化了(10 MCP),流程也能画成图了(11 LangGraph)。但本地脚本跑着玩,和"能部署给别人用"之间还差一步——打包成容器:
→ AI Agent 开发实战(12):Docker 部署 Agent
第 12 篇把你的 Agent 装进 Docker 镜像,环境一致、随处可跑,不再"在我机器上能跑"。
系列衔接:09(手写 Agent)→ 10(工具标准化)→ 11(流程编排)→ 12(部署),工程化这条线从"能跑"一步步走向"能交付"。
更多推荐


所有评论(0)