Day 29 / 共 60 天 · 阶段5 工具系统

工具调用:把 LLM 的"我要用某工具"落到一次真调用

工具能跑了(D27/D28),但 LLM 说的是一句可能格式跑偏的话:"Action: 搜索工具 / Action Input: {city:上海}"。中间要把它变成一次精确的 tool.invoke(...)。今天读 tools/tool_calling.pyToolCalling 这个小而关键的数据结构)和 tools/tool_usage.py 的调度逻辑:工具名对不上怎么模糊匹配、参数是破损 JSON 怎么一路修、解析失败怎么三级降级重试。这一层专治"LLM 不靠谱"。

📍 你在 60 天里的位置(阶段5 工具系统 · 共 6 天)
D27 BaseTool D28 StructuredTool D29 工具调用 D30 缓存复用 D31 MCP 工具 D32 自定义与最佳实践
💡 先用一个类比兜住今天 LLM 就像一个字写得潦草、还老写错菜名的顾客。他点单写"来份 蕃茄炒蛋"(错别字),参数写成 {份数:2(漏了右括号)。tool_usage 就是那个经验老到的服务员:菜名对不上就找最像的那道菜(模糊匹配 _select_tool),订单 JSON 坏了就一层层想办法读懂_validate_tool_input 修复),实在读不懂就换个方式再问一遍(三级降级 _tool_calling)。目标只有一个:把潦草订单变成后厨能执行的标准工单。
L01

痛点:LLM 的输出天生不可靠

🤔 痛点Day 08 的执行循环里,模型输出被解析成 AgentAction(tool="搜索", tool_input="...")。但模型给的工具名可能带空格、大小写不一致、甚至写了个近似名;参数可能是标准 JSON,也可能是 Python 字典字面量,还可能是缺引号/缺括号的破损串。如果直接拿去调用,十有八九报错。谁来把这些"人类级别的模糊"变成"机器级别的精确"?
💡 一句话本质 ToolUsage 是"工具调用的容错调度中枢"。它做三件事:① 把模型的话解析成结构化的 ToolCalling(tool_name + arguments),解析走"先试直接解析 → 不行用函数调用 LLM → 再不行重试"的三级降级;② 用字符串相似度把模型给的工具名匹配到真实工具;③ 用四种手段依次尝试把破损的参数串修成 dict。核心思想:LLM 会犯错是常态,容错不是可选项而是必需品。

关键方法一览(tool_usage.py:76ToolUsage 类):

# tool_usage.py
class ToolUsage:
    def use(self, calling, tool_string): ...            # :132 总入口
    def parse_tool_calling(self, tool_string): ...      # :121 解析入口
    def _tool_calling(self, tool_string): ...           # :861 三级降级解析
    def _validate_tool_input(self, tool_input): ...     # :884 四招修 JSON
    def _select_tool(self, tool_name): ...              # :759 模糊匹配工具
    def _use(self, tool_string, tool, calling): ...     # :469 真正执行
L02

ToolCalling:一个"要调什么"的最小结构

整个 tool_calling.py 只有两个小模型(tool_calling.py:11):

# tool_calling.py:11
class ToolCalling(BaseModel):
    tool_name: str = Field(..., description="The name of the tool to be called.")
    arguments: dict[str, Any] | None = Field(
        ..., description="A dictionary of arguments to be passed to the tool.")

class InstructorToolCalling(PydanticBaseModel):
    tool_name: str = PydanticField(..., description="The name of the tool to be called.")
    arguments: dict[str, Any] | None = PydanticField(
        ..., description="A dictionary of arguments to be passed to the tool.")
