返回博客列表
技术文章LangGraphAgentAILLMPythonLangChain

第一次用 LangGraph 写 Agent,我会先讲清这些事

2026年06月🇨🇳 中文

LLM 为什么还不够?

大语言模型(LLM)的能力很强,但在不借助外部工具的情况下,它能做的事情本质上只有一件:接收一段文本(prompt),基于训练时学到的模式和知识,逐 token 生成响应文本。它没有持久记忆——每次调用都是无状态的,不能主动访问外部数据,也不会自发地做多步推理。

举个例子。你问 ChatGPT:"北京明天天气怎么样?"它能回答吗?如果不联网,它只能给一个模糊的"对不起,我无法获取实时天气信息"。如果你让它"帮我计算 (15 + 23) × 7 ÷ 4",它可能会算错——LLM 是语言模型,不是计算器。再如果你让它"帮我查一下 GitHub 上 star 最多的 Python 项目是什么",它给的是训练数据截止之前的答案,而不是实时信息。

这三个场景暴露了裸 LLM 的边界:

  • 不能访问外部数据(天气、数据库、API)
  • 不擅精确计算(数学、逻辑推理)
  • 知识有时效性(训练截止日之后的都不知道)

Agent 的解决思路很简单:给 LLM 装上"手"(工具)和"循环"(推理-行动-观察)。 LLM 不再是一次性输出最终答案,而是进入一个循环:它先思考该做什么,如果发现需要外部信息,就调用一个工具获取结果,然后根据工具返回的结果继续思考,直到它认为可以给出最终答案。

这个模式叫 ReAct(Reasoning + Acting),是当前 Agent 设计最基础的范式。

但如果你动手写过 Agent,很快就会发现一个工程问题:代码复杂度增长得很快。

一个简单的 while-true 循环当然可以跑通,但当你需要处理多轮工具调用、管理不断增长的消息历史、支持条件分支(调完这个工具后要不要再调另一个?)、调试某一步哪里出了问题——裸写循环很快就会变成一个难以维护的状态机。

这就是 LangGraph 要解决的问题。 它用有向图来编排 Agent 的执行流程:

  • State:管理整个 Agent 的共享上下文(消息历史、中间变量等)
  • Node:封装具体的执行单元(LLM 调用、工具执行、格式化输出)
  • Edge / Conditional Edge:控制节点之间的流转,条件边实现动态路由

三个概念加在一起,Agent 的"推理-行动-观察"循环就被建模成了一张有向图——天然支持循环、可观测、可扩展。

LangGraph 到底帮我们整理了什么?

LangGraph 的核心思想:用有向图来编排 Agent 的执行流程。图中的节点代表执行单元,边代表数据流向。每次执行时,从入口节点开始,沿着边依次执行各个节点。每一步的 State 变化都可以被追踪和调试。

下面逐一拆解三个核心概念:State、Node、Edge。

State:让 Agent 记住一路发生过什么

State 是 Agent 的"共享内存"——图中所有节点都能读写同一份 State。 每个节点函数执行时,第一个参数就是当前的 State;执行完后返回一个 dict,LangGraph 自动把这个 dict 的内容合并回 State,下一个节点看到的就是更新后的数据。

这个设计的意义在于:Agent 的每步决策都能看到完整历史,不需要手动在节点间传参数。

State 里存什么?最基础的场景下只有一个 messages 字段——一个列表,存放完整的对话记录:

[HumanMessage("帮我算一下 15 + 23 × 7"),   # 用户的输入
 AIMessage(tool_calls=[calculator...]),      # LLM 说:我准备调计算器
 ToolMessage("计算结果: 176"),               # 工具执行后的结果
 AIMessage("答案是 176")]                    # LLM 拿到结果后的最终回答

每执行一个节点,这个列表就变长一步。LLM 推理时读到的是整段历史——它知道前面自己说过什么、用户要什么、工具返回了什么。这就是 Agent "有记忆"的根本原因。

为什么消息不会丢失? 每个 Node 执行完返回一个 dict(比如 {"messages": [new_msg]}),LangGraph 需要把这个 dict 合入现有 State。它有一个合并规则:add_messages——旧消息列表 + 新消息 → 追加后的消息列表。如果没有这个规则,默认行为是直接覆盖——新的 1 条消息替换掉全部的对话历史,Agent 瞬间"失忆"。这个合并规则在后面代码中会看到。

