从零开始，用 LangGraph 搭建你的第一个 AI Agent

过去我们调用大模型，通常只有一个固定流程：

用户提问，模型回答。

但真正的 Agent 不只是“会聊天”。它需要能够分析任务、选择工具、执行操作、观察结果，并根据新的信息决定下一步该做什么。

这篇文章将带你从零开始，用 Python 和 LangGraph 搭建一个具备以下能力的 AI Agent：

  • 自动判断是否需要调用工具
  • 执行工具并读取结果
  • 根据工具结果继续推理
  • 记住同一会话中的历史消息
  • 通过流程图控制执行逻辑

即使你之前没有接触过 LangGraph，也可以跟着本文一步一步完成。

一、什么是 LangGraph？

LangGraph 是一个用于构建有状态 Agent 的编排框架。

普通的 LLM 应用更像一条直线：

用户输入 → 调用模型 → 返回答案

而 Agent 的执行过程往往包含循环和分支：

json

1
2
3
4
5
6
7
8
9
10
11
用户输入
   ↓
模型判断
   ↓
是否需要工具？
   ├─ 否 → 返回答案
   └─ 是 → 调用工具
              ↓
          获取工具结果
              ↓
          再次交给模型

LangGraph 的作用，就是把这个过程明确地描述成一张图。

理解 LangGraph，只需要先掌握三个概念：

\1. State

State 是 Agent 当前掌握的状态，例如：

  • 用户发送的消息
  • 模型生成的回答
  • 工具执行结果
  • 当前任务进度
  • 用户信息

你可以把它理解成 Agent 的共享记忆。

\2. Node

Node 是 Agent 可以执行的一个步骤。

例如：

  • 调用大模型
  • 查询数据库
  • 搜索网页
  • 执行 Python 函数
  • 请求人工审批

通常，一个节点只负责一件事情。

\3. Edge

Edge 决定节点执行完成之后，下一步应该去哪里。

普通 Edge 表示固定跳转，Conditional Edge 则可以根据当前状态动态选择下一步。

这正是 Agent 能够自主决策的关键。

二、创建项目

首先创建项目目录：

bash

1
2
mkdir langgraph-agent
cd langgraph-agent

创建 Python 虚拟环境：

bash

1
python -m venv .venv

在 macOS 或 Linux 中激活环境：

bash

1
source .venv/bin/activate

Windows PowerShell：

bash

1
.venv\Scripts\activate

安装需要的依赖：

bash

1
pip install -U langgraph langchain-openai

接下来设置 OpenAI API Key。

macOS 或 Linux：

export OPENAI_API_KEY=”你的 API Key”

Windows PowerShell：

$env

:OPENAI_API_KEY=”你的 API Key”

不要把 API Key 直接写进代码，更不要提交到公开的 GitHub 仓库。

三、创建 Agent 工具

在项目中创建一个名为

agent.py

的文件。

首先导入依赖并初始化模型：

python

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from typing import Literal

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import (
    StateGraph,
    MessagesState,
    START,
    END,
)
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver


llm = ChatOpenAI(
    model="gpt-5.4-mini",
    temperature=0,
)

这里使用的模型需要支持 Tool Calling。你也可以替换成其他支持工具调用的模型。

接下来定义一个乘法工具：

python