只有两个字段"调哪个工具"(tool_name)+"传什么参数"(arguments)。一次工具调用的本质就这么点信息。
arguments 可为 None有些工具不需要参数(比如"获取当前时间"),所以允许 None
为何有两个几乎一样的类ToolCalling 用普通方式解析;InstructorToolCalling 用 instructor 库(强约束 LLM 输出结构)解析。分开是为了给两种解析路径各自的目标类型(L03 会看到怎么选)。
大白话你可以把 ToolCalling 理解成一张"工单":只写"叫谁 + 给它什么料"。今天大部分工作,就是想尽办法从 LLM 那段潦草输出里,可靠地填好这张工单。填好了,后面 invoke(D28)就能照单执行。
数据结构:从潦草文本到标准工单 LLM 原始输出 Action: 搜索工具 Action Input: {city:上海 ToolUsage 解析+修复+匹配 ToolCalling(工单) tool_name: "搜索工具" arguments: {"city":"上海"} 三个动作各治一种"不靠谱":名字→_select_tool | 参数→_validate_tool_input | 整体→_tool_calling 名字对不上 → 相似度 JSON 破损 → 四招修 解析失败 → 三级降级
图注:三种"不靠谱"各有一招对治,最终填好 ToolCalling 这张标准工单。
L03

_tool_calling:三级降级的解析策略

解析核心 _tool_callingtool_usage.py:861)是一个"try 里套 try、失败就递归重试"的结构:

# tool_usage.py:861
def _tool_calling(self, tool_string):
    try:
        try:
            return self._original_tool_calling(tool_string, raise_error=True)  # ① 首选:直接解析
        except Exception:
            if self.function_calling_llm:
                return self._function_calling(tool_string)   # ② 退而求其次:让 LLM 帮忙解析
            return self._original_tool_calling(tool_string)  # 无备用 LLM → 再试一次不抛错版
    except Exception as e:
        self._run_attempts += 1
        if self._run_attempts > self._max_parsing_attempts:  # ③ 试够上限 → 放弃,返回错误提示
            self._telemetry.tool_usage_error(llm=self.function_calling_llm)
            if self.task:
                self.task.increment_tools_errors()
            return ToolUsageError(
                f"{I18N_DEFAULT.errors('tool_usage_error').format(error=e)}\n"
                f"Moving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}")
        return self._tool_calling(tool_string)               # ★递归重试
① _original_tool_calling首选:直接从 self.action(已解析的工具名+输入)构造 ToolCalling,参数走 _validate_tool_input 修复。最快、最省钱。
② _function_calling直接解析失败且配了 function_calling_llm → 让一个 LLM 用结构化输出专门帮忙把这段话转成 {tool_name, arguments}。更可靠但要多花一次 LLM 调用。
③ 试够 _max_parsing_attempts重试次数用完(默认 3 次,大模型 2 次,见 tool_usage.py:100)→ 不再硬刚,返回一个 ToolUsageError,里面附上"可用工具列表"提示模型换招。
递归 self._tool_calling还没到上限就递归自己重来一遍。每递归一次 _run_attempts += 1,靠这个计数做刹车。
控制流:三级降级 + 递归重试 ① 直接解析(免费) ② LLM 帮解析(付费) ③ 超上限 → 返回错误 失败 仍失败 递归 _tool_calling(_run_attempts+1) 任一级成功 → 返回 ToolCalling ✅ 未到 _max_parsing_attempts 就重试;到了才放弃
图注:从免费到付费到放弃,逐级升级;每轮失败递归重试,用计数封顶防烧钱。
💡 设计取舍①:为什么先"免费直接解析"再"付费 LLM 解析" 让 LLM 帮忙解析(_function_calling)最可靠,但每次都要多一轮模型调用——又慢又费钱。所以源码把它放在"备选"位:先用零成本的直接解析(_original_tool_calling + JSON 修复)碰运气,只有失败了才升级到 LLM 解析。成本优先、可靠性兜底。而且用 _max_parsing_attempts 封顶,防止在一个死解析上无限烧钱。这是"分层容错 + 成本控制"的经典组合。
L04

_validate_tool_input:修 JSON 的四连招

参数往往是破损的。_validate_tool_inputtool_usage.py:884)依次用四种手段尝试:

