目标

第 09 篇你用 while 循环硬撑出一个 Agent:模型思考 → 调工具 → 看结果 → 再思考,直到收敛。它能跑,但两个问题很明显:

  1. 控制流散在 Python 逻辑里,加一个分支就要改循环体,越写越乱;
  2. 状态靠一个 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 成立:

  1. 程序输出当前时间(说明 agent 调了模型 → 触发 get_current_timetools 执行 → 回到 agent 汇总)
  2. 把问题换成"讲个笑话",模型不再调工具,直接回答并走到 END(说明 should_continue 路由正确)
  3. 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(部署),工程化这条线从"能跑"一步步走向"能交付"。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