主题
原文：https://www.mstx.cn/harness/419.html · 创建：2026-09-04 · 更新：2026-09-05 · 阅读：1136
工具调用不是"调用一个函数"那么简单
在3.2节中，我们搭建了一个最小Harness。那个Harness中的工具调用机制是最朴素的：把工具的Schema告诉LLM，LLM返回一个工具调用请求，我们解析参数，执行对应的Python函数，把结果放回对话。
这个流程在简单场景下能工作。但当你面对真实的Agent应用时，很快会发现一系列问题：
- LLM返回的参数格式不对怎么办？（它说{"city": "北京"}，但你的函数期望的是{"city_name": "北京"}）
- LLM调用了一个不存在的工具怎么办？
- LLM同时请求了多个工具调用，但它们之间有依赖关系怎么办？
- 工具执行耗时很长，用户在等待，体验很差怎么办？
- 工具的返回结果很大（比如查询返回了1000条记录），全部塞进上下文会爆掉怎么办？
这些问题揭示了一个事实：工具调用机制是Agent与外部世界之间的"外交官"。它不仅负责传递信息，还负责处理误解、协调时序、管理资源。一个设计良好的工具调用机制，能让LLM的能力得到充分发挥；一个粗糙的实现，则会让Agent频繁失败。
本节将基于3.2节的最小Harness，构建一个更健壮的工具调用机制。我们会逐步处理上述的每一个问题。
工具调用的完整生命周期
在深入实现之前，先理解工具调用在Agent循环中的完整生命周期：
┌─────────────────────────────────────────────────────────┐
│ 1. 准备阶段：将工具Schema注入LLM上下文 │
│ · 从Tool Registry获取所有工具的Schema │
│ · 组装为LLM API所需的tools参数 │
│ · 考虑：工具太多时如何选择？ │
├─────────────────────────────────────────────────────────┤
│ 2. 决策阶段：LLM返回响应 │
│ · 可能返回：文本回答 / 工具调用请求 / 两者混合 │
│ · 需要解析：工具名、参数JSON、调用ID │
│ · 考虑：参数解析失败怎么办？ │
├─────────────────────────────────────────────────────────┤
│ 3. 执行阶段：运行工具 │
│ · 查找工具、校验参数、执行函数 │
│ · 考虑：超时如何处理？失败如何重试？ │
│ · 考虑：多个工具调用是并行还是串行？ │
├─────────────────────────────────────────────────────────┤
│ 4. 反馈阶段：将结果返回给LLM │
│ · 将工具结果格式化为LLM可读的消息 │
│ · 考虑：结果太大如何截断？ │
│ · 考虑：失败信息如何传达给LLM？ │
└─────────────────────────────────────────────────────────┘这四个阶段构成了一个完整的工具调用周期。每个阶段都有其特定的工程挑战。让我们逐一实现。
阶段一：工具Schema的准备与优化
问题：工具Schema的质量直接影响LLM的表现
LLM对工具的理解完全依赖于你提供的Schema。一个写得不好的Schema，会让LLM：
- 选错工具（因为描述含糊）
- 填错参数（因为参数说明不清楚）
- 该调用时不调用（因为描述让LLM觉得这个工具不重要）
实现：Schema的标准化与质量检查
在ToolRegistry中添加Schema质量检查：
python
# harness/tools.py 的扩展
class ToolRegistry:
def __init__(self):
self._tools = {}
def register(self, name, description, func, parameters):
# 注册前进行基本的质量检查
self._validate_schema(name, description, parameters)
self._tools[name] = {
"name": name,
"description": description,
"func": func,
"parameters": parameters,
}
def _validate_schema(self, name, description, parameters):
"""验证工具Schema的基本质量"""
# 检查描述长度
if len(description) < 10:
print(f"警告：工具 '{name}' 的描述过短（{len(description)}字符），"
f"LLM可能无法准确理解其用途")
# 检查参数定义
if parameters.get("type") != "object":
raise ValueError(f"工具 '{name}' 的parameters必须是object类型")
properties = parameters.get("properties", {})
for prop_name, prop_def in properties.items():
if "description" not in prop_def:
print(f"警告：工具 '{name}' 的参数 '{prop_name}' 缺少description，"
f"LLM可能无法正确填写该参数")
if "type" not in prop_def:
raise ValueError(f"工具 '{name}' 的参数 '{prop_name}' 缺少type定义")
# 检查required参数是否都在properties中
required = parameters.get("required", [])
for req_param in required:
if req_param not in properties:
raise ValueError(f"工具 '{name}' 的required参数 '{req_param}' "
f"不在properties中")
def get_schemas(self, max_tools: int = None) -> list:
"""获取工具Schema，可选限制数量"""
schemas = []
for tool in self._tools.values():
schemas.append({
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["parameters"],
}
})
# 如果工具太多，只返回前max_tools个
# 后续章节会讨论更智能的工具选择策略
if max_tools and len(schemas) > max_tools:
print(f"警告：工具数量（{len(schemas)}）超过限制（{max_tools}），"
f"仅返回前{max_tools}个")
return schemas[:max_tools]
return schemas这个质量检查在注册时就发现问题，而非等到LLM调用失败后才发现Schema写得不好。这是一个重要的工程原则：越早发现问题，修复成本越低。
阶段二：LLM响应的解析与验证
问题：LLM的输出不可完全信任
LLM可能会：
- 返回格式错误的JSON
- 调用不存在的工具
- 遗漏必需的参数
- 参数类型错误（比如把字符串填进了数字参数）
实现：严格的响应解析
在Runtime中添加一个专门的解析方法：
python
# harness/runtime.py 的扩展
import json
from typing import List, Dict, Any, Optional
class ToolCallRequest:
"""一个解析后的工具调用请求"""
def __init__(self, call_id: str, name: str, arguments: Dict[str, Any]):
self.call_id = call_id
self.name = name
self.arguments = arguments
def __repr__(self):
return f"ToolCallRequest(name={self.name}, args={self.arguments})"
class AgentRuntime:
# ... 之前的代码 ...
def _parse_tool_calls(self, message) -> List[ToolCallRequest]:
"""解析LLM响应中的工具调用请求"""
calls = []
for tool_call in message.tool_calls:
call_id = tool_call.id
name = tool_call.function.name
# 解析参数JSON
try:
args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
# JSON解析失败：返回一个特殊的错误调用
# 让LLM看到错误并重新尝试
print(f"警告：工具 '{name}' 的参数JSON解析失败：{e}")
print(f"原始参数：{tool_call.function.arguments}")
args = {"__parse_error__": str(e)}
calls.append(ToolCallRequest(
call_id=call_id,
name=name,
arguments=args,
))
return calls
def _validate_tool_call(self, call: ToolCallRequest) -> Optional[str]:
"""验证工具调用是否合法，返回错误信息（如果非法）"""
# 检查工具是否存在
if call.name not in self.tools._tools:
return f"工具 '{call.name}' 不存在。可用工具：{list(self.tools._tools.keys())}"
# 检查参数是否为字典
if not isinstance(call.arguments, dict):
return f"工具 '{call.name}' 的参数必须是JSON对象"
# 检查是否有解析错误标记
if "__parse_error__" in call.arguments:
return f"参数JSON解析失败：{call.arguments['__parse_error__']}"
# 检查必需参数
tool_def = self.tools._tools[call.name]
required = tool_def["parameters"].get("required", [])
missing = [p for p in required if p not in call.arguments]
if missing:
return f"工具 '{call.name}' 缺少必需参数：{missing}"
return None # 验证通过这个解析和验证逻辑的核心思想是：不信任LLM的输出，但给LLM纠正的机会。当解析失败或验证不通过时，错误信息会被作为工具结果返回给LLM，让LLM看到问题并重新尝试。
阶段三：工具的执行策略
问题一：多个工具调用是并行还是串行？
LLM可以在一次响应中请求多个工具调用。这些调用可能是独立的（可以并行），也可能有依赖（必须串行）。
当前最简单可靠的策略：串行执行。 并行执行更快，但引入了一个风险：如果第二个工具的参数依赖于第一个工具的结果，并行就会出错。在Harness的当前阶段，可靠性比性能更重要。
python
def _execute_tool_calls(self, calls: List[ToolCallRequest]) -> List[Dict[str, Any]]:
"""串行执行多个工具调用"""
results = []
for call in calls:
# 验证
error = self._validate_tool_call(call)
if error:
# 验证失败：返回错误信息给LLM
results.append({
"tool_call_id": call.call_id,
"role": "tool",
"content": f"工具调用失败：{error}"
})
continue
# 执行（带重试）
result = self._execute_with_retry(call.name, call.arguments)
# 格式化结果
results.append({
"tool_call_id": call.call_id,
"role": "tool",
"content": result.content if result.success else f"工具执行失败：{result.error}"
})
return results注意：这里每个工具调用的结果都必须带上tool_call_id。这是LLM API的要求——它需要知道每个结果对应哪个调用请求。
问题二：工具执行超时
一个工具可能因为各种原因卡住：网络请求挂起、数据库查询慢、死循环。如果不加超时控制，整个Agent都会被卡住。
python
import signal
from contextlib import contextmanager
class TimeoutError(Exception):
pass
@contextmanager
def timeout_context(seconds: int):
"""在指定时间内执行代码，超时则抛出TimeoutError"""
def timeout_handler(signum, frame):
raise TimeoutError(f"操作超时（>{seconds}秒）")
# 设置信号处理器
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
class ToolRegistry:
def __init__(self, tool_timeout: int = 10):
self._tools = {}
self.tool_timeout = tool_timeout
def execute(self, name: str, arguments: Dict[str, Any]) -> ToolResult:
"""执行工具，带超时控制"""
import time
start = time.time()
if name not in self._tools:
return ToolResult(
success=False, content="",
error=f"未知工具：{name}",
tool_name=name
)
tool = self._tools[name]
try:
# 在超时上下文中执行
with timeout_context(self.tool_timeout):
result = tool["func"](**arguments)
duration = int((time.time() - start) * 1000)
return ToolResult(
success=True,
content=str(result),
tool_name=name,
duration_ms=duration
)
except TimeoutError as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False,
content="",
error=str(e),
tool_name=name,
duration_ms=duration
)
except Exception as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False,
content="",
error=f"{type(e).__name__}: {e}",
tool_name=name,
duration_ms=duration
)一个重要的注意点：signal.SIGALRM只在Unix/Linux系统上可用。如果你在Windows上开发，需要使用其他超时机制（比如concurrent.futures的线程池超时）。本书假设Unix环境，但你应该知道这个限制。
问题三：工具结果过大
一个查询工具可能返回几千条记录。如果全部塞进LLM的上下文，会消耗大量token，甚至超出窗口限制。
python
class ToolRegistry:
def __init__(self, tool_timeout: int = 10, max_result_length: int = 2000):
self._tools = {}
self.tool_timeout = tool_timeout
self.max_result_length = max_result_length
def _truncate_result(self, content: str) -> str:
"""截断过长的工具结果"""
if len(content) <= self.max_result_length:
return content
truncated = content[:self.max_result_length]
return truncated + f"\n... [结果已截断，原始长度{len(content)}字符，当前显示前{self.max_result_length}字符]"
def execute(self, name, arguments):
# ... 执行工具 ...
content = str(result)
content = self._truncate_result(content)
# ...截断不是最好的解决方案（最好的方案是让工具自己返回摘要），但它是一个安全底线——确保无论如何，工具结果不会撑爆上下文窗口。
阶段四：将结果反馈给LLM
问题：错误信息如何传达给LLM？
当工具执行失败时，你有两个选择：
- 把错误信息返回给LLM，让它决定怎么做
- 在Harness层面直接处理（比如重试后放弃）
经验法则：把错误信息返回给LLM。LLM看到错误后可能会：
- 换一个工具
- 调整参数后重试
- 向用户说明情况
这比Harness层面的静默处理更灵活，也更符合Agent的设计理念——让LLM参与决策。
实现：统一的反馈格式
python
def _build_tool_feedback(self, call: ToolCallRequest, result: ToolResult) -> Dict[str, Any]:
"""构建给LLM的工具结果消息"""
if result.success:
content = result.content
else:
# 失败时，给出结构化的错误信息
content = (
f"工具 '{call.name}' 执行失败。\n"
f"错误：{result.error}\n"
f"调用参数：{json.dumps(call.arguments, ensure_ascii=False)}\n"
f"请根据错误信息调整策略：尝试修改参数重试、使用其他工具、或告知用户。"
)
return {
"role": "tool",
"tool_call_id": call.call_id,
"content": content,
}关键细节：失败消息中包含了调用参数。这让LLM能看到自己填了什么参数，从而理解为什么失败——是参数错了，还是工具本身的问题。
完整的Runtime更新
整合以上所有改进，更新后的AgentRuntime核心循环如下：
python
class AgentRuntime:
def run(self, user_input: str) -> AgentResult:
start_time = time.time()
self.memory.initialize(user_input)
for iteration in range(self.config.runtime.max_iterations):
# 检查超时
elapsed = time.time() - start_time
if elapsed > self.config.runtime.timeout_seconds:
return AgentResult(
status="timeout",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason=f"超时（>{self.config.runtime.timeout_seconds}秒）"
)
# 调用LLM
try:
response = self.client.chat.completions.create(
model=self.config.llm.model,
messages=self.memory.get_context(),
tools=self.tools.get_schemas(),
temperature=self.config.llm.temperature,
max_tokens=self.config.llm.max_tokens,
)
except Exception as e:
elapsed = time.time() - start_time
return AgentResult(
status="error",
content=f"LLM调用失败：{e}",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason="LLM调用异常"
)
message = response.choices[0].message
if message.tool_calls:
# 解析工具调用
calls = self._parse_tool_calls(message)
print(f" [迭代{iteration + 1}] LLM请求{len(calls)}个工具调用")
for c in calls:
print(f" - {c.name}({json.dumps(c.arguments, ensure_ascii=False)})")
# 将LLM的决策加入记忆
self.memory.add_llm_message(message)
# 执行工具调用
feedbacks = self._execute_tool_calls(calls)
# 将结果加入记忆
for feedback in feedbacks:
self.memory.messages.append(feedback)
print(f" → {feedback['content'][:100]}...")
else:
# 正常终止
elapsed = time.time() - start_time
return AgentResult(
status="success",
content=message.content or "",
iterations=iteration + 1,
duration_ms=int(elapsed * 1000),
termination_reason="正常终止"
)
elapsed = time.time() - start_time
return AgentResult(
status="max_iterations",
content="已达到最大循环次数",
iterations=self.config.runtime.max_iterations,
duration_ms=int(elapsed * 1000),
termination_reason=f"达到最大迭代次数"
)测试改进后的工具调用机制
用几个边界场景来测试：
场景一：LLM调用不存在的工具
你> 帮我查一下明天的股票行情
[迭代1] LLM请求1个工具调用
- get_stock_price({"symbol": "AAPL"})
→ 工具调用失败：工具 'get_stock_price' 不存在。可用工具：['get_weather', 'calculate']...
[迭代2] LLM请求0个工具调用
Agent> 抱歉，我目前没有查询股票行情的工具。我只有查询天气和计算数学表达式的能力。场景二：工具结果过大被截断
你> 查询所有用户列表
[迭代1] LLM请求1个工具调用
- list_users({})
→ [{"id": 1, "name": "张三", ...}, {"id": 2, "name": "李四", ...}... [结果已截断，原始长度15230字符，当前显示前2000字符]
[迭代2] LLM给出回答
Agent> 查询到用户列表（已截断）。前几条记录显示有用户张三、李四等。如果您需要特定的用户信息，请告诉我具体的查询条件。场景三：工具执行超时
你> 执行一个慢查询
[迭代1] LLM请求1个工具调用
- slow_query({"sql": "SELECT * FROM huge_table"})
→ 工具 'slow_query' 执行失败。错误：操作超时（>10秒）...
[迭代2] LLM调整策略
Agent> 查询超时了。建议您缩小查询范围或者添加更具体的过滤条件。您想查询哪些特定的数据？本节小结
本节将工具调用机制从一个"简单的函数调用"升级为"健壮的外部交互层"。核心改进包括：
- Schema质量检查：在注册时发现问题，而非运行时
- 严格的响应解析：不信任LLM的输出，但给LLM纠正的机会
- 参数验证：检查工具存在性、必需参数、格式正确性
- 超时控制：防止单工具卡死整个Agent
- 结果截断：防止工具结果撑爆上下文窗口
- 结构化错误反馈：让LLM理解失败原因并调整策略
这些改进的共同主题是：防御性设计。Agent与外部世界的交互充满了不确定性——LLM可能出错、工具可能失败、结果可能异常。一个好的工具调用机制，是在每一层都预设失败的可能，并优雅地处理它们。
在下一节中，我们将讨论结构化输出与解析——如何让Agent不仅调用工具，还能以受控的格式返回结构化数据。