1. 本章目标 #

第 2 章已经用普通函数当过工具；第 4 章手写过 bind_tools + ToolMessage 循环。
那些示例能跑，但业务工具一多就会碰到：

可以把问题分成两半：

问题侧 典型症状 本章对应节
定义侧 选型错、参数飘 §3～5
挂载 / 执行侧 调了没执行、执行崩了 §6～9

本章把「工具」本身讲透：

用 @tool 定义可复用业务工具：写清 schema、学会 bind_tools、处理好错误，并组装成业务工具包。

学完你应能：

本章有几条实测结论会让人意外，先摆在这里，读到对应小节时会有完整证据：

你可能以为 实际情况 见
docstring 里写了 Args:，模型就能看到每个参数的说明 默认整段 docstring 塞进描述，参数一个 description 都没有 §3.3
工具名随便起，中文也行 必须匹配 ^[a-zA-Z0-9_-]+$，中文名会被 API 用 400 拒掉 §4.2
args_schema 字段名和函数参数名不一致会报错 构造时不报，等到真的调用才 TypeError §5.4
工具抛异常，Agent 会告诉用户「查询失败」 不做处理时 agent.invoke() 直接把异常抛给你，进程崩 §9.1
return_direct=True 后最后一条消息还是 AIMessage 是 ToolMessage §8.1

参考文档：

2. 工具在系统里的位置 #

工具 = 模型可请求、由你的代码真正执行的外部能力。
模型负责「要不要调、调哪个、参数是什么」；工具函数负责「查库 / 调 API / 算数 / 读配置」。

容易混的一点：模型发出 tool_calls 不等于已经查到天气或订单——那只是「调用意图」。真正执行发生在你的函数或驾驭层（create_agent）里，结果再以 ToolMessage 回灌。

打个比方：模型是个不能离开座位的顾问。它手里有一本工具目录（名字、说明、要填哪些字段），能做的只是填一张申请单递出来——「我要调 get_weather，参数是 {"city": "北京"}」。真正跑腿的是你的代码；跑完把结果写在纸上递回去（ToolMessage），它才知道北京到底什么天气。

这个比方能解释很多现象：

用户问题
   ↓
模型（看到工具 schema）──► 发出 tool_calls
   ↓
执行工具函数 ──► ToolMessage（结果回灌）
   ↓
模型组织最终回答

再补一个初学常忽略的事实：一条 AIMessage 里可以同时带多个 tool_calls。用户一句话里问了三件事，模型往往会一次提出三个调用，接着你会看到连续三条 ToolMessage。§10.2 有真实轨迹。所以「执行工具」这一步天然是个循环，不是单次。

三条边界记牢：

角色 做什么 不做什么
模型 选型、填参、组织最终回答 真的访问数据库 / HTTP
工具函数 执行业务逻辑、返回观察结果 代替模型做「该不该调」的决策
驾驭层 / 你的循环 调度执行、对齐 tool_call_id、维护 messages 替你写业务规则

两条常见挂载路径：

路径 写法 谁执行工具 何时用
Agent（推荐日常） create_agent(..., tools=[...]) 驾驭层（create_agent）自动执行并回灌 业务默认
模型绑定（细控） model.bind_tools([...]) 你自己执行并构造 ToolMessage 教学、调试、特殊编排

第 2 章走的是 Agent 路径；第 4 章演示过手工路径。本章两边都会写：先把工具定义好，再谈怎么挂。定义质量差，两条路径都会一起翻车。

3. 用 @tool 定义工具 #

最简方式：给函数加 @tool。装饰之后，它不再只是普通 callable，而是带有 name / description / args 等元数据的工具对象，可直接交给模型或 Agent。

默认规则：

模型「看见」的主要是这三样，而不是函数体里的实现细节。所以：说明书（名 + 描述 + schema）比实现注释更影响行为。

3.1. 1.basic_tool.py #

先学会「定义 → 打印元数据 → 本地 invoke」，再接模型。

# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool


# @tool 把普通函数包装成工具对象（BaseTool 的子类实例）
@tool
# query 是必填参数（无默认值）；limit 有默认值 10，因此模型可以不填
def search_database(query: str, limit: int = 10) -> str:
    # 这段 docstring 会被整段（含下面的 Args 部分）当成工具描述
    """在客户库中按关键词检索记录。

    Args:
        query: 要搜索的关键词
        limit: 最多返回多少条
    """
    # 模拟检索结果（真实项目里这里换成查数据库的代码）
    return f"找到与「{query}」相关的 {limit} 条记录。"


# 工具名：默认取函数名
print("name:", search_database.name)
# 工具描述：默认取整段 docstring
print("description:", search_database.description)
# 参数 schema：由类型注解推断出来的字典
print("args:", search_database.args)

# 不经过模型，直接本地调用工具——这是给工具做单元测试的标准方式
print("invoke:", search_database.invoke({"query": "张三", "limit": 3}))

输出：

name: search_database
description: 在客户库中按关键词检索记录。

Args:
    query: 要搜索的关键词
    limit: 最多返回多少条
args: {'query': {'title': 'Query', 'type': 'string'}, 'limit': {'default': 10, 'title': 'Limit', 'type': 'integer'}}
invoke: 找到与「张三」相关的 3 条记录。

先看懂这段输出——下一小节的「意外」就藏在这里：

换句话说，你在 docstring 里辛苦写的「要搜索的关键词」，并没有落到 query 这个字段上。

要点：

  1. 先本地 tool.invoke({...}) 测通，再交给模型——把「业务逻辑错」和「模型选型错」分开。
  2. docstring 写清「干什么、何时用」；参数名用业务语义（order_id 比 x 好）。
  3. invoke 入参用字典（键=参数名）。传裸字符串也能用，但有条件——见下面的补充。

关于 invoke 传裸字符串：search_database.invoke("张三") 其实能跑，会被当成 {"query": "张三"}（第一个参数），limit 用默认值 10。但这只在只有一个必填参数时成立。如果有两个必填参数，就会报 ValidationError: Field required。类型也会自动转换，int_tool.invoke("5") 能得到 5。

结论：统一用字典。裸字符串的便利只在最简单的情况下有效，而且读代码时看不出参数名是什么。

3.2. 模型到底收到了什么 #

args 是 LangChain 给你看的简化视图。想知道发给 API 的原始 JSON 长什么样，用 convert_to_openai_tool 转一下：

# 导入 json 用于格式化打印
import json

# 这个函数把 LangChain 工具转成 OpenAI 风格的工具定义（各家 API 的通用格式）
from langchain_core.utils.function_calling import convert_to_openai_tool
# 导入 tool 装饰器
from langchain.tools import tool


# 定义一个工具用于观察
@tool
# 和 §3.1 完全相同的工具
def search_database(query: str, limit: int = 10) -> str:
    # 整段 docstring 会成为工具描述
    """在客户库中按关键词检索记录。

    Args:
        query: 要搜索的关键词
        limit: 最多返回多少条
    """
    # 返回模拟结果
    return f"找到与「{query}」相关的 {limit} 条记录。"


# 转换并打印：ensure_ascii=False 保证中文正常显示，indent=2 便于阅读
print(json.dumps(convert_to_openai_tool(search_database), ensure_ascii=False, indent=2))

输出：

{
  "type": "function",
  "function": {
    "name": "search_database",
    "description": "在客户库中按关键词检索记录。\n\nArgs:\n    query: 要搜索的关键词\n    limit: 最多返回多少条",
    "parameters": {
      "properties": {
        "query": {
          "type": "string"
        },
        "limit": {
          "default": 10,
          "type": "integer"
        }
      },
      "required": [
        "query"
      ],
      "type": "object"
    }
  }
}

这就是模型侧「工具目录」的原文。三件事值得注意：

  1. description 是一整块纯文本，Args: 和换行符都在里面
  2. properties 里两个参数都只有类型，没有说明
  3. required 只有 query——limit 因为有默认值而变成可选

第 6 章讲结构化输出时看过类似的 JSON Schema，机制同一套：Pydantic/类型注解 → JSON Schema → 随请求发给模型。

3.3. 坑：Args: 段默认不会变成参数说明 #

在 docstring 里写 Google 风格的 Args: 段，容易以为模型能读到每个参数的说明。默认并不会。 LangChain 只是把整段 docstring 原样当作工具描述，参数的 description 是空的。

后果是：参数说明混在一大段描述里，模型得自己猜哪句对应哪个字段。参数少时通常还能猜对；字段一多（尤其是几个含义相近的字符串字段）就开始飘。