1
2
3
4
@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。"""
    return a * b

@tool

会把普通 Python 函数转换成模型可以调用的工具。

函数名、参数类型和说明文字都很重要，因为模型会根据这些信息判断：

  • 这个工具能做什么
  • 什么时候应该调用
  • 调用时应该传入哪些参数

然后把工具绑定到模型：

python

1
2
tools = [multiply]
llm_with_tools = llm.bind_tools(tools)

这一步相当于告诉模型：

“当你遇到需要计算乘法的任务时，可以调用这个函数。”

需要注意的是，绑定工具并不代表工具会自动执行。模型只负责生成调用工具的请求，真正执行工具还需要后面的 ToolNode。

四、创建模型节点

接下来创建一个负责调用大模型的节点：

python

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def call_model(state: MessagesState):
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个严谨的 AI 助手。"
                "遇到乘法计算时，必须调用工具，"
                "不要自己猜测计算结果。"
            ),
        },
        *state["messages"],
    ]

    response = llm_with_tools.invoke(messages)

    return {
        "messages": [response]
    }

这个节点完成了三件事：

  1. 从 State 中读取历史消息
  2. 把消息发送给模型
  3. 把模型回复写回 State

MessagesState 是 LangGraph 提供的一种内置状态结构，它会自动管理消息列表。

如果模型认为不需要调用工具，它会直接生成最终回答。

如果模型认为需要调用工具，返回的消息中就会包含 tool_calls。

五、创建路由逻辑

我们需要检查模型的返回结果，并决定下一步应该去哪里：

python

1
2
3
4
5
6
7
8
9
10
def route(
    state: MessagesState,
) -> Literal["tools", "end"]:

    last_message = state["messages"][-1]

    if last_message.tool_calls:
        return "tools"

    return "end"

路由逻辑非常简单：

  • 如果最后一条模型消息包含工具调用请求，进入工具节点
  • 如果没有工具调用请求，结束整个流程

这就是 Agent 的决策分支。

六、组装 LangGraph

现在把模型节点、工具节点和路由连接起来：

python

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
builder = StateGraph(MessagesState)

builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "model")

builder.add_conditional_edges(
    "model",
    route,
    {
        "tools": "tools",
        "end": END,
    },
)

builder.add_edge("tools", "model")

逐行解释一下。

首先注册两个节点：

python

1
2
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))

model 节点负责调用大模型。

tools 节点负责执行模型请求的工具，并把执行结果转换成 ToolMessage。

接着设置入口：

python

1
builder.add_edge(START, "model")

每次运行 Agent，都先进入模型节点。

然后加入条件路由：

python

1
builder.add_conditional_edges(...)

模型调用完成后，LangGraph 会执行 route 函数。

如果模型请求工具，就进入 tools；否则进入 END。

最后，让工具节点重新连接到模型节点：

python

1
builder.add_edge("tools", "model")

这行代码非常关键。

工具只负责返回原始结果。模型还需要读取这个结果，并把它组织成用户能理解的最终回答。

因此，完整循环是：

模型 → 工具 → 模型

如果第二次调用模型后又产生了新的工具请求，这个循环还可以继续运行。

七、给 Agent 加入记忆

创建一个内存 Checkpointer：

python

1
memory = InMemorySaver()

然后编译 Graph：

python

1
2
3
agent = builder.compile(
    checkpointer=memory
)

Checkpointer 会保存每次运行后的 Graph State。

为了区分不同会话，我们还需要设置 thread_id：

python

1
2
3
4
5
config = {
    "configurable": {
        "thread_id": "user-001"
    }
}

同一个 thread_id 会继续之前的会话。

不同的 thread_id 则代表不同用户或不同对话。

八、第一次运行 Agent

向 Agent 提交一个计算任务：

python

1
2
3
4
5
6
7
8
9
10
11
12
13
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "23 乘以 19 等于多少？",
            }
        ]
    },
    config,
)

print(result["messages"][-1].content)

运行程序：

bash

1
python agent.py

在这个过程中，Agent 实际执行了以下步骤：

  1. 用户询问“23 乘以 19”
  2. 模型判断这是一个乘法任务
  3. 模型生成 multiply 工具调用
  4. ToolNode 执行 multiply(23, 19)
  5. 工具返回 437
  6. LangGraph 把结果重新交给模型
  7. 模型生成最终回答

这就是一个最基础的 ReAct 循环：

思考 → 行动 → 观察 → 回答

九、测试 Agent 的记忆

接下来继续发送一条消息：

python

1
2
3
4
5
6
7
8
9
10
11
12
13
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "把刚才的结果再乘以 2。",
            }
        ]
    },
    config,
)

print(result["messages"][-1].content)

我们没有在新消息里重新告诉 Agent“刚才的结果是 437”。

但因为两次调用使用了相同的 thread_id，Checkpointer 保存了之前的消息状态，所以 Agent 能够理解“刚才的结果”指的是什么。

如果将 thread_id 改成另一个值：

python

1
2
3
4
5
config = {
    "configurable": {
        "thread_id": "user-002"
    }
}

LangGraph 就会创建一段新的会话。

十、完整代码

最终的

agent.py

如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from typing import Literal

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import (
    StateGraph,
    MessagesState,
    START,
    END,
)
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver


llm = ChatOpenAI(
    model="gpt-5.4-mini",
    temperature=0,
)


@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。"""
    return a * b