# tool_usage.py:884
def _validate_tool_input(self, tool_input: str | None) -> dict:
    if tool_input is None:
        return {}
    if not isinstance(tool_input, str) or not tool_input.strip():
        raise Exception("Tool input must be a valid dictionary in JSON or Python literal format")

    try:                                    # 招式1:标准 JSON
        arguments = json.loads(tool_input)
        if isinstance(arguments, dict):
            return arguments
    except (JSONDecodeError, TypeError):
        pass

    try:                                    # 招式2:Python 字面量(单引号、True/None 等)
        arguments = ast.literal_eval(tool_input)
        if isinstance(arguments, dict):
            return arguments
    except (ValueError, SyntaxError):
        repaired_input = repair_json(tool_input)

    try:                                    # 招式3:JSON5(宽松 JSON,允许尾逗号/注释)
        arguments = json5.loads(tool_input)
        if isinstance(arguments, dict):
            return arguments
    except (JSONDecodeError, ValueError, TypeError):
        pass

    try:                                    # 招式4:repair_json 强力修复(补括号/引号)
        repaired_input = str(repair_json(tool_input, skip_json_loads=True))
        if self.agent and self.agent.verbose:
            PRINTER.print(content=f"Repaired JSON: {repaired_input}", color="blue")
        arguments = json.loads(repaired_input)
        if isinstance(arguments, dict):
            return arguments
    except Exception as e:
        ...
    error_message = "Tool input must be a valid dictionary in JSON or Python literal format"
    self._emit_validate_input_error(error_message)      # 四招全败 → 发事件 + 抛错
    raise Exception(error_message)
招式1 json.loads最标准。模型输出规范时一步到位。
招式2 ast.literal_eval模型有时输出 Python 风格:{'city': '上海'}(单引号)、True/None。标准 JSON 不认,Python 字面量能读。
招式3 json5.loads宽松 JSON:容忍尾逗号、注释、无引号键。覆盖模型的"半规范"输出。
招式4 repair_json★最强力:skip_json_loads=True 直接进修复引擎,能补上缺失的括号、引号。{city:上海 这种也能救回来。verbose 时会打印修复后的样子。
全败 → 抛错 + 发事件四招都不行才放弃,_emit_validate_input_error 发一个 ToolValidateInputErrorEvent(阶段4 D26 的事件系统)方便观测,再抛异常回到 L03 的重试。
⚠️ 边界:招式必须"渐进",不能一上来就用最强的 repair_json 为什么不直接用 repair_json(招式4)省事?因为强力修复可能"过度纠正"——把本来合法但它没料到的结构改坏。所以顺序是从最严格(json)到最宽松(repair):能用标准方式读通的绝不动它,只有前面都失败才动用"猜着修"。反模式:图省事直接上最激进的解析器,反而可能把好数据改错。"最小必要修复"才安全。
L05

_select_tool:工具名对不上就找最像的

模型给的工具名未必精确,_select_tooltool_usage.py:759)用字符串相似度匹配:

# tool_usage.py:759
def _select_tool(self, tool_name: str) -> Any:
    sanitized_input = sanitize_tool_name(tool_name)
    order_tools = sorted(                           # ★按相似度从高到低排序
        self.tools,
        key=lambda tool: SequenceMatcher(
            None, sanitize_tool_name(tool.name), sanitized_input).ratio(),
        reverse=True)
    for tool in order_tools:
        sanitized_tool = sanitize_tool_name(tool.name)
        if (sanitized_tool == sanitized_input       # 完全相等 → 直接命中
                or SequenceMatcher(None, sanitized_tool, sanitized_input).ratio() > 0.85):
            return tool                             # 相似度 > 0.85 → 认为是它
    if self.task:
        self.task.increment_tools_errors()
    if tool_name and tool_name != "":
        error = f"Action '{tool_name}' don't exist, these are the only available Actions:\n{self.tools_description}"
        crewai_event_bus.emit(self, ToolSelectionErrorEvent(...))
        raise Exception(error)
    error = f"I forgot the Action name, these are the only available Actions: {self.tools_description}"
    ...
    raise Exception(error)