想让 Args: 真正生效，加 parse_docstring=True：

# 导入 json 用于格式化打印
import json

# 导入 tool 装饰器
from langchain.tools import tool


# parse_docstring=True 让 LangChain 解析 Google 风格 docstring
@tool(parse_docstring=True)
# 参数签名不变
def search_v2(query: str, limit: int = 10) -> str:
    # 第一行摘要会成为工具描述；Args 段会被拆到各个参数上
    """在客户库中按关键词检索记录。

    Args:
        query: 要搜索的关键词
        limit: 最多返回多少条
    """
    # 返回模拟结果
    return f"找到与「{query}」相关的 {limit} 条记录。"


# 描述里只剩摘要，Args 段已被剥离
print("description:", repr(search_v2.description))
# 每个参数上都多了 description
print("args:", json.dumps(search_v2.args, ensure_ascii=False, indent=2))

输出：

description: '在客户库中按关键词检索记录。'
args: {
  "query": {
    "description": "要搜索的关键词",
    "title": "Query",
    "type": "string"
  },
  "limit": {
    "default": 10,
    "description": "最多返回多少条",
    "title": "Limit",
    "type": "integer"
  }
}

差别一目了然：

默认 parse_docstring=True
description 整段 docstring（含 Args: 与缩进） 只有摘要行
参数的 description 无 逐个填好

两条路都能走通，选一条并保持一致：

做法 适合
@tool(parse_docstring=True) + Google 风格 docstring 参数不多、想少写代码
Pydantic args_schema + Field(description=...)（§5.2） 业务工具、需要枚举/约束/默认值

最差的是这种中间状态：写了 Args: 却没加 parse_docstring=True，自以为模型看到了参数说明，其实没有。

用 parse_docstring=True 时，下面两种情况会报错（实测确认过）：

情况 结果
docstring 只有一句话，没有 Args: 段 ValueError: Found invalid Google-Style docstring.
Args: 里写了函数签名中不存在的参数 ValueError: Arg nonexistent in docstring not found in function signature.
Args: 只标注了部分参数 正常，未标注的参数就是没有 description

第一条最容易撞上：给原本只有单行 docstring 的老工具加上 parse_docstring=True，会直接报错。要么补齐 Args: 段，要么别加这个参数。

好消息：这两个失败都发生在装饰器执行时——导入模块的瞬间就炸，不会拖到线上，属于「响亮的失败」。

3.4. 两个边界情况 #

不写类型注解：不报错，但 schema 会静默降级。

# 导入 tool 装饰器
from langchain.tools import tool


# 注册工具
@tool
# 注意 query 没有类型注解
def no_annotation(query) -> str:
    # 描述照常从 docstring 取
    """没有类型注解的工具。"""
    # 返回固定结果
    return "ok"


# 观察生成的 schema
print(no_annotation.args)

输出：

{'query': {'title': 'Query'}}

type 字段整个消失了。模型只知道有个叫 query 的参数，却不知道该填字符串、数字还是数组。这不会抛异常，只会让填参质量悄悄变差——典型的静默失败。所以「类型注解必写」不是风格建议，而是功能要求。

完全不写 docstring：直接报错。

# 导入 tool 装饰器
from langchain.tools import tool

# 用 try 捕获构造期的异常
try:
    # 注册一个既没有 docstring 也没有 description 的工具
    @tool
    # 函数签名正常，唯一的问题是没有说明
    def no_doc(query: str) -> str:
        # 函数体里没有 docstring，直接就是返回语句
        return "ok"

# 捕获并打印错误
except ValueError as e:
    # 打印异常类型与信息
    print(f"{type(e).__name__}: {e}")

输出：

ValueError: Function must have a docstring if description not provided.

这个失败同样「响亮」：装饰器执行时就炸——比静默降级友好得多。LangChain 强制你至少给模型留一句说明：要么写 docstring，要么在 @tool(description=...) 里显式传入。

4. 自定义名称与描述 #

默认用函数名 / docstring 往往够用。需要覆盖时常见两种动机：

  1. 函数名偏实现（如 search），对外希望更语义化（web_search）
  2. docstring 偏实现说明，但模型需要更强的选型指令（「算术必须调用，禁止心算」）

可在 @tool(...) 上显式传入名称与 description。好处是：对外「说明书」和内部实现命名解耦了。函数叫什么随团队习惯，模型看到的名字可以单独优化。

4.1. 2.custom_name_desc.py # 

# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool


# @tool 的第一个位置参数就是工具名，这里覆盖为 web_search
@tool("web_search")
# 函数本身仍可叫 search，两者互不影响
def search(query: str) -> str:
    # 没有传 description 时，描述仍取自 docstring
    """在互联网上搜索信息。"""
    # 返回模拟搜索结果
    return f"关于「{query}」的搜索结果。"


# 同时覆盖名称和描述
@tool(
    # 第一个位置参数：对外工具名
    "calculator",
    # description 会覆盖 docstring，这里写成带指令性的选型说明
    description="执行四则运算。遇到任何算术问题都请调用本工具，不要心算。",
)
# 内部函数名仍可用简写
def calc(expression: str) -> str:
    # 传了 description 后，这句 docstring 对模型不再可见
    """计算数学表达式。"""
    # 白名单：只允许数字、四则运算符、括号、小数点和空格
    allowed = set("0123456789+-*/(). ")
    # set(expression) <= allowed 判断表达式的字符是否都在白名单内
    if not set(expression) <= allowed:
        # 不合法时返回错误说明，让模型知道该改写表达式
        return "错误：表达式包含不允许的字符。"
    # 用空的 __builtins__ 执行，屏蔽内置函数，降低（但不能消除）风险
    return str(eval(expression, {"__builtins__": {}}, {}))


# 打印两个工具对外暴露的名字
print(search.name, calc.name)
# 打印 calc 的描述，确认是 description 而不是 docstring
print(calc.description)

输出：

web_search calculator
执行四则运算。遇到任何算术问题都请调用本工具，不要心算。

description 是写给模型的「使用说明书」，比写给人类的代码注释更重要。多工具时，描述要能互斥：

4.2. 工具名不能随便起：^[a-zA-Z0-9_-]+$ #

前面说「工具名建议 snake_case」，但没说清违反了会怎样。实测结论比「建议」严厉得多。

LangChain 完全不校验工具名，什么都能构造成功：

# 导入 tool 装饰器
from langchain.tools import tool


# 工具名里带空格
@tool("web search")
# 函数本身写法完全正常
def f1(query: str) -> str:
    # 描述取自 docstring
    """搜索信息。"""
    # 返回占位结果
    return "ok"


# 工具名用中文
@tool("查询天气")
# 函数本身写法也完全正常
def f2(city: str) -> str:
    # 描述取自 docstring
    """查询城市天气。"""
    # 返回占位结果
    return "ok"


# 两个都能顺利构造出来，name 原样保留
print(repr(f1.name), repr(f2.name))

输出：

'web search' '查询天气'

但一旦绑到模型上发请求，API 会拒绝：

# 导入 dotenv 用于加载 API key
from dotenv import load_dotenv
# 导入统一模型初始化函数
from langchain.chat_models import init_chat_model
# 导入 tool 装饰器
from langchain.tools import tool

# 加载 .env 中的密钥
load_dotenv(override=True)


# 工具名带空格
@tool("web search")
# 入参为检索词
def bad_name(query: str) -> str:
    # 描述取自 docstring
    """搜索信息。"""
    # 返回模拟结果
    return f"关于 {query} 的结果"


# 初始化模型
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)

# 绑定并调用，捕获 API 返回的错误
try:
    # bind_tools 本身不报错，真正发请求时才失败
    model.bind_tools([bad_name]).invoke("搜索一下 LangChain")
# 打印错误类型与信息
except Exception as e:
    # 输出 BadRequestError 及其详细说明
    print(f"{type(e).__name__}: {e}")

输出：

BadRequestError: Error code: 400 - {'error': {'message': "Invalid 'tools[0].function.name': string does not match pattern. Expected a string that matches the pattern '^[a-zA-Z0-9_-]+$'.", ...}}

规则很明确：只允许字母、数字、下划线、连字符。所以：

工具名 能构造 API 接受
get_weather √ √
get-weather √ √
web search（空格） √ × 400
web.search（点） √ × 400
查询天气（中文） √ × 400

