作者：李剑锋

前言

在上一节课中（Agent" data-itemshowtype="0" linktype="text" data-linktype="2">万字长文！从零开始构建你的第一个 ReAct Agent），我们学习了如何不依赖任何框架，仅通过 Python 基础语法和外部 API 调用，手动实现了一个简易的 Agent 系统。我们通过列表维护消息上下文、编写正则匹配动作指令、循环控制执行流程等方式，实现了一个“Thought → Action → Observation → Answer”的基本逻辑。虽然这个过程帮助我们深入理解了 Agent 的核心运行机制，但也暴露出许多问题，例如流程控制复杂、状态管理困难、缺乏模块化与可视化支持等。

因此本节课中，我们将迈出关键一步，引入当前 Agent 生态中非常热门的框架——LangGraph。它基于“有向图”模型，将 Agent 的运行流程抽象为“节点 + 状态流转”，具备结构清晰、易扩展、原生支持多工具/多轮调用等显著优势。通过与上一节手写 Agent 的对比学习，你将切实体会到：借助 LangGraph，我们可以用更高效、更优雅的方式构建复杂智能体系统。

LangGraph 简介

LangGraph 最核心的设计理念就是将智能体流程图形化建模。在 LangGraph 中，每一个操作单元（比如大模型调用、函数执行、判断逻辑）都是一个 节点（Node），节点之间通过 边（Edge） 相连接，构成了完整的智能体工作流。特别是，它支持 条件边（Conditional Edges），可以根据状态决定分支路径，类似于传统的 if-else 语句。这样构建出来的智能体，就像画流程图一样简单、直观，极大地增强了结构可视性与逻辑解耦能力。

另外，LangGraph 并不只是“把流程画出来”这么简单，它还原生支持三个非常重要的功能：

• ✅ Cyclic Graphs（循环图）：支持智能体在不同节点之间循环跳转，方便实现多轮思考与行动；
• ✅ Persistence（状态持久化）：图中状态可以随时保存、恢复，支持断点续跑与回溯；
• ✅ Human-in-the-loop（人类参与环节）：支持在流程中插入人工确认、反馈或干预节点，增强系统可控性。

这些功能，若在传统实现中靠代码维护会非常复杂，而 LangGraph 提供了图层级的原生支持，几乎开箱即用。

代码实战

前期准备

首先我们可以沿用上一节课创建的环境（没看过的小伙伴可以点击链接查阅一下），但是我们需要额外在终端安装以下这些库：

pip install langgraph langchain langchain-openai langchain-community pygraphviz

然后我们还可以配置好大模型的环境，我们只需要写入api_key的信息即可：