sanitize_tool_name先把两边名字都规范化(去空格、统一格式)。Day 27 就见过它,这里是它发挥作用的地方——保证"搜索 工具"和"搜索工具"能对上。
SequenceMatcher.ratio()Python 标准库的字符串相似度,返回 0~1。用它给所有工具打分排序。
> 0.85 阈值★相似度超过 0.85 就认为模型指的是这个工具。容忍小拼写差异("calcualtor" vs "calculator"),又不至于乱认。
两种错误消息名字非空但没匹配上 → "这个 Action 不存在,可用的是……";名字为空(模型忘了写)→ "我忘了写 Action 名,可用的是……"。都附上工具列表引导模型改正。
📝 例子:模糊匹配救场 真实工具名 Search the internet。模型输出 Action: search internet(漏词)。
→ sanitize 后比对,相似度约 0.8~0.9,超过 0.85 阈值 → 匹配成功,照常执行。
但若模型写 Action: 计算器(和任何工具都不像)→ 相似度太低 → 抛错并把可用工具列表喂回,让模型重选。
L06

use:把解析、选择、执行串成总调度

对外总入口 usetool_usage.py:132)把前面几块串起来:

# tool_usage.py:132
def use(self, calling, tool_string) -> str:
    if isinstance(calling, ToolUsageError):          # 解析阶段就失败了
        error = calling.message
        if self.agent and self.agent.verbose:
            PRINTER.print(content=f"\n\n{error}\n", color="red")
        if self.task:
            self.task.increment_tools_errors()
        return error                                 # 把错误当结果返回(喂回给 LLM)

    try:
        tool = self._select_tool(calling.tool_name)  # ★选工具(L05)
    except Exception as e:
        error = getattr(e, "message", str(e))
        if self.task:
            self.task.increment_tools_errors()
        if self.agent and self.agent.verbose:
            PRINTER.print(content=f"\n\n{error}\n", color="red")
        return error                                 # 选不到 → 返回错误提示
    ...
    # 之后走 _use(...) 真正执行(L07 / D30)
先判 ToolUsageError如果传进来的 calling 本身就是解析失败的错误对象,直接把错误信息 return——它会作为"观察"喂回 LLM。
_select_tool 包 try选工具可能抛错(没匹配上)。这里 catch 住,转成返回值而不是让异常冒泡崩掉 Agent。
increment_tools_errors每次工具相关错误都给 task 的错误计数 +1——用于统计和后续"错误太多就收尾"的判断。
错误即返回值★贯穿全篇的模式:几乎所有错误都转成字符串返回,而非抛异常。因为返回值会进 LLM 的对话历史,让模型"看到问题、自己纠正"。
💡 为什么"错误当返回值"是 Agent 框架的核心哲学普通程序里,错误就该抛异常、让上层处理。但在 Agent 里,LLM 就是那个"上层处理者"——它有推理能力,能看懂"你的工具名不存在,可用的是 A/B/C"并自己改。所以把错误写成 LLM 能读的自然语言、塞回对话,比抛异常崩溃有用得多。这是"人在环"之外的"模型在环自愈"。
L07

参数过滤:只喂工具认识的字段

_use 真正调用工具前,会把参数按 schema 过滤一遍(tool_usage.py:570):

# tool_usage.py:570
if calling.arguments:
    try:
        acceptable_args = tool.args_schema.model_json_schema()["properties"].keys()
        arguments = {                      # ★只保留 schema 里声明的字段
            k: v for k, v in calling.arguments.items() if k in acceptable_args}
        result = tool.invoke(input=arguments, config=fingerprint_config)
    except Exception:
        arguments = calling.arguments      # 过滤失败 → 用原始参数兜底
        result = tool.invoke(input=arguments, config=fingerprint_config)