中文工具名会被拒，这一条要特别记住——做中文业务时很容易顺手把工具名写成中文，而错误要等到真正发请求才出现；报错信息还是英文、指向 tools[0].function.name，第一次遇到不容易想到是工具名的问题。

结论：工具名一律用英文 snake_case；中文只放在 description 里。描述用中文完全没问题（前面例子里的中文描述都正常工作）。

4.3. 关于示例里的 eval #

上面 calc 的白名单看起来严密，其实挡不住所有麻烦：

# 白名单字符集合
allowed = set("0123456789+-*/(). ")

# 逐个测试几种表达式能否通过白名单
for expr in ["1+1", "9**9", "9**9**9", "(1).__class__"]:
    # set(expr) <= allowed 判断字符是否全在白名单里
    print(f"{expr!r:18s} 通过白名单: {set(expr) <= allowed}")

输出：

'1+1'              通过白名单: True
'9**9'             通过白名单: True
'9**9**9'          通过白名单: True
'(1).__class__'    通过白名单: False

访问属性那种注入被挡住了，但 ** 是两个 * 拼起来的，白名单管不了——9**9**9 是个天文数字，会让进程算到内存耗尽。这类问题叫资源耗尽型攻击，光靠字符白名单防不住。

所以本章的 calc 只用来演示「怎么覆盖名称和描述」，不要照搬到生产。真正要给模型算术能力，用 §10 那种把运算符做成 Literal["+", "-", "*", "/"] 枚举的写法：模型只能在四个符号里选一个，** 根本表达不出来，从结构上消掉问题。

更普适的原则：能用枚举 + 结构化参数表达的能力，就不要让模型自由拼字符串。字符串是最难校验的输入形式。第 10 章讲 middleware 与 HITL 时会继续这个话题。

5. 参数 Schema #

模型填参质量，很大程度上取决于 schema 是否清晰。
可以把 schema 想成「给模型填的表单」：字段名、类型、说明、可选/必填、枚举值，都会影响它怎么填。

由简到繁三档：

档位 做法 适用 能否约束取值
基础 类型注解 + parse_docstring=True 参数少、类型简单 只能靠描述「建议」
进阶 Pydantic args_schema + Field(description=...) 枚举、默认值、约束、多字段 能，且本地会校验
兼容 JSON Schema 字典 动态生成 / 与外部 schema 对齐 取决于你写的 schema

业务工具建议尽早升到「进阶」：多花几行 Field，能少掉大量「参数抽成整句问话」的翻车。最后一列是关键区别——基础档只能在描述里写「天数 1～7」这种软约束，模型想填 100 也拦不住；进阶档用 Literal 或 Field(ge=1, le=7) 能真正锁死。

5.1. 类型注解够用时 #

两个参数、类型直观时，注解 + parse_docstring=True 就够：

# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool


# 加上 parse_docstring=True，Args 段才会落到各参数的 description 上（见 §3.3）
@tool(parse_docstring=True)
# city 无默认值所以必填；days 有默认值所以可选
def get_forecast(city: str, days: int = 3) -> str:
    # 首行摘要成为工具描述，Args 段成为参数说明
    """查询城市未来几天的天气预报。

    Args:
        city: 城市名，如「北京」
        days: 预报天数，1～7
    """
    # 返回模拟预报
    return f"{city} 未来 {days} 天：晴转多云。"


# 打印工具名
print("name:", get_forecast.name)
# 打印描述（只有摘要行）
print("description:", get_forecast.description)
# 打印参数结构（带上了 description）
print("args:", get_forecast.args)

输出：

name: get_forecast
description: 查询城市未来几天的天气预报。
args: {'city': {'description': '城市名，如「北京」', 'title': 'City', 'type': 'string'}, 'days': {'default': 3, 'description': '预报天数，1～7', 'title': 'Days', 'type': 'integer'}}

注意 days: int = 3 这个默认值做了两件事：本地调用时可以省略；同时在 schema 里表现为「不在 required 里」，等于告诉模型「这个可以不填」。

不过这里也暴露了类型注解方案的天花板：docstring 里写「预报天数，1～7」只是一句给模型看的建议，模型完全可以填 days=100，int 类型拦不住。想真正约束取值范围，得升级到 Pydantic。

5.2. 3.pydantic_schema.py（推荐业务工具） #

字段一多，或需要 Literal / 取值约束时，用 Pydantic 显式声明 args_schema。
这和第 6 章「用 schema 约束模型输出」是同一思路，只是这里约束的是工具入参。

# 从 typing 导入 Literal，用于把取值限制在固定几个选项内
from typing import Literal

# 从 pydantic 导入 BaseModel 与 Field
from pydantic import BaseModel, Field

# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool


# 继承 BaseModel 定义入参模型
class WeatherInput(BaseModel):
    # 类的 docstring 会成为 schema 的顶层 description
    """天气查询入参。"""

    # 必填字段：没有 default，所以会进入 required 列表
    location: str = Field(description="城市名或坐标，如「上海」")
    # Literal 会在 JSON Schema 里变成 enum，模型只能二选一
    units: Literal["C", "F"] = Field(
        # 有默认值 → 可选字段
        default="C",
        # 描述帮助模型理解该填什么
        description="温度单位：C 或 F",
    )
    # 布尔开关字段
    include_forecast: bool = Field(
        # 默认关闭
        default=False,
        # 说明这个开关的作用
        description="是否包含未来 5 日预报",
    )


# args_schema 指定入参模型，此时函数自己的类型注解不再用于生成 schema
@tool(args_schema=WeatherInput)
# 函数参数名必须和 schema 字段名逐一对应（否则调用时才报错，见 §5.4）
def get_weather(
    # 对应 schema 的 location
    location: str,
    # 对应 schema 的 units
    units: str = "C",
    # 对应 schema 的 include_forecast
    include_forecast: bool = False,
) -> str:
    # 这段 docstring 成为工具描述（说明何时该调用）
    """查询当前天气；需要事实天气时必须调用，不要编造。"""
    # 根据单位给出模拟温度
    temp = 22 if units == "C" else 72
    # 拼装基础返回文本
    result = f"{location} 当前 {temp}°（{units}），晴。"
    # 只有模型明确要求预报时才追加这一段
    if include_forecast:
        result += " 未来 5 日：多云。"
    # 返回字符串，它将成为 ToolMessage 的内容
    return result


# 查看最终的参数结构
print(get_weather.args)
# 本地测一条，确认业务逻辑正确
print(get_weather.invoke({"location": "上海", "units": "C"}))
# 确认描述来自函数 docstring，而不是 WeatherInput 的 docstring
print(repr(get_weather.description))

输出（args 已格式化便于阅读）：

{'location': {'description': '城市名或坐标，如「上海」',
              'title': 'Location',
              'type': 'string'},
 'units': {'default': 'C',
           'description': '温度单位：C 或 F',
           'enum': ['C', 'F'],
           'title': 'Units',
           'type': 'string'},
 'include_forecast': {'default': False,
                      'description': '是否包含未来 5 日预报',
                      'title': 'Include Forecast',
                      'type': 'boolean'}}
上海 当前 22°（C），晴。
'查询当前天气；需要事实天气时必须调用，不要编造。'

对照 §3.2 那份「只有 type、没有 description」的 schema，差距很直观：每个字段都有说明，units 还多了 enum 把取值锁死。模型填参质量的上限，由这份 schema 决定。

顺便说清一个容易混的点：用了 args_schema 之后，工具描述仍然取自函数的 docstring，WeatherInput 的 docstring 只是 schema 内部的顶层描述。所以「这个工具什么时候该调用」写在函数上，「每个字段是什么意思」写在 Field 里。

设计原则（与第 6 章结构化输出一致）：

5.3. 参数校验是真的在跑 #

args_schema 不只是给模型看的说明书，同时也是本地的运行时校验。填错了会在工具执行前就被 Pydantic 拦下来：

# 承接 §5.2 的 get_weather 工具

# 情况一：units 填了枚举之外的值
try:
    # K 不在 Literal 的两个选项里
    get_weather.invoke({"location": "上海", "units": "K"})
# Pydantic 会抛 ValidationError
except Exception as e:
    # 打印异常类型与详细信息
    print(f"{type(e).__name__}: {e}")

# 情况二：漏掉必填的 location
try:
    # 只传了 units，没传必填的 location
    get_weather.invoke({"units": "C"})
# 同样是 ValidationError，但原因不同
except Exception as e:
    # 打印异常类型与详细信息
    print(f"{type(e).__name__}: {e}")

