工具调用:把 LLM 的"我要用某工具"落到一次真调用
工具能跑了(D27/D28),但 LLM 说的是一句可能格式跑偏的话:"Action: 搜索工具 / Action Input: {city:上海}"。中间要把它变成一次精确的 tool.invoke(...)。今天读 tools/tool_calling.py(ToolCalling 这个小而关键的数据结构)和 tools/tool_usage.py 的调度逻辑:工具名对不上怎么模糊匹配、参数是破损 JSON 怎么一路修、解析失败怎么三级降级重试。这一层专治"LLM 不靠谱"。
{份数:2(漏了右括号)。tool_usage 就是那个经验老到的服务员:菜名对不上就找最像的那道菜(模糊匹配 _select_tool),订单 JSON 坏了就一层层想办法读懂(_validate_tool_input 修复),实在读不懂就换个方式再问一遍(三级降级 _tool_calling)。目标只有一个:把潦草订单变成后厨能执行的标准工单。痛点:LLM 的输出天生不可靠
AgentAction(tool="搜索", tool_input="...")。但模型给的工具名可能带空格、大小写不一致、甚至写了个近似名;参数可能是标准 JSON,也可能是 Python 字典字面量,还可能是缺引号/缺括号的破损串。如果直接拿去调用,十有八九报错。谁来把这些"人类级别的模糊"变成"机器级别的精确"?ToolUsage 是"工具调用的容错调度中枢"。它做三件事:① 把模型的话解析成结构化的 ToolCalling(tool_name + arguments),解析走"先试直接解析 → 不行用函数调用 LLM → 再不行重试"的三级降级;② 用字符串相似度把模型给的工具名匹配到真实工具;③ 用四种手段依次尝试把破损的参数串修成 dict。核心思想:LLM 会犯错是常态,容错不是可选项而是必需品。关键方法一览(tool_usage.py:76 的 ToolUsage 类):
# 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 真正执行
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)就能照单执行。_tool_calling:三级降级的解析策略
解析核心 _tool_calling(tool_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,靠这个计数做刹车。_function_calling)最可靠,但每次都要多一轮模型调用——又慢又费钱。所以源码把它放在"备选"位:先用零成本的直接解析(_original_tool_calling + JSON 修复)碰运气,只有失败了才升级到 LLM 解析。成本优先、可靠性兜底。而且用 _max_parsing_attempts 封顶,防止在一个死解析上无限烧钱。这是"分层容错 + 成本控制"的经典组合。_validate_tool_input:修 JSON 的四连招
参数往往是破损的。_validate_tool_input(tool_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(招式4)省事?因为强力修复可能"过度纠正"——把本来合法但它没料到的结构改坏。所以顺序是从最严格(json)到最宽松(repair):能用标准方式读通的绝不动它,只有前面都失败才动用"猜着修"。反模式:图省事直接上最激进的解析器,反而可能把好数据改错。"最小必要修复"才安全。_select_tool:工具名对不上就找最像的
模型给的工具名未必精确,_select_tool(tool_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: 计算器(和任何工具都不像)→ 相似度太低 → 抛错并把可用工具列表喂回,让模型重选。use:把解析、选择、执行串成总调度
对外总入口 use(tool_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 的对话历史,让模型"看到问题、自己纠正"。参数过滤:只喂工具认识的字段
_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 指纹用于审计。边界 + 今日小结
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':'上海'}\")) # 单引号字面量
"
_use 里执行工具时,其实还夹着一层"先查缓存、命中就不重复跑"。明天读 agents/tools_handler.py + agents/cache/cache_handler.py + tool_usage.py 的缓存分支:on_tool_use 怎么记结果、cache_function 怎么决定要不要缓存、_check_tool_repeated_usage 怎么防"连续调同一个工具同参数"。