Node:把每一步动作拆成独立函数

Node 是最小的执行单元,每个节点就是一个 Python 函数,签名为 (state) -> dict:接收当前 State,返回需要更新的部分。

这次 Agent 有两个核心节点:

  • call_model:LLM 推理节点。把 State 里的完整对话历史发给 LLM,LLM 判断下一步该做什么——直接回答还是调用工具。如果决定调用工具,返回的 AIMessage 会包含 tool_calls 字段。
  • call_tool:工具执行节点。LangGraph 内置的 ToolNode,自动解析上一步产生的 tool_calls、执行对应的工具函数、把结果写回 State。

节点的职责单一且明确,不会互相干扰——这就是 LangGraph 架构比裸 while 循环更清晰的地方。

Edge:决定下一步该往哪里走

边决定节点之间的流转。有两种:

普通边(Edge):无条件跳转。从节点 A 执行完后,一定去节点 B。比如工具执行完后一定回到 LLM 节点,让 LLM 根据工具返回的结果继续推理。

条件边(Conditional Edge):有条件跳转。根据当前 State 的内容,动态决定去哪个节点。这是 Agent 路由的核心。比如:LLM 返回了 tool_calls?→ 去执行工具的节点。没有 tool_calls?→ 直接结束。

条件边由一个路由函数实现——函数接收 State,返回一个字符串,这个字符串就是目标节点的名字。整个 Agent 的"智能"就体现在这短短几行条件判断上。

StateGraph:把节点和流转真正串起来

StateGraph 把 Node 和 Edge 组装成一张有向图,调用 .compile() 后得到一个可执行的 Agent 对象。你可以像调普通函数一样 invoke() 它,传入初始 State,拿到完整执行结果。

整体的执行流程:

          START
            │
            ▼
     ┌─────────────┐
     │  call_model  │  ← LLM 推理节点
     └──────┬──────┘
            │ (条件边: 有 tool_calls?)
     ┌──────┴──────┐
     │ 有          │ 无
     ▼             ▼
┌──────────┐       END
│ call_tool│       工具执行节点
└────┬─────┘
     │ (普通边: 回到 LLM)
     └──────► call_model

这就是 LangGraph 的 ReAct 循环:LLM 推理 → 判断是否需要工具 → 需要就执行工具再回到 LLM → 不需要就结束。三个概念加在一起,Agent 的"推理-行动-观察"循环被建模成了一张有向图——天然支持循环、可观测、可扩展。

写一个能计算、能搜索的小 Agent