输出：

ValidationError: 1 validation error for WeatherInput
units
  Input should be 'C' or 'F' [type=literal_error, input_value='K', input_type=str]
    For further information visit https://errors.pydantic.dev/2.13/v/literal_error
ValidationError: 1 validation error for WeatherInput
location
  Field required [type=missing, input_value={'units': 'C'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing

这层校验的意义在于：模型偶尔还是会填错，而错误会在进入业务代码之前就被挡住，不会带着 units="K" 跑进函数体产生诡异结果。代价是会抛异常——所以 §9 的错误兜底才有必要，否则一次填错就能打崩整个 Agent。

5.4. 坑：args_schema 和函数签名不一致时，构造不报错 #

这是典型的「延迟失败」：schema 里叫 location，函数参数叫 city，装饰器不会有任何抱怨。

# 导入 Pydantic 组件
from pydantic import BaseModel, Field
# 导入 tool 装饰器
from langchain.tools import tool


# schema 里字段名叫 location
class MismatchInput(BaseModel):
    # 类 docstring 成为 schema 顶层描述
    """入参。"""

    # 唯一的字段名是 location
    location: str = Field(description="城市名")


# 挂上 schema
@tool(args_schema=MismatchInput)
# 但函数参数名叫 city，和 schema 不一致
def weather_bad(city: str) -> str:
    # 描述取自 docstring
    """查天气。"""
    # 函数体正常使用 city
    return f"{city} 晴"


# 构造阶段完全正常，schema 用的是 location
print("构造成功，args:", weather_bad.args)

# 按 schema 的字段名调用，问题才暴露
try:
    # 校验能通过，但传给函数时参数名对不上
    weather_bad.invoke({"location": "上海"})
# 捕获 TypeError
except TypeError as e:
    # 打印异常类型与信息
    print(f"{type(e).__name__}: {e}")

输出：

构造成功，args: {'location': {'description': '城市名', 'title': 'Location', 'type': 'string'}}
TypeError: weather_bad() got an unexpected keyword argument 'location'

机制是：schema 负责校验并生成模型可见的定义；通过之后，LangChain 把校验后的字段按关键字参数传给函数。字段名对不上，函数自然收不到。

麻烦在于时机——如果这个工具只挂在 Agent 上、又只在某个少见分支才被调用，你可能到线上才第一次看到这个 TypeError。防御办法很简单：写完带 args_schema 的工具，立刻 tool.invoke({...}) 跑一次。这也是 §3.1 强调「先本地单测」的原因之一。

5.5. 两个不给模型看的特殊参数 #

有些参数是给你的代码用的，不该出现在模型的表单里——比如当前用户 ID、数据库连接、租户标识。让模型填这类参数不仅没意义，还有安全风险（模型可以填别人的 user_id）。

config: RunnableConfig 会被自动排除在模型可见 schema 之外：

# 导入 RunnableConfig 类型
from langchain_core.runnables import RunnableConfig
# 导入 tool 装饰器
from langchain.tools import tool


# 注册工具
@tool
# config 这个参数名有特殊含义，会被 LangChain 识别并注入
def with_config_param(query: str, config: RunnableConfig) -> str:
    # 描述取自 docstring
    """搜索信息。"""
    # 从 config 里读取调用方传进来的元数据
    return f"query={query}, tags={config.get('tags')}"


# 模型可见的参数里没有 config
print("模型可见的 args:", with_config_param.args)
# 调用时 config 通过第二个参数传入，而不是放在字典里
print("invoke:", with_config_param.invoke({"query": "x"}, config={"tags": ["t1"]}))

输出：

模型可见的 args: {'query': {'title': 'Query', 'type': 'string'}}
invoke: query=x, tags=['t1']

InjectedToolArg 用于你自己定义的隐藏参数：

# 导入 Annotated 用于给类型附加元信息
from typing import Annotated

# InjectedToolArg 是一个标记，表示该参数由代码注入而非模型填写
from langchain_core.tools import InjectedToolArg
# 导入 tool 装饰器
from langchain.tools import tool


# 注册工具
@tool
# user_id 用 Annotated 标注成注入参数
def with_injected(query: str, user_id: Annotated[str, InjectedToolArg]) -> str:
    # 描述取自 docstring
    """搜索信息。"""
    # 函数体内正常使用这个参数
    return f"query={query}, user_id={user_id}"


# 模型只看得到 query
print("模型可见:", list(with_injected.args.keys()))
# 但完整 schema 里两个字段都在（校验时仍需要它）
print("完整 schema:", list(with_injected.args_schema.model_json_schema()["properties"].keys()))
# 调用时必须由你显式提供 user_id
print("invoke:", with_injected.invoke({"query": "x", "user_id": "u001"}))

输出：

模型可见: ['query']
完整 schema: ['query', 'user_id']
invoke: query=x, user_id=u001

两者的分工：

参数 谁提供 典型用途
普通参数 模型填 城市名、订单号、检索词
config: RunnableConfig 调用方通过 config= 传 tags、metadata、thread_id 等运行时上下文
Annotated[T, InjectedToolArg] 你的代码在执行前注入 当前用户 ID、租户、权限令牌

安全意义值得强调：凡是「不该由用户输入决定」的参数，都该用注入，而不是让模型填。第 15 章讲 RAG 权限过滤时用的就是这个思路——把当前用户的部门 dept 由代码注入进检索工具，而不是指望模型老老实实填自己的身份。

6. bind_tools：让模型发起调用 #

bind_tools 把工具 schema 绑定到聊天模型：模型在回复里可以带上 tool_calls。
它不会自动执行工具——执行与回灌要你自己做，或交给 create_agent。

这是初学者最常见的误解：

× bind_tools = 模型会自己去查天气
√ bind_tools = 模型被允许「提出」查天气；谁执行另说
bind_tools 的职责：告诉模型「你有这些工具可用」
create_agent 的职责：绑定 + 执行 + 回灌 + 循环直到最终回答

手写循环的价值：把 Agent 黑盒拆成三步，调试「没调工具 / 参数错 / id 对不齐」时特别清楚。业务默认仍用 Agent。

6.1. 4.bind_tools_loop.py # 

# 从 langchain.chat_models 导入统一初始化函数
from langchain.chat_models import init_chat_model
# HumanMessage 表示用户消息，ToolMessage 表示工具执行结果
from langchain.messages import HumanMessage, ToolMessage
# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key；override=True 表示覆盖已存在的同名环境变量
load_dotenv(override=True)


# 注册天气工具
@tool
# 只有一个入参 city
def get_weather(city: str) -> str:
    # 描述告诉模型这个工具能干什么
    """获取指定城市的天气信息。"""
    # 模拟返回（真实项目里换成调用天气 API）
    return f"{city} 今天晴，气温 25°C。"


# 初始化模型；temperature=0 让选型行为更稳定可复现
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
# bind_tools 返回一个新对象，原 model 不受影响
model_with_tools = model.bind_tools([get_weather])

# 建立「工具名 → 工具对象」索引：模型只会返回名字，得靠它找到真正的函数
toolkit = {get_weather.name: get_weather}

# 手工维护消息列表，初始只有用户这一句
messages = [HumanMessage("北京今天天气怎么样？")]

# 第 1 步：模型推理，返回的 AIMessage 可能带 tool_calls
ai = model_with_tools.invoke(messages)
# 把这条 AIMessage 追加进轨迹（这一步不能省，否则下一轮模型看不到自己的决定）
messages.append(ai)
# 打印模型提出的调用意图
print("tool_calls:", ai.tool_calls)

# 第 2 步：遍历每一个调用意图（可能不止一个）
for call in ai.tool_calls:
    # 按名字取出对应的工具对象
    selected = toolkit[call["name"]]
    # 用模型给的参数字典执行工具，拿到观察结果
    observation = selected.invoke(call["args"])
    # 把结果包成 ToolMessage 追加进轨迹
    messages.append(
        ToolMessage(
            # 工具返回的内容
            content=observation,
            # 必须回填本次调用的 id，模型靠它把结果和请求配对
            tool_call_id=call["id"],
        )
    )

# 第 3 步：带着工具结果再问一次模型，让它组织自然语言回答
final = model_with_tools.invoke(messages)
# 这一次 content 不再为空，tool_calls 也应为空
print(final.content)

输出（最后一句是模型生成的自然语言，每次跑措辞会不同）：

tool_calls: [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_00_U7e6OQNcApDU457HaAPL5889', 'type': 'tool_call'}]
北京今天天气晴朗，气温 25°C，天气不错，适合外出活动。

两个细节值得细看：

三步对照：

步 你在干什么 轨迹上多了什么
1 invoke 模型 AIMessage（常带 tool_calls）
2 tool.invoke + 追加 ToolMessage
3 再 invoke 模型 最终 AIMessage.content

对照记忆：

API 作用
bind_tools 模型侧：允许发出 tool_calls
tool.invoke(args) 本地执行工具
ToolMessage 把结果写回 messages（tool_call_id 必对齐）
create_agent 上面整段循环的自动化

排错对照表：

症状 大概率原因
ai.tool_calls 是空列表 描述没写清「何时用」；或系统提示没引导；或模型不支持 tool calling
有 tool_calls 但最终回答胡编 忘了把 ToolMessage 追加进 messages
API 报 400，提示 tool message 缺失 tool_call_id 没对齐，或有多个 tool_calls 只回了一条
报 KeyError 在 toolkit[call["name"]] 模型报的工具名不在你的索引里（通常是绑定的工具和索引不一致）

最后一条尤其要注意：ai.tool_calls 里有几个调用，就必须回几条 ToolMessage，一条都不能少。第 6 章讲结构化输出重试时踩过同一个坑——AIMessage 带了 tool_calls，后面却没有对应的 ToolMessage，API 就会拒绝整个请求。

7. 挂到 create_agent #

定义好工具后，挂到 Agent 只需放进 tools=[...]。驾驭层（create_agent）会完成执行与回灌——也就是 §6 那三步的自动化版。

和 bind_tools 的选用：

场景 更合适
业务助理、多步工具、要少写样板代码 create_agent
只想看模型会不会发出 tool_calls、或自定义循环 bind_tools
既要 Agent 又要细调模型参数 init_chat_model 得到实例再传给 create_agent（第 3 章）

7.1. 5.agent_with_tools.py # 

# 从 langchain.agents 导入 create_agent 工厂函数
from langchain.agents import create_agent
# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)


# 注册天气工具
@tool
# 单一入参 city
def get_weather(city: str) -> str:
    # 描述里同时写清「能力」和「何时用」
    """获取指定城市的天气信息。需要天气事实时必须调用。"""
    # 模拟返回
    return f"{city} 今天晴，气温 25°C。"


# 创建 Agent，内部会自动完成 §6 那三步循环
agent = create_agent(
    # 模型标识字符串，格式为「厂商:模型名」
    model="deepseek:deepseek-v4-flash",
    # 挂载工具列表，驾驭层（create_agent）负责执行与回灌
    tools=[get_weather],
    # 系统提示：约束角色与策略，禁止编造
    system_prompt="你是中文助手。查天气必须调用 get_weather，不要编造。",
)

# 一次 invoke 内部可能包含多轮「模型→工具→模型」
result = agent.invoke(
    # 输入格式是带 messages 键的字典
    {"messages": [{"role": "user", "content": "深圳天气如何？"}]}
)
# messages 列表的最后一条就是最终回答
print(result["messages"][-1].content)

输出：

深圳今天天气晴朗，气温为 25°C。天气不错，适合外出活动哦！

和 §6 对比一下代码量：手写循环 40 多行，Agent 版十几行，而且自动处理了多轮、多工具、id 对齐。日常业务没有理由手写循环——手写的价值只在于理解机制和调试。

system_prompt 与工具 docstring 要同向，分工是：

写在哪 负责什么 例子
system_prompt 策略层：角色、必须/禁止、语气 「查天气必须调用 get_weather，不要编造」
工具 description 能力层：这个工具干什么 「获取指定城市的天气信息」
参数 description 字段层：每个参数填什么 「城市名，如北京、上海」

三层任意一层缺失，选型和填参都更容易漂。反过来，遇到「模型该调工具却不调」时，按这三层逐一检查通常能定位。

调试时建议打印完整 messages（第 2 / 4 章的方法），不要只看最后一句——中间的 tool_calls 和 ToolMessage 才能告诉你模型到底选了什么、填了什么参数。

8. 工具返回值怎么选 #

工具返回值会变成模型下一轮看到的「观察」。返回形态不同，后续行为也不同。

返回类型 模型看到什么 适用
str 一段可读文本 大多数业务工具（默认首选）
dict / 对象 序列化后的结构化字段 下游还要按字段推理
内容块列表 多模态 ToolMessage 截图、图片等（需模型支持）
return_direct=True 不再经模型润色，直接把工具输出当最终结果 订单状态等「查完即返」

入门阶段优先返回简洁字符串：字段名清晰，别塞整页 JSON 噪音。需要结构化给下游程序用时，再考虑 dict 或第 6 章的结构化输出。

顺便提一个小坑。如果参数声明成 a: float, b: float，那么「128 + 256」的结果会是 384.0 而不是 384：

ToolMessage: '384.0'

模型通常能在最终回答里正确写成「384」，但观察值里那个 .0 会一路带到日志和轨迹里。整数场景要么用 int，要么在返回前格式化一下。返回值是给模型读的文本，值得像写给用户的文案一样收拾干净。

8.1. return_direct=True：查完即返 #

默认流程是：工具结果 → 再进模型 → 自然语言最终答。
打开 return_direct=True 后，Agent 拿到该工具结果就立刻结束，不再让模型润色。

适合：订单状态、验证码、固定格式回执——不需要模型再总结。
不适合：多工具拼接、需要对比/改写语气、还要继续决策的场景。

# 从 langchain.agents 导入 create_agent，用于创建智能体
from langchain.agents import create_agent
# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool
# 从 dotenv 导入 load_dotenv，用于加载环境变量
from dotenv import load_dotenv

# 加载 .env 中的密钥；override=True 表示覆盖已存在的同名变量
load_dotenv(override=True)


# return_direct=True：Agent 拿到该工具结果后直接结束，不再经模型二次润色
@tool(return_direct=True)
# 入参为订单号
def fetch_order_status(order_id: str) -> str:
    # 描述里写清何时调用
    """查询订单物流状态；用户问订单进度时调用。"""
    # 返回固定格式的回执
    return f"订单 {order_id} 已发货，预计 2 天后送达。"


# 创建 Agent，挂载查单工具
agent = create_agent(
    # 本文默认 DeepSeek 模型
    model="deepseek:deepseek-v4-flash",
    # 注册工具
    tools=[fetch_order_status],
    # 系统提示：引导模型在查订单时调用工具
    system_prompt="你是订单助手。用户询问订单进度时必须调用 fetch_order_status。",
)

# 调用 Agent，询问订单状态
result = agent.invoke(
    {
        # 输入是带 messages 键的字典
        "messages": [
            {
                # 角色为 user
                "role": "user",
                # 用户的问题
                "content": "帮我查一下订单 A1001 的物流进度。",
            }
        ]
    }
)

# 遍历完整消息轨迹，用 enumerate 拿到序号便于对照
for i, msg in enumerate(result["messages"]):
    # 打印分隔线与消息类型
    print(f"\n===== [{i}] {type(msg).__name__} =====")
    # 用 getattr 安全取 content，不同消息类型都有这个属性
    content = getattr(msg, "content", None)
    # 只有非空时才打印，避免刷屏
    if content:
        print("content:", content)
    # tool_calls 只有 AIMessage 才有，用 getattr 兼容其他类型
    tool_calls = getattr(msg, "tool_calls", None)
    # 有调用意图时打印出来
    if tool_calls:
        print("tool_calls:", tool_calls)

# 打印最终对外结果，注意此时它来自 ToolMessage
print("\n最终回复:", result["messages"][-1].content)

输出（省略了分隔线）：

[0] HumanMessage  content: 帮我查一下订单 A1001 的物流进度。
[1] AIMessage     content: 我来帮您查询订单 A1001 的物流进度。
                  tool_calls: [{'name': 'fetch_order_status', 'args': {'order_id': 'A1001'}, ...}]
[2] ToolMessage   content: 订单 A1001 已发货，预计 2 天后送达。

最终回复: 订单 A1001 已发货，预计 2 天后送达。

轨迹只有 3 条，最后一条是 ToolMessage。 这是本节最该记住的事实。

把同一个工具去掉 return_direct=True 再跑一次，对照非常清楚：

[0] HumanMessage  content: 帮我查一下订单 A1001 的物流进度。
[1] AIMessage     content: ''                              ← 空，只提交了调用
[2] ToolMessage   content: 订单 A1001 已发货，预计 2 天后送达。
[3] AIMessage     content: 订单 A1001 已发货，预计 2 天后送达。请问还有其他需要帮助的吗？

汇总成表：

return_direct=True 默认
消息条数 3 4
最后一条的类型 ToolMessage AIMessage
最终内容 工具返回的原文 模型改写过的版本
模型调用次数 1 2

有三个实际影响：

  1. 省一次模型调用，延迟和成本都更低。查单这类高频场景很可观。
  2. 输出完全可控，工具返回什么，用户就看到什么，不会被模型加上「请问还有其他需要帮助的吗」这类尾巴。
  3. result["messages"][-1] 的类型变了。如果你的代码里写了 isinstance(last, AIMessage) 之类的判断，或依赖 last.tool_calls 属性，在 return_direct 下会走到意外分支。取 .content 是安全的，两种消息都有。

另外注意上面轨迹里 [1] 的 AIMessage 是有内容的（「我来帮您查询…」），而默认版本那条是空的。这只是模型的随机行为，不由 return_direct 决定——它可能在提交调用的同时说句话，也可能什么都不说。不要依赖这条消息的 content 有值。

9. 错误处理 #

工具失败时，理想行为是：把可理解的错误变成 ToolMessage，让模型改参或换策略，而不是整个进程堆栈炸掉。

为什么「抛异常就结束」不好？因为模型常见错误是参数抽歪（订单号多了空格、城市名带了「市」）。错误能回灌，它往往能自纠；进程直接挂，用户只看到 500。

常见三层：

层 做法 何时用
工具内防御 try/except 后 return "错误：..." 可预期的业务错误（单号不存在等）
Agent 中间件 @wrap_tool_call 捕获异常并转 ToolMessage 统一兜底、避免未捕获异常打崩循环
可观测 LangSmith 看失败轨迹 定位偶发 / 参数问题

9.1. 先看不处理会怎样：整个 Agent 崩掉 #

抽象地说「进程会崩」没有说服力，直接跑一次。下面这个工具用 ORDERS[key] 直接取值，查不到就会抛 KeyError——这是最常见的疏忽写法：

# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv
# 从 langchain.agents 导入 create_agent
from langchain.agents import create_agent
# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool

# 加载 .env 中的 API key
load_dotenv(override=True)

# 模拟订单库，只有两条数据
ORDERS = {"A1001": "已发货", "A1002": "运输中"}


# 注册查单工具
@tool
# 入参为订单号
def lookup_order(order_id: str) -> str:
    # 描述取自 docstring
    """按订单号查询物流状态。订单号形如 A1001。"""
    # 规范化：去空格并转大写
    key = order_id.strip().upper()
    # 直接用 [] 取值，查不到会抛 KeyError —— 这里故意不做防御
    return f"订单 {key} 状态：{ORDERS[key]}。"


# 创建 Agent，不加任何错误处理
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 挂载会抛异常的工具
    tools=[lookup_order],
    # 引导模型调用工具
    system_prompt="你是订单助手。查订单必须调用 lookup_order。",
)

# 查一个不存在的订单号，用 try 观察异常是否会传到最外层
try:
    # A9999 不在 ORDERS 里
    result = agent.invoke({"messages": [{"role": "user", "content": "帮我查订单 A9999"}]})
    # 如果没抛异常，打印最终回答
    print(result["messages"][-1].content)
# 捕获任何异常并打印类型与信息
except Exception as e:
    # 实测这里会打印 KeyError: 'A9999'
    print(f"整个 Agent 崩了！{type(e).__name__}: {e}")

输出：

整个 Agent 崩了！KeyError: 'A9999'

注意这里发生了什么：异常穿透了整个 Agent，从 agent.invoke() 直接抛到你的代码里。 没有 ToolMessage，没有降级回答，模型完全没有机会向用户解释。如果这是 Web 服务，用户看到的就是 500。

这就是本节存在的理由。下面两节是两种修法。

顺便纠正一个容易产生的误解：这个演示之所以用「订单查不到」而不是「10 除以 0」，是因为除零这类错误模型自己就会规避——你问它「10 除以 0 等于多少」，它会推理出除数不能为 0，然后干脆不调用工具，直接给你解释。工具里的 raise 一次都不会执行，也就演示不出兜底效果。想验证错误处理，得挑一个模型无法预判的失败：订单号不存在、网络超时、第三方返回 500。

9.2. 修法一：工具内返回错误字符串 #

这是第一道防线，也是优先选择：业务上「查不到」根本不是系统崩溃，而是一条正常的观察结果。

# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool

# 模拟订单库
ORDERS = {"A1001": "已发货", "A1002": "运输中"}


# 注册工具
@tool
# 入参为订单号
def lookup_order(order_id: str) -> str:
    # 描述里说明订单号格式，帮助模型正确填参
    """按订单号查询状态。order_id 形如 A1001。"""
    # 规范化输入：去掉首尾空格并转成大写，容忍模型的格式偏差
    key = order_id.strip().upper()
    # 用 not in 判断而不是直接取值，避免 KeyError
    if key not in ORDERS:
        # 返回错误说明字符串，而不是 raise
        return f"错误：未找到订单 {order_id}。可用示例：A1001、A1002。"
    # 查到则返回正常状态
    return f"订单 {key} 状态：{ORDERS[key]}。"

把它挂到 Agent 上，查 A9999，轨迹变成：

[0] HumanMessage  '帮我查订单 A9999'
[1] AIMessage     ''
[2] ToolMessage   '错误：未找到订单 A9999。可用示例：A1001、A1002。'
[3] AIMessage     '抱歉，未查询到订单 A9999 的物流信息。系统中没有找到该订单。
                   可用示例订单号有：A1001、A1002。请问您是否需要查询这两个订单中的某一个呢？'

进程活着，模型看到了错误，还把「可用示例」转达给了用户。对比 §9.1 的 KeyError，同样的用户输入，体验天差地别。

那句 key = order_id.strip().upper() 也不是摆设。实测让模型查「订单 a1001（小写）」，它老实地填了 {'order_id': 'a1001'}，靠工具内的规范化才查到：

[1] AIMessage    tool_calls args=[{'order_id': 'a1001'}]
[2] ToolMessage  '订单 A1001 状态：已发货。'
[3] AIMessage    '查询结果：订单 A1001 状态为 已发货。'

模型不会帮你规范化输入，这活儿得工具自己干。 大小写、首尾空格、中文「市」字后缀、全角字符，都属于工具该容忍的范围。

错误文案的写法建议：

9.3. 修法二：6.wrap_tool_errors.py（Agent 统一兜底） #

工具内防御能覆盖你想到了的错误。想不到的呢——第三方超时、空指针、依赖库内部异常？用 middleware 兜一层。

官方推荐用 wrap_tool_call 把未预期异常转成模型可读的 ToolMessage。第 10 章会系统讲 middleware；这里先掌握「给工具调用外包一层 try/except」这一招。

# 从 collections.abc 导入 Callable 用于类型标注
from collections.abc import Callable

# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv
# 从 langchain.agents 导入 create_agent
from langchain.agents import create_agent
# wrap_tool_call 是把函数注册成「工具调用拦截器」的装饰器
from langchain.agents.middleware import wrap_tool_call
# ToolMessage 用于构造工具结果消息
from langchain.messages import ToolMessage
# ToolCallRequest 描述一次待执行的工具调用
from langchain.tools.tool_node import ToolCallRequest
# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool

# 加载 .env 中的 API key
load_dotenv(override=True)

# 模拟订单库
ORDERS = {"A1001": "已发货", "A1002": "运输中"}


# 注册工具，这里故意保留会抛 KeyError 的写法
@tool
# 入参为订单号
def lookup_order(order_id: str) -> str:
    # 描述取自 docstring
    """按订单号查询物流状态。订单号形如 A1001。"""
    # 规范化输入
    key = order_id.strip().upper()
    # 直接取值，查不到就抛 KeyError，用来验证中间件能否兜住
    return f"订单 {key} 状态：{ORDERS[key]}。"


# 用 wrap_tool_call 装饰后，每次工具调用都会先经过这个函数
@wrap_tool_call
# 函数名随意，签名必须是 (request, handler)
def handle_tool_errors(
    # 本次调用请求，含工具名、参数、调用 id
    request: ToolCallRequest,
    # handler 是「真正执行这次调用」的默认逻辑
    handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
    # 拦截器自己的说明
    """把工具异常转换成模型可消费的 ToolMessage。"""
    # 正常路径：调用 handler 执行原工具
    try:
        # 成功时原样返回 handler 产出的 ToolMessage
        return handler(request)
    # 失败路径：捕获所有异常，不让它继续往上抛
    except Exception as e:
        # 构造一条错误 ToolMessage 顶替原本的结果
        return ToolMessage(
            # 把异常类型和信息都告诉模型，并提示下一步动作
            content=f"工具错误：{type(e).__name__} - {e}。请检查参数后重试。",
            # 必须回填调用 id，否则消息无法与请求配对
            tool_call_id=request.tool_call["id"],
        )


# 创建 Agent，通过 middleware 参数挂上兜底逻辑
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 仍然是那个会抛异常的工具
    tools=[lookup_order],
    # middleware 接收列表，可以叠多个
    middleware=[handle_tool_errors],
    # 系统提示
    system_prompt="你是订单助手。查订单必须调用 lookup_order。",
)

# 同样查不存在的订单号
result = agent.invoke({"messages": [{"role": "user", "content": "帮我查订单 A9999"}]})

# 打印完整轨迹，观察异常是否被转成了 ToolMessage
for i, msg in enumerate(result["messages"]):
    # 打印序号与消息类型
    print(f"[{i}] {type(msg).__name__}: {str(getattr(msg, 'content', ''))[:130]!r}")

输出：

[0] HumanMessage: '帮我查订单 A9999'
[1] AIMessage: ''
[2] ToolMessage: "工具错误：KeyError - 'A9999'。请检查参数后重试。"
[3] AIMessage: '抱歉，我查询订单 A9999 时没有找到对应的物流信息，提示订单号不存在。请您确认一下订单号是否正确，或者提供其他信息，我再帮您查询。'

同样是抛 KeyError 的工具，加了中间件之后进程就不崩了，异常变成一条 ToolMessage，模型据此给出得体回答。和 §9.1 的崩溃对照，中间件的价值一目了然。

wrap_tool_call 的执行位置值得理解一下：

模型发出 tool_calls
   ↓
handle_tool_errors(request, handler)   ← 你的拦截器在这里
   ↓  try: handler(request)
   实际执行 lookup_order
   ↓  except: 换成错误 ToolMessage
ToolMessage 回灌给模型

它包在每一次工具执行的外面。所以除了兜异常，同样的位置还能做超时控制、调用计时、参数审计、敏感操作拦截——第 10 章会用它做安全护栏。

一个实现细节：ToolCallRequest 从 langchain.tools.tool_node 导入，实际类型是 langgraph.prebuilt.tool_node.ToolCallRequest（LangChain 做了转发）。两条路径都能用，本文统一用前者。

9.4. 两层怎么配合 #

错误类型 优先做法
单号不存在、权限不足、参数校验失败 工具内 return "错误：..."
空指针、超时、第三方 500、依赖库异常 wrap_tool_call 兜底

为什么不能只留中间件、省掉工具内防御？因为两者给模型的信息质量差很多：

工具内返回 中间件兜底
模型看到 错误：未找到订单 A9999。可用示例：A1001、A1002。 工具错误：KeyError - 'A9999'。请检查参数后重试。
模型能否自纠 能，知道该试哪些值 难，KeyError 对它没有业务含义
泄露实现细节 不会 会（异常类型、内部键名）

最后一行还是个安全问题：把原始异常文本回灌给模型，等于可能把内部表名、文件路径、SQL 片段送进对话，而对话内容通常会展示给用户。生产环境建议中间件只回一句笼统的「工具执行失败，请稍后重试」，详细异常写进日志。

原则：

可预期的业务错误 → 工具内返回说明（信息友好）；未预期异常 → middleware 兜底（保证不崩）；两者都要让模型「看得到」，但兜底那层别把内部细节抖出去。

10. 业务工具包 #

前面各节是零件；本节把它们收成可复用模块——本章实战。

设计目标：

  1. 多工具：天气 / 计算 / 查单 / 制度，覆盖选型与互斥描述
  2. 有 schema：天气用 Pydantic；计算用 Literal 运算符（比裸 eval 更安全）
  3. 有错误回灌：查单、制度未命中时返回「错误：...」
  4. 可导出：BUSINESS_TOOLS + build_office_agent()，第 9 章可继续叠策略

数据仍用模拟字典，方便无外部依赖跑通；接真实 API 时只换函数体即可。

10.1. business_toolkit.py # 

# 从 typing 导入 Literal，用于把参数取值限制在固定选项内
from typing import Literal

# 从 pydantic 导入 BaseModel 与 Field，用于声明入参 schema
from pydantic import BaseModel, Field
# 从 langchain.agents 导入 create_agent 工厂函数
from langchain.agents import create_agent
# 从 langchain.tools 导入 tool 装饰器
from langchain.tools import tool
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)


# ---------- 模拟数据 ----------
# 订单库：键为订单号，值含状态与预计天数
ORDERS = {
    # 已发货订单
    "A1001": {"status": "已发货", "eta_days": 2},
    # 运输中订单
    "A1002": {"status": "运输中", "eta_days": 1},
    # 已签收订单
    "A1003": {"status": "已签收", "eta_days": 0},
}

# 制度库：键为主题关键词，值为条文摘要
POLICIES = {
    # 报销制度
    "报销": "差旅报销需在返程 7 日内提交，单笔超 1000 元需主管审批。",
    # 请假制度
    "请假": "年假提前 3 天申请；病假需当日同步直属上级。",
    # 加班制度
    "加班": "加班需事先在系统提交，月末统一调休或结算。",
}


# ---------- 工具定义 ----------
# 用 Pydantic 声明天气工具的入参（对应 §5.2 的做法）
class WeatherInput(BaseModel):
    # 类 docstring 成为 schema 的顶层描述
    """天气查询入参。"""

    # 必填字段：城市名，description 帮助模型正确填写
    city: str = Field(description="城市名，如北京、上海")
    # 枚举字段：只能填两种单位之一，默认摄氏
    units: Literal["C", "F"] = Field(
        # 有默认值 → 模型可以不填
        default="C",
        # 字段说明
        description="温度单位",
    )


# 挂上 Pydantic schema
@tool(args_schema=WeatherInput)
# 参数名必须与 WeatherInput 的字段一一对应
def get_weather(city: str, units: str = "C") -> str:
    # docstring 成为工具描述，写明「何时必须调用」
    """查询城市当前天气。用户问天气时必须调用，不要编造。"""
    # 根据单位选择模拟温度值
    temp = 26 if units == "C" else 79
    # 返回简洁字符串
    return f"{city} 当前约 {temp}°（{units}），晴间多云。"


# 用 description 覆盖描述，写成带指令性的选型说明
@tool(
    description="计算两个数的加减乘除。任何算术都请调用，不要心算。",
)
# op 用 Literal 限定为四个运算符，从结构上避免了 §4.3 的 eval 隐患
def calculate(a: float, b: float, op: Literal["+", "-", "*", "/"]) -> str:
    # 已传 description，这句 docstring 对模型不可见
    """对 a、b 做四则运算。"""
    # 加法
    if op == "+":
        # 结果转字符串返回
        return str(a + b)
    # 减法
    if op == "-":
        # 结果转字符串返回
        return str(a - b)
    # 乘法
    if op == "*":
        # 结果转字符串返回
        return str(a * b)
    # 走到这里说明是除法，先挡住除零，返回错误说明而不是 raise
    if b == 0:
        # 错误文案让模型知道该换参数
        return "错误：除数不能为 0。"
    # 除法
    return str(a / b)


# 查单工具
@tool
# 入参为订单号
def lookup_order(order_id: str) -> str:
    # 描述里说明订单号格式
    """按订单号查询物流状态。订单号形如 A1001。"""
    # 规范化输入：容忍空格和小写（§9.2 验证过模型真会填小写）
    key = order_id.strip().upper()
    # 用 .get() 而不是 []，查不到返回 None 而不抛 KeyError
    row = ORDERS.get(key)
    # 未命中时返回错误说明，并给出可用示例便于模型自纠
    if not row:
        # 错误文案带上可用订单号，模型能据此追问用户
        return f"错误：未找到订单 {order_id}。可用示例：A1001、A1002、A1003。"
    # 命中时拼装状态文本
    return f"订单 {key}：{row['status']}，预计 {row['eta_days']} 天后相关节点完成。"


# 制度查询工具
@tool
# 入参为制度主题
def lookup_policy(topic: str) -> str:
    # 描述里列出可选主题，帮助模型填参
    """查询公司制度摘要。topic 可为：报销、请假、加班。"""
    # 遍历制度库，用「关键词包含」做模糊匹配
    for key, text in POLICIES.items():
        # 只要主题里含有关键词就算命中，能容忍「报销流程」这类问法
        if key in topic:
            # 命中即返回条文
            return text
    # 全部未命中时返回错误说明与可选项
    return "错误：未匹配到制度。请尝试：报销 / 请假 / 加班。"


# 导出工具包列表，供 Agent 或其他模块复用
BUSINESS_TOOLS = [get_weather, calculate, lookup_order, lookup_policy]


# 用函数封装 Agent 构造，方便复用与测试
def build_office_agent():
    # 函数说明
    """组装多工具办公助理。"""
    # 返回配置好的 Agent
    return create_agent(
        # 模型标识
        model="deepseek:deepseek-v4-flash",
        # 一次挂载全部四个工具
        tools=BUSINESS_TOOLS,
        # 系统提示里逐一点明「什么问题用哪个工具」，帮助选型
        system_prompt=(
            "你是公司办公助手，回答简洁可靠。"
            "查天气用 get_weather；算术用 calculate；"
            "查订单用 lookup_order；问制度用 lookup_policy。"
            "禁止编造订单状态与制度条文。"
        ),
    )


# 只有直接运行本文件时才执行下面的演示代码
if __name__ == "__main__":
    # 构造 Agent
    agent = build_office_agent()
    # 一句话里塞三个不同类型的问题，观察模型如何选型
    result = agent.invoke(
        {
            # 输入是带 messages 键的字典
            "messages": [
                {
                    # 角色为用户
                    "role": "user",
                    # 同时涉及天气、订单、算术
                    "content": "上海天气怎么样？订单 A1001 到哪了？另外 128+256 等于多少？",
                }
            ]
        }
    )
    # 打印最终回答
    print(result["messages"][-1].content)

10.2. 真实运行结果 #

先各自单测（对应下面自测清单的第 1 步）：

上海 当前约 26°（celsius），晴间多云。
384.0
订单 A1001：已发货，预计 2 天后相关节点完成。
错误：未找到订单 A9999。可用示例：A1001、A1002、A1003。
差旅报销需在返程 7 日内提交，单笔超 1000 元需主管审批。

再跑组合问题，完整轨迹如下：

[0] HumanMessage: '上海天气怎么样？订单 A1001 到哪了？另外 128+256 等于多少？'
[1] AIMessage: ''
     -> get_weather({'city': '上海'})
     -> lookup_order({'order_id': 'A1001'})
     -> calculate({'a': 128, 'b': 256, 'op': '+'})
[2] ToolMessage: '上海 当前约 26°（celsius），晴间多云。'
[3] ToolMessage: '订单 A1001：已发货，预计 2 天后相关节点完成。'
[4] ToolMessage: '384.0'
[5] AIMessage: '三件事都查到了：...'

最终回答：

三件事都查到了：

1. **上海天气**：当前约 26°C，晴间多云。
2. **订单 A1001**：已发货，预计 2 天后完成相关节点。
3. **128+256**：等于 384。

还有其他需要帮忙的吗？

这段轨迹有三点值得注意：

  1. 一条 AIMessage 里带了 3 个 tool_calls（§2 预告过的并行调用）。模型没有一个一个来，而是一次把三个申请单全递出来，接着是 3 条 ToolMessage。整个过程只用了 2 次模型调用，不是 4 次——这对延迟和成本影响很大。
  2. 选型全对：天气→get_weather、订单→lookup_order、算术→calculate，参数也都填对了（包括 op: '+' 这个枚举）。功劳属于「系统提示逐一点明用途」+「描述互斥」+「枚举锁死取值」这三件事的叠加。
  3. calculate 返回了 384.0（§8 提到的 float 问题）。模型在最终回答里正确写成了「384」，但观察值里的 .0 会留在日志中。想干净的话，把 a: float, b: float 换成 int，或在返回前判断一下是否为整数。

建议自测顺序：

  1. 对每个工具单独 tool.invoke({...})（先保证业务逻辑对）
  2. 再 build_office_agent() 问组合问题（看选型与协作）
  3. 用第 2 / 4 章方法打印完整 messages，确认调了哪些工具、参数是否合理
  4. 故意问不存在的订单号，确认走「错误：...」而不是编造
  5. 故意问「深圳的社保政策」这类工具覆盖不到的问题，看它是否老实说不知道

这套工具包就是本章实战；第 9 章会在此基础上讲多工具协作与动态策略。

11. 实用约定与坑 #

下面几条能避开大多数初学翻车。带 ⚠ 的是静默失败——不报错，只是悄悄变差，最难发现：

定义侧

执行侧

口诀：

说明写清，schema 写严；先单测工具，再交给模型；错误要回灌，不要只崩进程。

12. 练习 #

每题做完后，尽量打印完整 messages 或对工具做 invoke 单测，而不是只看最终一句。

  1. 看清模型收到了什么：用 convert_to_openai_tool 打印 business_toolkit.py 里四个工具的定义，确认每个参数都有 description。找出哪个工具的参数说明最弱并补上。
  2. 对比 parse_docstring：写一个三参数工具，分别在加与不加 parse_docstring=True 的情况下打印 args，把差异记下来。
  3. 复现延迟失败：故意让 args_schema 的字段名和函数参数名不一致，确认构造阶段没有任何报错，只有 invoke 时才 TypeError。
  4. 复现崩溃再修好：把 lookup_order 改成 ORDERS[key] 直接取值，查一个不存在的订单号，确认 agent.invoke() 抛 KeyError；然后分别用「工具内返回错误」和 wrap_tool_call 修好，对比模型看到的两种错误文本。
  5. 加工具：在业务工具包中增加 get_employee_email(name: str) -> str（可用字典模拟），并更新 system_prompt。注意查不到时要返回「错误：...」。
  6. Pydantic 入参：给 lookup_order 加上 args_schema，增加可选字段 verbose: bool = False（为 True 时多返回一行温馨提示）。
  7. 手工环：不用 create_agent，只用 bind_tools + ToolMessage 完成一次「查 A1002 订单」。
  8. return_direct 对照：给 fetch_order_status 开关 return_direct=True，分别打印 len(result["messages"]) 和 type(result["messages"][-1]).__name__，确认是 3/ToolMessage 与 4/AIMessage。
  9. 注入参数：给 lookup_order 加一个 user_id: Annotated[str, InjectedToolArg]，确认它不在模型可见的 args 里，并在工具内用它做一次简单的权限判断。

13. 本章小结 #

  1. 工具是模型可请求、由代码执行的外部能力；日常用 create_agent(tools=...)，细控用 bind_tools。模型只是「填申请单」，跑腿的永远是你的代码。
  2. 模型能看到的只有三样：工具名、描述、参数 schema。用 convert_to_openai_tool 可以看到发给 API 的原文。
  3. docstring 里的 Args: 段默认不会变成参数说明，要么加 parse_docstring=True，要么改用 Pydantic args_schema + Field(description=...)。后者更适合业务工具。
  4. 工具名必须匹配 ^[a-zA-Z0-9_-]+$，中文名会被 API 拒绝；中文放描述里没问题。
  5. args_schema 同时是模型的说明书和本地的运行时校验；但字段名与函数参数名不一致属于延迟失败，写完立刻单测。
  6. bind_tools 只产生 tool_calls；执行与 ToolMessage 回灌要自己做或交给 Agent。一条 AIMessage 可以带多个调用，回灌时一条都不能少。
  7. 返回值默认用简洁 str；return_direct=True 省一次模型调用，但会让最后一条消息变成 ToolMessage。
  8. 错误处理是两层：工具内 return "错误：..." 给模型友好且可自纠的信息，wrap_tool_call 兜住想不到的异常保证不崩。完全不处理时，一次 KeyError 就能打崩整个 Agent。
  9. 本章产出：业务工具包（天气 / 计算 / 查单 / 制度），为第 9 章多工具办公助理打底。

工具如果已经跑在别的进程里，或要同时给多个宿主用，不要把源码贴进 Agent，改走 第 17 章 MCP。

下一章：create_agent 深入——系统提示策略、多工具协作、动态选模型 / 工具，做出更完整的多工具办公助理。

← 上一节 7.Runnable与LCEL 下一节 9.create_agent深入 →