tools = [multiply]
llm_with_tools = llm.bind_tools(tools)


def call_model(state: MessagesState):
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个严谨的 AI 助手。"
                "遇到乘法计算时，必须调用工具，"
                "不要自己猜测计算结果。"
            ),
        },
        *state["messages"],
    ]

    response = llm_with_tools.invoke(messages)

    return {
        "messages": [response]
    }


def route(
    state: MessagesState,
) -> Literal["tools", "end"]:

    last_message = state["messages"][-1]

    if last_message.tool_calls:
        return "tools"

    return "end"


builder = StateGraph(MessagesState)

builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "model")

builder.add_conditional_edges(
    "model",
    route,
    {
        "tools": "tools",
        "end": END,
    },
)

builder.add_edge("tools", "model")


memory = InMemorySaver()

agent = builder.compile(
    checkpointer=memory
)


config = {
    "configurable": {
        "thread_id": "user-001"
    }
}


result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "23 乘以 19 等于多少？",
            }
        ]
    },
    config,
)

print(result["messages"][-1].content)


result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "把刚才的结果再乘以 2。",
            }
        ]
    },
    config,
)

print(result["messages"][-1].content)

十一、接下来可以怎样扩展？

现在这个 Agent 只有一个乘法工具，但它的结构已经完整。

你可以继续增加：

  • 网页搜索工具
  • 天气查询工具
  • 数据库查询工具
  • 邮件发送工具
  • 文件读取工具
  • 日历管理工具
  • 企业内部 API
  • RAG 知识库检索

只需要定义新的工具，并加入 tools 列表：

tools = [ multiply, search_web, query_database, send_email, ]

模型就可以根据用户的任务，自主决定调用哪个工具。

你还可以继续给 Graph 增加新的节点，例如：

  • 任务规划节点
  • 内容审核节点
  • 人工审批节点
  • 结果验证节点
  • 错误恢复节点

这也是 LangGraph 相比普通聊天机器人更强的地方：每一步执行逻辑都可以被明确控制。

十二、从 Demo 到生产环境

本文使用的 InMemorySaver 只适合本地学习和测试。

程序停止之后，内存中的状态就会消失。生产环境通常需要使用数据库支持的 Checkpointer，例如 PostgreSQL。

除此之外，一个真正可上线的 Agent 还需要考虑：

工具权限

不要让 Agent 拥有超出任务需要的权限。

例如，一个只需要查询订单的 Agent，不应该拥有删除订单或修改用户信息的权限。

参数校验

不要直接信任模型生成的工具参数。

在执行数据库、文件、付款、邮件等操作之前，必须验证输入。

循环限制

Agent 可能因为模型判断错误而反复调用工具，因此需要设置：

  • 最大执行步数
  • 超时时间
  • Token 限制
  • 费用限制

人工审批

发送邮件、删除文件、修改数据库、执行付款等高风险操作，不应该完全自动完成。

可以使用 LangGraph 的中断机制，在关键节点暂停执行，等待用户确认后再继续。

日志与追踪

普通应用出错时，可以查看函数调用栈。

Agent 出错时，还需要知道：

  • 模型当时收到了什么
  • 为什么选择某个工具
  • 工具返回了什么
  • 状态在哪个节点发生了变化

因此，完整的 Trace 对 Agent 调试非常重要。

结语

到这里，我们已经完成了一个最小但完整的 LangGraph Agent。

它拥有：

  • 大模型决策
  • 工具调用
  • 条件路由
  • 循环执行
  • 会话状态
  • 短期记忆

LangGraph 最值得学习的地方，不是让 Agent 看起来更“智能”，而是让它的执行过程变得可控制、可观察、可暂停，也更容易测试。

不要把 Agent 想象成一个神秘的黑盒。

把复杂任务拆成状态、节点和边，再明确规定每一步如何执行，你就能逐渐构建出真正可靠的 Agent 系统。

官方参考：