from langchain_openai import ChatOpenAI
aliyun_api_key = '你的api_key'
model = ChatOpenAI(
    model="qwen-plus",  # 或其他你在 DashScope 控制台启用的模型名
    openai_api_key=aliyun_api_key,
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

假如在国外的朋友可以照着原本教程的代码配置Tavily Search，这个API密钥我们可以通过链接（(https://www.tavily.com/）去获取。

from langchain_community.tools.tavily_search import TavilySearchResults
import os
os.environ["TAVILY_API_KEY"] = "你的api_key"
tool = TavilySearchResults(max_results=4) #increased number of results
print(type(tool))
print([tool.name](http://tool.name/))

但是由于国内无法直接使用，因此这里我们可以替换成国内可用的一个搜查工具——博查。博查算是我发现国内可用里面比较方便快捷的网络搜索工具了，其他百度啊Bing这种用起来都比较复杂，博查算是充值就能用的了。我们可以在博查的官网（https://open.bochaai.com/）注册一个账号（可以直接微信登录）。

进去后就可以点击右上角的控制台找到账号充值的位置，然后充值10块钱进去尝试一下（真希望广告主能看看我们，这要是个广告位就不用我花钱了）。

然后再点击右边的API KEY管理生成一个属于我们自己的API KEY。

拿到这个API KEY以后呢，我们再点击开发文档（https://bocha-ai.feishu.cn/wiki/HmtOw1z6vik14Fkdu5uc9VaInBb）里看看怎么去获取到对应的信息和内容。博查里面其实有好几种不同的搜索方式，这里演示我就使用最普通的Web Search好了。

拿到了这个代码内容以后呢，我们其实就要把这个调用的方式封装成LangGraph支持的格式了。这里其实我们就可以去求助一下AI，让其帮我们改一下。那AI哐哧哐哧就写了一个工具出来（当然这里也要几轮的交互，最重要的其实是要把返回的数据结构给到AI，就是真正的通过上面的代码调用一次，把返回的数据结构传给AI，那AI就知道要怎么写了）。我们只需要在最下面写入API_KEY并看看返回的结果即可。

from typing import Type
from pydantic import BaseModel, Field, PrivateAttr
from langchain_core.tools import BaseTool
import requests

# 输入参数定义
class BoChaSearchInput(BaseModel):
    query: str = Field(..., description="搜索的查询内容")

# LangChain Tool 定义
class BoChaSearchResults(BaseTool):
    name: str = "bocha_web_search"
    description: str = "使用博查API进行网络搜索，可以用来查找实时信息或新闻"
    args_schema: Type[BaseModel] = BoChaSearchInput

    # 私有属性，用于保存调用参数
    _api_key: str = PrivateAttr()
    _count: int = PrivateAttr()
    _summary: bool = PrivateAttr()
    _freshness: str = PrivateAttr()

    # 初始化方法
    def __init__(self, api_key: str, count: int = 5, summary: bool = True, freshness: str = "noLimit", **kwargs):
        super().__init__(**kwargs)
        self._api_key = api_key
        self._count = count
        self._summary = summary
        self._freshness = freshness

    # Tool 实际运行逻辑
    def _run(self, query: str) -> str:
        url = "https://api.bochaai.com/v1/web-search"
        headers = {
            "Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "query": query,
            "summary": self._summary,
            "freshness": self._freshness,
            "count": self._count
        }

        try:
            response = requests.post(url, headers=headers, json=payload, timeout=10)
            response.raise_for_status()
            data = response.json()

            # ✅ 正确解析返回值中的搜索结果位置
            results = data.get("data", {}).get("webPages", {}).get("value", [])
            ifnot results:
                returnf"未找到相关内容。\n[DEBUG] 返回数据：{data}"

            # 格式化输出结果
            output = ""
            for i, item in enumerate(results[:self._count]):
                title = item.get("name", "无标题")
                snippet = item.get("snippet", "无摘要")
                url = item.get("url", "")
                output += f"{i+1}. {title}\n{snippet}\n链接: {url}\n\n"

            return output.strip()

        except Exception as e:
            returnf"搜索失败: {e}"

tool = BoChaSearchResults(api_key="你的博查API Key", count=4)

# 单次使用（模拟 Agent 工具调用）
result = tool.invoke({"query": "阿里巴巴2024年的ESG报告"})
print(result)

测试完没问题以后，我们下面就正式开始关于LangGraph的Agent构件了！

温馨提示一下，大家测试完记得把单次使用的代码删掉或者注释掉，不然每运行一次都返回结果的话也会浪费token的！

LangGraph Agent 构建

无论是节点（node）还是条件边（conditional edge），它们之所以能够实现特定功能，归根结底都依赖于函数的逻辑驱动来完成：

• 🔵 蓝色圆形节点 代表大模型的调用，例如 llm: call_openai
• 🟠 橙色圆形节点 表示工具的执行，例如 action: take_action
• 🔺 菱形条件节点 通过函数 exists_action 判断下一步是调用工具还是结束流程
• ⚫ 黑色终止节点 表示流程终点（END）
• 🟦 蓝色长方形框 表示智能体的“状态存储”（Agent State），用于在每一轮中保留对话历史和工具调用等关键信息

基于下图这个简洁示例，我们可以直观理解 LangGraph 的运行机制：用户提出问题 → 调用 LLM → 判断是否需要执行工具 → 如需执行则调用工具并将结果返回给 LLM → 重复判断，直到流程结束，将最终答案返回给用户。这个过程清晰高效，充分体现了 LangGraph 在构建多轮智能体流程中的优势。

AgentState 构建

在 LangGraph 中，图的运行围绕一个统一的Agent State进行。这个状态结构可以是简单的消息列表（如 messages），也可以是你自定义的复杂结构，包含：

• 用户输入内容（input）；
• 对话历史（chat_history）；
• 中间步骤与工具调用记录（intermediate_steps）；
• 逻辑判断需要的标志位等。

每个节点都可以读取或修改这个状态，从而实现智能体行为的记忆、调整与追踪。这种显式的状态流，不仅便于调试与维护，也为多轮对话、任务跟踪等高级功能打下了基础。

在实际构建 LangGraph 智能体时，我们首先需要定义一个 AgentState 类，用于描述图中各节点在运行过程中共享和修改的“状态结构”。最简单的 AgentState 定义如下，它只包含一个字段 messages，用于存储当前的消息列表：

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

我们来仔细分析这一段代码的含义。

首先，AgentState 继承自 TypedDict，这是 Python 中用于描述“字典结构”的一种类型注解方式，能够让每个字段拥有明确的名称和类型。默认情况下，Python 的字典是动态的，可以存放任意键值对，例如：

data = {"name": "小明", "age": 18}

虽然这种写法很灵活，但当我们构建一个空字典、或者用于多人协作时，别人很难一眼看出这个字典中应该包含哪些字段、每个字段又应该是什么类型。这时就可以通过 TypedDict 来提前约定字典结构，例如告诉开发者：“这个字典中必须有一个 messages 字段，它是一个列表，并且列表中的元素都是 AnyMessage 类型”。

来看字段定义部分：

messages: Annotated[list[AnyMessage], operator.add]

其中 messages 是字典的键，右侧的 Annotated[...] 是这个字段的类型说明。Annotated 是 Python 的类型提示工具，它允许我们在基本类型的基础上附加额外的元信息，这些信息虽然不会被 Python 自身使用，但可以被像 LangGraph 或 FastAPI 这样的框架所读取和利用。

比如下面的例子：

name: Annotated[str, "这是用户输入的名字"]

它本质上和 name: str 是等价的，但在此基础上，我们还给这个变量增加了注释信息，让外部工具能够识别“这个字段是做什么的”。

在 LangGraph 中，这个 Annotated 的作用非常关键。它告诉框架：如果多个节点都返回 messages 字段的内容，请使用 operator.add 来合并它们。而 operator.add 则是 Python 的内置函数，作用等价于 +，在这里主要用于列表的拼接操作：

operator.add([1, 2], [3, 4])  # 返回 [1, 2, 3, 4]

所以当多个节点输出了消息列表时，LangGraph 就会自动将这些列表合并为一个完整的消息历史。

其中 list[AnyMessage] 表示消息列表，AnyMessage 是 LangChain 中对所有消息类型的统一封装，它可以是用户消息（HumanMessage）、AI 回复（AIMessage）、系统提示（SystemMessage）或工具调用信息（ToolMessage）等。

通过这种机制，我们可以将每一步交互的上下文（包括系统提示、用户提问、AI 回复等）都保存在 AgentState["messages"] 中，从而让整个智能体始终保持清晰连贯的对话历史。

例如，在一次典型的调用过程中，AgentState 可能如下变化：

# 第一步：初始化系统提示词
AgentState = {
    "messages": [SystemMessage(content="你是一个有用的人工智能助手")]
}

# 第二步：用户发起提问
AgentState = {
    "messages": [
        SystemMessage(content="你是一个有用的人工智能助手"),
        HumanMessage(content="你好")
    ]
}

# 第三步：AI 生成回复
AgentState = {
    "messages": [
        SystemMessage(content="你是一个有用的人工智能助手"),
        HumanMessage(content="你好"),
        AIMessage(content="你好，我是通义千问大模型")
    ]
}

这样每次 LangGraph 中的节点运行时，都可以通过完整的消息历史做出更有上下文感知的判断或回应。

当然，AgentState 并不一定只包含 messages 一个字段。我们也可以根据需要拓展为更复杂的结构，例如：

class AgentState(TypedDict):
    input: str
    chat_history: list[BaseMessage]
    agent_outcome: Union[AgentAction, AgentFinish, None]
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

通过这种方式，我们就能更精准地控制智能体的中间状态、处理逻辑和返回结果。

另外，除了operate.add这一种操作的方式以外，其实还有其他的操作方式，比如说我们可以自己设置一个函数，让AgentState里每种类型的信息只保留一条，比如有新的AIMessage进来就只保留新的AIMessage，旧的就删掉这样。最后我们再在AgentState中的operate.add替换为reduce_message即可，具体的代码如下：

from uuid import uuid4
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage

"""
在之前的示例中，我们通常为 `messages` 这个状态字段设置默认的合并方式 `operator.add`（也就是使用 `+`），
它的作用是每次都将新消息追加到原来的消息列表末尾。

现在，为了支持“替换已有消息”的能力，我们为 `messages` 字段指定了一个自定义的合并函数 `reduce_messages`。
这个函数的功能是：如果新消息中有某条消息的 id 与已有消息重复，就用新消息替换旧的；否则就将其追加到末尾。
"""

# 自定义消息合并函数
def reduce_messages(left: list[AnyMessage], right: list[AnyMessage]) -> list[AnyMessage]:
    # 为新传入的消息设置唯一 id（如果还没有的话）
    for message in right:
        ifnot message.id:
            message.id = str(uuid4())  # 生成一个唯一的 UUID 作为 id

    # 拷贝一份旧的消息列表，用于合并操作
    merged = left.copy()

    # 遍历新消息
    for message in right:
        # 检查该新消息是否和旧消息中的某条 id 相同
        for i, existing in enumerate(merged):
            if existing.id == message.id:
                # 如果 id 相同，则用新消息替换旧消息
                merged[i] = message
                break
        else:
            # 如果没有重复 id，则将新消息追加到末尾
            merged.append(message)

    return merged

# 定义 Agent 的状态结构，使用自定义合并函数来处理 messages 字段
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], reduce_messages]

综上所述，AgentState 的本质就是构建智能体“流程图”中的状态存储核心。它通过类型约束明确字段结构，通过 Annotated 机制指导 LangGraph 的状态合并行为，是整个智能体正常运作的关键基础之一。接下来我们会进一步看看其他组件是如何与 AgentState 配合完成智能体任务流程的。

创建 Agent 框架

所谓创建框架，其实就是我们需要按照下面图片的样子搭建起Agent的基本节点内容：

这一切的起点，就是创建一个名为 Agent 的类，并在其中定义一个构造方法（def __init__:）。这个构造方法会在我们每次实例化 Agent 时自动被调用，用于初始化这个 Agent 的内部状态。我们后续要实现的节点配置、交互逻辑等核心内容，都会在这个构造方法中逐步完善。

class Agent:
    def __init__(self):

这里的 self 是 Python 中类的一个关键字，代表“当前实例对象的引用”。借助 self，我们可以在类的方法中访问并修改这个对象自身的属性或方法。

举个例子，如果我们希望为 Agent 设置一个系统提示词 system（默认为空字符串），就可以这样写：

class Agent:
    def __init__(self, system=""):
      self.system = system

abot = Agent("你是一个友好的机器人")
print(abot.system)  # 输出：你是一个友好的机器人

这里的 self.system = system 就是说：把这个system 储存在当前的 Agent 对象里，这样以后我们就可以通过 abot.name 来访问这个值。

接下来，我们将基于 LangGraph 来构建状态驱动的智能体流程图（graph）。这个流程图是智能体运行的核心框架，它由多个节点组成，每个节点都共享一个统一的状态结构。

构建流程图的第一步，就是使用 LangGraph 中的 StateGraph 类。它的初始化需要传入一个状态定义，也就是我们在上一节中构建好的 AgentState 类型结构。具体示例如下：

from langgraph.graph import StateGraph
class Agent:
    def __init__(self, system=""):
      self.system = system
      graph = StateGraph(AgentState)

当我们完成了 graph 的定义之后，接下来的工作，就是往这个流程图中逐步添加各类节点，定义它们之间的连接关系与状态流转逻辑。下面是一个典型的智能体初始化结构，我们在其中依次添加了几个关键节点：

from langgraph.graph import StateGraph, END

class Agent:
    def __init__(self, system=""):
        self.system = system
        graph = StateGraph(AgentState)

        # 添加大模型调用节点
        graph.add_node("llm", self.call_openai)

        # 添加工具执行节点
        graph.add_node("action", self.take_action)

        # 添加条件跳转逻辑：
        # 如果 llm 的输出中包含 tool_calls，就跳转到 action 节点执行；
        # 否则直接结束流程（END）
        graph.add_conditional_edges(
            "llm",
            self.exists_action,
            {True: "action", False: END}
        )

        # 执行完 action 节点后，再回到 llm 节点继续处理对话
        graph.add_edge("action", "llm")

我们来逐条解释一下这些步骤的作用：

• ✅ 添加节点：使用 graph.add_node("节点名称", 对应函数) 的方式，将某个逻辑单元注册为图中的一个节点。这里的 "llm" 和 "action" 就分别对应模型调用和工具调用两个核心逻辑。

• 🔀 添加条件跳转：通过 graph.add_conditional_edges(...) 方法，我们可以为某个节点添加“判断分支”。这个方法接受三个参数：

  其中 END 是 LangGraph 中的流程终止标志，表示不再进入其他节点。

1. 当前节点名（如 "llm"）；
2. 判断函数（如 self.exists_action），返回 True 或 False；
3. 判断结果对应的下一跳分支，用字典表示，例如 {True: "action", False: END}。

通过以上这几个步骤，我们就可以用代码将一个完整的对话流程以图的方式搭建起来了。LangGraph 会根据你定义的节点和边，自动控制整个智能体在运行时的流程调度。接下来我们将继续构建这些节点的具体功能函数，让它们真正动起来。

除了设置节点以外，其实我需要设置一些其他的内容，比如说我们要设置一下graph的起始点，不然代码就知道从哪开始了。还有就是我们现在的graph对象还只是一个草图，假如想要让其成为一个可执行的版本，我们还需要通过.compile()方法使其“编译”成最终执行版本。然后我们也需要去将所有能够使用的工具以及作为Agent大脑的模型接入进来，这样才能完整的构造出一个可真正被使用的模型。

class Agent:
    def __init__(self, model, tools, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges(
            "llm",
            self.exists_action,
            {True: "action", False: END}
        )
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile()
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

那我具体来讲讲新的这两部分：

• self.tools = {t.name: t for t in tools}

那在LangChain当中，一般来说我们传入工具都是一个列表，比如说长这样：

tools = [
    Tool(name="search_weather", func=search_weather_func),
    Tool(name="calculate", func=calculator_func)
]

那么这一句执行完后，self.tools 就是：

{
    "search_weather": Tool(...),
    "calculate": Tool(...)
}

这个字典的作用是在工具调用节点（例如 take_action()）中，通过：

tool = self.tools[t['name']]
result = tool.invoke(t['args'])

快速从名字找到对应工具并调用。

• self.model = model.bind_tools(tools)

这句话的作用就是把这些工具“绑定”到语言模型中，让模型在推理时知道可以调用哪些工具，并在需要时自动生成调用指令（tool_calls）。

在没有绑定工具前，模型只会生成纯文本回复：

AI: 今天天气晴，25度。

绑定工具后，模型可以生成结构化调用：

{
  "tool_calls": [
    {
      "name": "search_weather",
      "args": {
        "city": "北京"
      }
    }
  ]
}

所以这就是在初始阶段将model和tools进行绑定，告诉模型在后续的流程中到底有哪些工具可以进行使用。

这样我们初始的Agent就搭建完成啦！我们根据下图构建了控制整体内部记忆的AgentState，并且也构建起传入llm然后根据conditional_edge判断是否执行的流程，这样只需要我们将三个函数补充起来，我们就可以开启Agent的判定和使用啦！

大语言模型调用(call_openai)

我们接下来要编写的是智能体中“大模型调用”的逻辑函数，也就是 call_openai 方法，它的核心职责是：根据当前对话状态，调用语言模型并返回回复消息。

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

由于我们在 Agent 类的初始化过程中，已经传入并绑定好了语言模型 model，同时也完成了工具注册，因此这里我们无需再额外配置模型调用方式，只需要负责整理输入信息，并将其传递给模型即可。

以下是该方法的几个关键点说明：

• state: AgentState ：

大家可能会好奇，这是干嘛呢？怎么又有state又有AgentState，难道这不是重复了吗？

那学过函数的可能都知道，我们一般在这个括号里写入的是变量。这里其实就是代表函数输入有两个变量，一个是内部变量self，另一个就是需要传入的变量state。

这个冒号其实代表的含义是要传入一个和 AgentState 定义的结构相同的字典对象才行，否则编辑器或类型检查器（如 mypy）会报错。

• messages = state['messages']

这段代码其实就是字典检索的方法，根据'messages'这个键，把对应后面所有存有对话记录的列表获取出来。最后将其保存在变量messages里面，这时候messages里就存放了一个列表了。

• messages = [SystemMessage(content=self.system)] + messages

这一段是判断我们在初始化 Agent 时是否提供了系统提示词 self.system。如果有，就将其封装成一个 SystemMessage 对象，并添加到 messages 列表的最前面，确保语言模型在生成回复时，能优先读取这一指导信息。

举个例子，假设当前 messages 是空的，self.system = "你是一个友好的人工智能助手"，那执行完后结果如下：

messages = []
if True:
  messages = [SystemMessage(content='你是一个友好的人工智能助手')] + messages
  
# 最后的messages将变成:
messages = [SystemMessage(content='你是一个友好的人工智能助手')]

• message = self.model.invoke(messages)

最后呢，我们就可以调用存好在内部变量self里面的model，并使用LangChain常用的调用方法.invoke()进行调用，同时将添加了系统提示词的所有信息放入，最终让大模型生成回复。

最终返回的内容其实是一个LangChain格式的AIMessage，最后我们return回来的东西其实就是可以用于符合AgentState格式定义的{'messages': [message]}。后续就可以把这部分信息给添加到AgentState中。

这样，我们就完成了一个完整的语言模型调用流程： 从输入 state 提取上下文 → 整理消息 → 模型调用 → 生成并返回新消息，整个过程简洁高效。

工具使用判定（exists_action）

在这个conditional_edge里，我们需要判断是要执行action节点。那我们怎么知道要不要执行工具呢？其实还是通过存在AgentState里面的内容来进行判定。

还记得我们前面讲过，大模型的返回内容里如果包含工具调用的提示，那就说明它打算用工具，那我们就跳转去执行 action 节点；反之则结束流程，走向 END。

    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

那从这个代码里也能看出，我们也是找到AgentState里最后一条传入的信息（[-1]就代表找的是最新的一条），然后使用LangGraph里的.tool_calls的方法去看到底有几条方法可以调用。

只要其数量（用len()函数来计算）大于0，那就返回True。要是等于0的话（当然不可能＜0），那就返回False。这样我们也就算是实现了控制了。

那我们可以重点看看这里的.tool_calls是干嘛用的：

其实.tool_calls 是大模型请求调用外部工具时，生成的结构化调用指令。也就是说，模型不仅能说话，还能“指挥”工具运行，它通过 .tool_calls 来告诉你。比如我们可以让通过提示词AI返回的内容写成下面的格式：

AIMessage(
    content="",  # 如果模型想调用工具，这里一般是空
    tool_calls=[
        {
            "id": "tool_call_1",
            "name": "search_weather",
            "args": {"city": "北京"}
        }
    ]
)

这里面的content就是模型回复的文本内容，假如调用工具的时候就不回复文本内容而是返回tool_calls的内容。那这个时候我们去看result.tool_calls其实就是获取了以下这个列表：

[
        {
            "id": "tool_call_1",
            "name": "search_weather",
            "args": {"city": "北京"}
        }
    ]
)

只要这个列表里的字典数量大于0，那就返回True，不然就返回False。很显然这里出现了一个，所以返回的len(result.tool_calls)应该就等于1，那1>0自然返回的就是True，去执行action节点了！

执行模型（take_action）

那最后一个我们需要写的函数就是执行模型了，其实和上面类似，第一步我们还是要获取到底哪些工具需要被调用。然后一个个去执行并获取结果。假如工具不存在，那就返回警告信息，假如没有问题就正常拿去执行即可。

def take_action(self, state: AgentState):
    # 从当前状态中取出最新一条消息的 tool_calls（即模型请求调用的工具列表）
    tool_calls = state['messages'][-1].tool_calls
    results = []  # 用于存放每个工具调用的执行结果（最终会作为 ToolMessage 返回）

    # 遍历每一个工具调用请求
    for t in tool_calls:
        print(f"Calling: {t}")  # 打印当前调用的信息，便于调试观察

        # 如果调用的工具名称不在 self.tools 中，说明模型调用了一个无效的工具
        ifnot t['name'] in self.tools:
            print("\n ....bad tool name....")  # 打印警告信息
            result = "bad tool name, retry"    # 返回提示信息，通知模型重试
        else:
            # 调用注册的实际工具函数，并传入模型提供的参数
            result = self.tools[t['name']].invoke(t['args'])

        # 将执行结果打包成一个 ToolMessage（LangChain 约定格式），用于反馈给模型
        results.append(
            ToolMessage(
                tool_call_id=t['id'],  # 对应这次调用的 ID
                name=t['name'],        # 工具名称
                content=str(result)    # 工具返回的结果（转成字符串）
            )
        )

    print("Back to the model!")  # 调用结束，准备将结果送回模型进行下一轮回复
    return {'messages': results}  # 返回格式为一个包含 ToolMessage 列表的 dict，LangGraph 会自动加进 state['messages']

最后我们返回的内容就是一个ToolMessage的列表，展示其完整的情况。

这里可以举一个简单的例子，假设我们获取到的工具就是像上面一样的搜索温度的例子：

tool_calls = [
    {
        "id": "call1",
        "name": "search_weather",
        "args": {"city": "北京"}
    }
]

查找工具 search_weather，执行：

search_weather("北京")  → "北京 今天晴，气温26℃

然后包装成 ToolMessage：

ToolMessage(
    tool_call_id="call1",
    name="search_weather",
    content="北京 今天晴，气温26℃"
)

最后这个信息也将保存到AgentState中等待进一步的运行。

实例创建及可视化

在我们构建出来这个Agent的框架以后，就可以进行创建了。当然在此之前我们需要先配置好提示词prompt。

prompt = """你是一名聪明的科研助理。可以使用搜索引擎查找信息。
你可以多次调用搜索（可以一次性调用，也可以分步骤调用）。
只有当你确切知道要找什么时才去检索信息。
如果在提出后续问题之前需要先检索信息，你也可以这样做！
"""

然后就可以将创建Agent类必要的三个内容一并传入：

abot = Agent(model, [tool], system=prompt)

然后假如是在Juypter Notebook去运行的话可以运行下面的代码打印出当前Agent的整体框架图：

from IPython.display import Image

Image(abot.graph.get_graph().draw_png())

假如本地运行安装不了pygraphviz库的话我们可以通过以下代码简单看到其整体的框架：

ascii_diagram = abot.graph.get_graph().draw_ascii()
print(ascii_diagram)

实际调用

在确保我们的模型创建好了以后，我们可以开始进行实际调用的环节了。

首先我们需要通过HumanMessage写入我们的问题，然后也是通过abot.graph.inoke的方式传入我们的问题给到我们创建的Agent类，最终我们可以打印出对应的结果：

messages = [HumanMessage(content="What is the weather in sf?")]
result = abot.graph.invoke({"messages": messages})
print(result['messages'][-1].content)

我们可以看到，模型并未虚构了一个气候的信息给我们，相反其真的调用了我们设置的工具bocha_web_search，并且得到了准确的信息！因此我们可以看到这个Agent流程是真实可用的！

完整代码如下所示：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Type
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field, PrivateAttr
from langchain_core.tools import BaseTool
import requests

# ----------------------------- Tool 定义 -----------------------------
class BoChaSearchInput(BaseModel):
    query: str = Field(..., description="搜索的查询内容")

class BoChaSearchResults(BaseTool):
    name: str = "bocha_web_search"
    description: str = "使用博查API进行网络搜索，可用来查找实时信息或新闻"
    args_schema: Type[BaseModel] = BoChaSearchInput

    _api_key: str = PrivateAttr()
    _count: int = PrivateAttr()
    _summary: bool = PrivateAttr()
    _freshness: str = PrivateAttr()

    def __init__(self, api_key: str, count: int = 5, summary: bool = True, freshness: str = "noLimit", **kwargs):
        super().__init__(**kwargs)
        self._api_key = api_key
        self._count = count
        self._summary = summary
        self._freshness = freshness

    def _run(self, query: str) -> str:
        url = "https://api.bochaai.com/v1/web-search"
        headers = {
            "Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "query": query,
            "summary": self._summary,
            "freshness": self._freshness,
            "count": self._count
        }

        try:
            response = requests.post(url, headers=headers, json=payload, timeout=10)
            response.raise_for_status()
            data = response.json()

            results = data.get("data", {}).get("webPages", {}).get("value", [])
            ifnot results:
                returnf"\u672a\u627e\u5230\u76f8\u5173\u5185\u5bb9\u3002\n[DEBUG] \u8fd4\u56de\u6570\u636e\uff1a{data}"

            output = ""
            for i, item in enumerate(results[:self._count]):
                title = item.get("name", "\u65e0\u6807\u9898")
                snippet = item.get("snippet", "\u65e0\u6458\u8981")
                url = item.get("url", "")
                output += f"{i+1}. {title}\n{snippet}\n\u94fe\u63a5: {url}\n\n"

            return output.strip()

        except Exception as e:
            returnf"\u641c\u7d22\u5931\u8d25: {e}"

# ----------------------------- 智能体状态定义 -----------------------------
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

# ----------------------------- Agent 实现 -----------------------------
class Agent:

    def __init__(self, model, tools, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile()
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            print(f"Calling: {t}")
            ifnot t['name'] in self.tools:
                print("\n ....bad tool name....")
                result = "bad tool name, retry"
            else:
                result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("Back to the model!")
        return {'messages': results}

# ----------------------------- 模型和启动 -----------------------------
prompt = """\
你是一名聪明的科研助理。可以使用搜索引擎查找信息。
你可以多次调用搜索（可以一次性调用，也可以分步骤调用）。
只有当你确切知道要找什么时，才去检索信息。
如果在提出后续问题之前需要先检索信息，你也可以这样做！
"""

aliyun_api_key = '你的api_key'
model = ChatOpenAI(
    model="qwen-plus",
    openai_api_key=aliyun_api_key,
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

tool = BoChaSearchResults(api_key="你的api_key", count=4)
abot = Agent(model, [tool], system=prompt)
# 绘制当前的 Graph
# ascii_diagram = abot.graph.get_graph().draw_ascii()
# print(ascii_diagram)
# ----------------------------- 测试使用 -----------------------------
messages = [HumanMessage(content="What is the weather in sf?")]
result = abot.graph.invoke({"messages": messages})
print(result['messages'][-1].content)

总结

在本节课中，我们成功将传统手搓 Agent 流程迁移到了 LangGraph 框架中，通过节点（Node）、边（Edge）和条件（Conditional Edge）构建出一个结构清晰、逻辑闭环的智能体流程图。所有对话状态被统一封装在 AgentState 中，实现了消息上下文的自动累积；工具调用与模型推理被分离为专职节点，大大提升了流程的可控性与可扩展性。同时，通过 model.bind_tools()，我们实现了模型自动识别何时该调用工具（如博查搜索），真正迈入了“让大模型主动思考再行动”的智能体范式。

其实平心而论，相比上节课我们手搓的Agent，其实LangGraph的流程显得是更复杂一些，因为里面用到的一些方法并不是我们日常学习的一些简单的方法。但是当我们继续往下学习我们会发现，只要我们熟悉了这种用法，这些方法在拓展性上会比起我们手搓要好得多，那我们下节课再见啦，886。

关注机智流并加入 AI 技术交流群，不仅能和来自大厂名校的 AI 开发者、爱好者一起进行技术交流，同时还有HuggingFace每日精选论文与顶会论文解读、Talk分享、通俗易懂的Agent知识与项目、前沿AI科技资讯、大模型实战教学活动等。

在「机智流」公众号后台回复下方标红内容即可加入对应群聊：

• cc | 大模型技术交流群
• hf | HuggingFace 高赞论文分享群
• 具身 | 具身智能交流群
• 硬件 | AI 硬件交流群
• 智能体 | Agent 技术交流群