else:
    result = tool.invoke(input={}, config=fingerprint_config)  # 无参数工具
acceptable_args从工具的 args_schema 里取出"合法字段名集合"。
字典推导过滤★只保留模型给的参数里工具认识的那些。模型有时会多塞无关字段(幻觉出一个参数),过滤掉能避免 invoke 报"未知参数"。
except 兜底如果连取 schema 都失败(极端情况),退回用原始参数直接调,尽量别让工具没机会跑。
tool.invoke(config=fingerprint_config)这里就接上了 D28 的 CrewStructuredTool.invoke——参数过滤后,正式落到一次真调用。fingerprint_config 带上 Agent 指纹用于审计。
💡 设计取舍②:过滤"多余参数"而不是直接报错 模型幻觉出一个不存在的参数很常见。严格做法:多一个参数就报错,逼模型重来——干净但浪费一整轮。源码做法:静默丢掉多余字段,用剩下合法的照常执行。宽容、少一轮重试。代价是可能掩盖"模型理解偏差"(它以为有这个参数说明它没完全读懂工具)。这是"鲁棒性优先于严格性"的权衡——对追求"能跑通"的 Agent 场景通常划算,但调试期你可能希望它报出来。
L08

边界 + 今日小结

⚠️ 边界:三级降级 + 递归重试可能"多花好几次 LLM 调用" 最坏情况:直接解析失败 → 调 function_calling_llm 解析(1 次 LLM)→ 又失败 → 递归重试 → 再调一次……直到 _max_parsing_attempts。也就是说一次工具调用背后可能藏着好几次隐形的 LLM 请求,既慢又费钱。所以:① 工具名和描述要写清楚(减少模型犯错);② 用支持原生函数调用的模型(Day 08 的原生路径基本绕开这套文本解析);③ 关注 _max_parsing_attempts——它是防止无限烧钱的最后一道闸。

👶 小白:这套复杂的容错,原生函数调用模式也走吗?

👨‍🏫 老师:基本不走。原生模式(Day 08 _invoke_loop_native_tools)里,模型直接吐结构化的 tool_calls,工具名和参数都是规范的,不需要模糊匹配和修 JSON。今天这套主要服务文本 ReAct 模式(老模型/无原生支持)。这也解释了为什么现代首选原生模式——它把一大堆"猜模型意图"的脏活省掉了。

🧠 今天你应该能回答

  • ToolCalling 只有哪两个字段?为什么有两个几乎一样的类?
  • _tool_calling 的三级降级是哪三级?为什么"免费的"排前面?
  • _validate_tool_input 的四招修 JSON 顺序为什么"从严到宽"?
  • _select_tool 用什么做模糊匹配?阈值是多少?
  • 为什么这一层几乎所有错误都"当返回值"而不是抛异常?
  • 参数过滤为什么静默丢多余字段?这个宽容有什么代价?

✋ 10 分钟动手

P=lib/crewai/src/crewai/tools
sed -n '1,25p'     $P/tool_calling.py        # ToolCalling 结构
sed -n '861,883p'  $P/tool_usage.py          # 三级降级
sed -n '884,930p'  $P/tool_usage.py          # 四招修 JSON
sed -n '759,802p'  $P/tool_usage.py          # 模糊匹配
python -c "
from difflib import SequenceMatcher
print(SequenceMatcher(None,'search the internet','search internet').ratio())
import json, ast
print(ast.literal_eval(\"{'city':'上海'}\"))   # 单引号字面量
"
明日预告 · Day 30:今天 _use 里执行工具时,其实还夹着一层"先查缓存、命中就不重复跑"。明天读 agents/tools_handler.py + agents/cache/cache_handler.py + tool_usage.py 的缓存分支:on_tool_use 怎么记结果、cache_function 怎么决定要不要缓存、_check_tool_repeated_usage 怎么防"连续调同一个工具同参数"。
← Day 28 StructuredTool Day 30 · 缓存复用 →