理论说完了,接下来写代码。我们的目标是构建一个简单的 Agent,它拥有两个工具:

  • 计算器(calculator:执行四则运算
  • 搜索(search:查询信息(用模拟数据)

这个场景足够简单,但覆盖了 Agent 的关键流程:理解意图 → 选择工具 → 调用工具 → 返回结果。

先把模型和依赖准备好

项目依赖 langgraphlangchainlangchain-openaipython-dotenv。安装它们,然后在项目根目录创建 .env 文件:

# .env
DEEPSEEK_API_KEY=sk-your-deepseek-api-key
DEEPSEEK_MODEL=deepseek-chat
DEEPSEEK_BASE_URL=https://api.deepseek.com

本文使用 DeepSeek 作为 LLM 后端,使用其他兼容 OpenAI 接口的模型调整配置即可。

给 Agent 两个可以调用的工具

# app/tools/calculator.py
import math
from langchain_core.tools import tool


@tool
def calculator_tool(expression: str) -> str:
    """执行数学计算。expression 是一个数学表达式,例如 '3 + 5 * 2' 或 'sqrt(16)'。"""
    try:
        allowed_names = {
            k: v for k, v in math.__dict__.items()
            if not k.startswith("__")
        }
        allowed_names["__builtins__"] = {}
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算出错: {str(e)}"


# app/tools/search.py
from langchain_core.tools import tool


@tool
def search_tool(query: str) -> str:
    """搜索信息。query 是要搜索的关键词或问题。"""
    mock_data = {
        "北京天气": "北京今天晴,气温 22°C ~ 32°C,空气质量良",
        "python star最多": "GitHub 上 star 最多的 Python 项目是 public-apis/public-apis,拥有超过 300k star",
        "langgraph": "LangGraph 是 LangChain 团队推出的图式 Agent 编排框架,支持有状态、多角色的 Agent 构建",
    }
    for key, value in mock_data.items():
        if key in query.lower():
            return value
    return f"未找到关于 '{query}' 的信息,请尝试更具体的关键词"

calculator_tool 的安全做法是限制 eval 的命名空间,只允许 math 模块里的函数(sqrtsin 等),拒绝 __builtins__search_tool 是一个模拟实现,实际项目中可替换为 SerpAPI、Tavily 等。

@tool 装饰器会扫描函数的名称、参数和文档字符串,自动生成一份"工具说明书"发给 LLM。当 LLM 面对用户问题"帮我算一下 15 + 23 × 7"时,它能从说明书里找到"有一个 calculator_tool,接收数学表达式字符串,可以执行计算",于是决定调用它。

把工具、模型和状态接到图里

现在把第 2 节的理论变成代码。整个 Agent 由三个部分构成:State 定义、节点函数、图的组装。

第一步:定义 State

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

这里最关键的细节是 Annotated[list[BaseMessage], add_messages]add_messages 就是 LangGraph 的合并规则——"追加,不覆盖"。在运行时,每个 Node 返回 {"messages": [new_msg]},LangGraph 拿到后调用这个规则把新消息追加到 State 的 messages 列表末尾。如果没有它,新消息会直接替换整个列表——之前的对话全部丢失,Agent 瞬间"失忆"。

第二步:初始化 LLM 和系统提示词

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

load_dotenv()

llm = ChatOpenAI(
    model=os.getenv("DEEPSEEK_MODEL", "deepseek-chat"),
    base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1"),
    api_key=os.getenv("DEEPSEEK_API_KEY", ""),
    temperature=0,
).bind_tools(tools)

system_prompt = SystemMessage(
    content="你是一个有用的助手,可以调用计算器和搜索工具来帮助用户。"
    "当用户问数学题时用计算器,问信息时用搜索工具。"
    "如果可以直接回答,就不需要调用工具。"
)

.bind_tools(tools) 把上一步定义的工具列表传给 LLM,告诉它"你有这两个工具可以用"。之后 LLM 在推理时,如果判断需要调用工具,返回的 AIMessage 会自动包含 tool_calls 字段——里面写明了要调哪个工具、传什么参数。

第三步:定义两个节点函数

from langgraph.constants import END
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import ToolNode


def call_model(state: AgentState) -> dict:
    response = llm.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}


def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "call_tool"
    return END

LangGraph 要求每个节点函数签名都是 (state) -> dict——接收 State,返回要更新的内容。call_modelshould_continue 要放在 add_node 里,所以必须定义成独立函数。

llm.invoke(...) 是 LangChain 的标准 API:把消息列表发给 LLM,返回一个 AIMessage 对象。不同模型(DeepSeek、GPT、Claude)返回的原始格式各不相同——有的包了一层 choices[0].message,有的用 content 字段,有的用 parts 字段。LangChain 统一把它们转成 AIMessage,结构很简单:.content 是文本回复,.tool_calls 是工具调用的列表。LangGraph 只认识 AIMessage,不关心背后是哪个模型。

should_continue 返回 "call_tool" 还是 END,决定了 add_conditional_edges 的走向——这是在把第 2 节的流程图变成可执行的逻辑。

第四步:组装并编译 StateGraph

workflow = StateGraph(AgentState)

workflow.add_node("call_model", call_model)
workflow.add_node("call_tool", ToolNode(tools))

workflow.add_edge(START, "call_model")
workflow.add_conditional_edges(
    "call_model", should_continue,
    {"call_tool": "call_tool", END: END},
)
workflow.add_edge("call_tool", "call_model")

agent = workflow.compile()

逐行理解:

  1. StateGraph(AgentState) — 创建一张空图,绑定 State 类型
  2. add_node("call_model", call_model) — 注册 LLM 推理节点
  3. add_node("call_tool", ToolNode(tools)) — 注册工具执行节点。ToolNode 是 LangGraph 内置的,自动完成"解析 tool_calls → 执行工具 → 将结果包装为 ToolMessage → 写回 State"这一整套流程
  4. add_edge(START, "call_model") — 入口:执行从 call_model 开始
  5. add_conditional_edges("call_model", should_continue, …) — 条件路由:call_model 执行完后,由 should_continue 根据 State 内容决定下一步。返回 "call_tool" 就去执行工具,返回 END 就结束
  6. add_edge("call_tool", "call_model")ReAct 循环的关键:工具执行完后回到 LLM 节点,让 LLM 看到工具返回的结果,判断是否需要再调工具还是可以给出最终答案
  7. .compile() — 编译为可执行对象

这 7 步就是把第 2 节的 ASCII 流程图翻译成了代码——START → call_model → 条件分叉 → call_tool → 循环回 call_model → END。

跑起来看看完整轨迹

from langchain_core.messages import HumanMessage

result = agent.invoke({
    "messages": [HumanMessage(content="帮我算一下 15 + 23 × 7")]
})

for msg in result["messages"]:
    print(f"[{msg.type}] {msg.content}")

输出:

[human] 帮我算一下 15 + 23 × 7
[ai]
[ai] tool_calls: [{'name': 'calculator_tool', 'args': {'expression': '15 + 23 * 7'}}]
[tool] 计算结果: 176
[ai] 15 + 23 × 7 = 176

这就是一个完整的 Agent 执行轨迹:用户输入 → LLM 判断用计算器 → 工具执行 → LLM 拿到结果给出答案。如果把输入换成"北京今天什么天气?",Agent 会自动切换为调用搜索工具。

写到这里,最容易卡住的几个点

忘记 Annotated,消息历史会被悄悄覆盖

# 错误写法:消息会被覆盖
class AgentState(TypedDict):
    messages: list[BaseMessage]  # 没有 Annotated + reducer

# 正确写法:消息会追加
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

如果忘记 Annotatedadd_messages,每次 Node 返回的新消息会覆盖旧消息,Agent 的上下文就会丢失。表现通常是 LLM "失忆"——明明刚才调了工具,下一轮却不记得结果。

条件边写错,Agent 很容易原地打转

如果你的 should_continue 逻辑有 bug(比如永远返回 "call_tool"),Agent 会陷入死循环。LangGraph 默认限制 25 步,超过会抛出 GraphRecursionError

agent.invoke() 的第二个参数(config 字典)中传入 recursion_limit 即可调整:

from langchain_core.messages import HumanMessage

# 排查时收窄,快速暴露问题
result = agent.invoke(
    {"messages": [HumanMessage(content="...")]},
    {"recursion_limit": 5},
)

# 复杂任务放开限制
result = agent.invoke(
    {"messages": [HumanMessage(content="...")]},
    {"recursion_limit": 50},
)

.env 读不到,通常是加载顺序或路径问题

如果你发现 DEEPSEEK_API_KEY 一直读取不到,检查两点:

  1. load_dotenv() 必须在 os.getenv() 之前调用
  2. .env 文件必须放在项目根目录(与 pyproject.toml 同级)

一个常见的低级错误是把 .env 放在了子目录里,或者 load_dotenv() 写在了 os.getenv() 之后。

如果你的 .env 不在当前工作目录下,可以指定路径:

from dotenv import load_dotenv
load_dotenv("/path/to/your/.env")

调试时我会先看这几样

LangGraph 提供了几种调试手段:

  • agent.get_graph().draw_mermaid_png():生成图结构的可视化图片,直观看到节点和边的拓扑关系
  • 检查中间 State:在 Node 函数里 print(state["messages"]) 看每一步的消息历史
  • 设置 debug=True:LangGraph 支持在编译时开启 debug 模式(取决于版本),或通过 invoke 时传入 config={"callbacks": [...]} 配合 LangSmith 做全链路追踪

最后回到 LangGraph 的核心

我们从裸 LLM 的能力边界出发,理解了为什么需要 Agent 和 LangGraph。用几个核心概念(State、Node、Edge、Conditional Edge)构建了一个完整的"计算 + 搜索"Agent,覆盖了从工具定义、状态管理、条件路由到最终运行的完整流程。

关键收获:

  • LangGraph 的核心是一个有向图:Node 做执行,Edge 做流转,State 做共享内存
  • 条件边是 Agent 的灵魂should_continue 这寥寥几行代码决定了 Agent 的行为模式
  • add_messages reducer 不能忘:否则消息历史丢失,Agent 变成"金鱼记忆"
  • 工具函数的错误处理很重要:把异常转成文本返回比让 Agent 崩溃好得多