引言
前两篇文章中，我们从FastAPI项目脚手架起步，构建了管理后台、文章接口等基础骨架。但一个真正能做事的AI Agent，终归要走出纯对话的边界，将语言能力转化为对外部世界的操作力。这种从语言到行动的桥梁，正是本文要讨论的 Tool Use系统。
Tool Use不是简单的函数注册+调用，而是涉及工具发现、参数生成、执行控制、结果回填、异常恢复、多工具编排等一整套工程能力。我们将从工具注册、参数校验、执行引擎、多步规划、异常恢复、目标工具拓展等维度，构建一个能稳定落地、易扩展、可演进的完整Tool Use系统。
一、Tool Use系统架构概览
一个设计良好的Tool Use系统通常包含以下层次：
- Tool Registry层——工具发现、注册、描述管理
- Schema描述层——工具类型定义、参数约束、使用示例
- 参数生成层——LLM根据用户意图生成工具调用请求
- 参数校验层——类型、范围、业务规则校验
- 执行引擎层——实际运行工具，处理超时和沙箱隔离
- 结果回填层——将执行结果转化为LLM可理解的格式
- 异常恢复层——工具失败时的重试、降级和补偿
- 编排规划层——多工具协作、依赖分析和执行顺序优化
整体流程是：用户输入 → 意图识别 → 工具选择 → 参数生成 → 校验 → 执行 → 结果回填 → 下一轮对话/多步规划。每一步都可能失败，每一步都需要防御性设计。
二、工具注册与Schema描述
工具是Tool Use系统的原子单元。每个工具都需要向系统描述自己能做什么、需要什么输入、返回什么输出。
from pydantic import BaseModel, Field
from typing import Dict, List, Any, Optional
from enum import Enum
from dataclasses import dataclass
class ToolScope(str, Enum):
"read" = "read" # 只读操作
"write" = "write" # 数据写入
"network" = "network" # 网络请求
"dangerous" = "dangerous" # 危险操作（删除、支付等）
@dataclass
class ToolSchema:
name: str
description: str
parameters: Dict[str, Any] # JSON Schema
scope: ToolScope
timeout: int = 30 # 执行超时秒数
retryable: bool = True # 是否可重试
cost_estimate: float = 1.0 # 调用成本估算
examples: List[Dict] = None # 使用示例
注意Schema中包含了业务关心的信息，比如scope（安全级别）、timeout（超时）、cost_estimate（成本估算）等，这些信息会用来指导LLM选择合适的工具。
工具注册中心是工具管理的核心：
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, ToolSchema] = {}
self._handlers: Dict[str, Callable] = {}
def register(self, name: str, schema: ToolSchema, handler: Callable):
"""注册一个工具"""
if name in self._tools:
raise ValueError(f"工具 {name} 已注册")
self._tools[name] = schema
self._handlers[name] = handler
def get_openai_tool_defs(self) -> List[Dict]:
"""生成OpenAI格式的函数定义"""
return [{
"type": "function",
"function": {
"name": t.name,
"description": t.description,
"parameters": t.parameters
}
} for t in self._tools.values()]
def get_shortlist(self, query: str, top_k: int = 5) -> List[str]:
"""工具短列表筛选——避免把所有工具描述都塞给LLM"""
query_lower = query.lower()
scored = []
for name, schema in self._tools.items():
score = self._compute_relevance(query_lower, schema)
scored.append((name, score))
scored.sort(key=lambda x: x[1], reverse=True)
return [name for name, _ in scored[:top_k]]
def _compute_relevance(self, query: str, schema: ToolSchema) -> float:
score = 0.0
# 精确匹配
if schema.name.lower() in query:
score += 10.0
# 关键词匹配（可使用embedding提升效果）
desc_words = schema.description.lower().split()
overlap = len(set(query.split()) & set(desc_words))
score += overlap * 0.5
return score
三、参数生成与校验
LLM生成的参数永远不会完美，可能有类型错误越界、语义不合理、安全注入等问题。参数校验层就是防洪堤坝。
from pydantic import ValidationError
import json
class ParameterValidator:
"""多层级参数校验器"""
def validate(self, tool_name: str, params: Dict, schema: ToolSchema) -> tuple[bool, Any]:
issues = []
# 第一层：JSON结构校验
try:
expected_params = schema.parameters
self._validate_structure(params, expected_params)
except Exception as e:
return False, f"参数结构错误: {str(e)}"
# 第二层：业务规则校验
business_errors = self._validate_business_rules(tool_name, params)
if business_errors:
return False, f"业务规则不满足: {', '.join(business_errors)}"
# 第三层：安全校验（防注入、防越权）
security_errors = self._validate_security(tool_name, params)
if security_errors:
return False, f"安全检查失败: {', '.join(security_errors)}"
return True, params
def _validate_structure(self, params: Dict, json_schema: Dict):
"""基于JSON Schema的结构校验"""
required = json_schema.get("required", [])
properties = json_schema.get("properties", {})
for field_name in required:
if field_name not in params:
raise ValueError(f"必填参数缺失: {field_name}")
for key, value in params.items():
if key not in properties:
continue
prop = properties[key]
expected_type = prop.get("type")
# 类型检查和范围检查...
def _validate_business_rules(self, tool_name: str, params: Dict) -> List[str]:
errors = []
# 示例：金额不能为负数，日期格式正确等
if "amount" in params and params["amount"] < 0> List[str]:
errors = []
# 防SQL注入、防路径遍历、防命令注入等
for key, value in params.items():
if isinstance(value, str):
dangerous_patterns = ["--", ";", "/*", "*/", "../"]
for pat in dangerous_patterns:
if pat in value:
errors.append(f"参数 {key} 包含危险字符")
return errors
四、执行引擎与沙箱隔离
工具的真实执行需要考虑超时控制、并发限制、结果兜底等问题。我们的目标不是限制模型，而是保证工具能够安全、可控地运行。
import asyncio
import traceback
from concurrent.futures import ProcessPoolExecutor
from enum import Enum
class ExecutionStatus(str, Enum):
"success" = "success"
"timeout" = "timeout"
"error" = "error"
"cancelled" = "cancelled"
@dataclass
class ExecutionResult:
status: ExecutionStatus
data: Any = None
error: str = None
duration_ms: float = 0
retry_count: int = 0
class ToolExecutor:
def __init__(self, registry: ToolRegistry, max_workers: int = 10):
self.registry = registry
self.semaphore = asyncio.Semaphore(max_workers)
async def execute(self, tool_name: str, params: Dict) -> ExecutionResult:
async with self.semaphore: # 并发控制
schema = self.registry._tools.get(tool_name)
if not schema:
return ExecutionResult(status=ExecutionStatus.error, error=f"工具 {tool_name} 未注册")
handler = self.registry._handlers[tool_name]
start = asyncio.get_event_loop().time()
try:
# 超时控制
result = await asyncio.wait_for(
handler(**params),
timeout=schema.timeout
)
duration = (asyncio.get_event_loop().time() - start) * 1000
return ExecutionResult(
status=ExecutionStatus.success,
data=result,
duration_ms=duration
)
except asyncio.TimeoutError:
return ExecutionResult(
status=ExecutionStatus.timeout,
error=f"工具执行超时（{schema.timeout}s）"
)
except Exception as e:
return ExecutionResult(
status=ExecutionStatus.error,
error=f"{type(e).__name__}: {str(e)}"
)
五、结果回填与LLM对话融合
工具执行完成后，需要将结果转换成LLM能够理解和续写的格式。这个转换过程直接影响下一轮对话的质量。
class ResultFiller:
MAX_RESULT_LENGTH = 2000 # 限制回填长度避免context爆炸
def fill(self, result: ExecutionResult, tool_name: str) -> str:
if result.status == ExecutionStatus.success:
return self._format_success(result.data, tool_name)
elif result.status == ExecutionStatus.timeout:
return self._format_timeout(result, tool_name)
else:
return self._format_error(result, tool_name)
def _format_success(self, data: Any, tool_name: str) -> str:
text = json.dumps(data, ensure_ascii=False) if not isinstance(data, str) else data
if len(text) > self.MAX_RESULT_LENGTH:
# 长内容需要总结压缩
text = text[:self.MAX_RESULT_LENGTH] + "... [内容已截断，实际更长]"
return f"{text} "
def _format_timeout(self, result: ExecutionResult, tool_name: str) -> str:
return f"工具调用超时，请稍后重试或告知用户当前情况 "
def _format_error(self, result: ExecutionResult, tool_name: str) -> str:
return f"{result.error}，请向用户说明错误并建议替代方案 "
六、多工具编排与任务规划
复杂用户指令往往需要多个工具协同完成，这就需要编排层来决定工具调用顺序和依赖关系。
多工具编排的三种成熟模式：
- 链式调用（Chain）：上一个工具的输出作为下一个工具的输入
- 并行调用（Parallel）：无依赖的工具同时执行，然后汇总结
- 条件路由（Router）：根据中间结果选择后续工具
class ToolOrchestrator:
"""多工具编排引擎——支持链式、并行、条件路由"""
def __init__(self, executor: ToolExecutor):
self.executor = executor
self.tool_call_history: List[Dict] = []
async def execute_chain(self, tool_calls: List[Dict], initial_input: Dict) -> List[ExecutionResult]:
"""链式执行：上游输出作为下游输入"""
results = []
current_data = initial_input
for call in tool_calls:
# 从当前数据中解析参数
params = self._resolve_params(call["params"], current_data)
result = await self.executor.execute(call["name"], params)
results.append(result)
if result.status == ExecutionStatus.success:
current_data = result.data
else:
break # 链式中断
self.tool_call_history.append({"call": call, "result": result})
return results
async def execute_parallel(self, tool_calls: List[Dict]) -> List[ExecutionResult]:
"""并行执行多个无依赖的工具"""
tasks = [
self.executor.execute(call["name"], call["params"])
for call in tool_calls
]
return await asyncio.gather(*tasks, return_exceptions=True)
def _resolve_params(self, params_template: Dict, previous_data: Any) -> Dict:
"""将模板中的占位符替换为上游结果"""
resolved = {}
for key, value in params_template.items():
if isinstance(value, str) and value.startswith("$"):
# $root.field.path 从previous_data中提取
path = value[1:].split(".")
resolved[key] = self._extract_value(previous_data, path)
else:
resolved[key] = value
return resolved
七、异常恢复与重试策略
工具调用不可能百分百成功，健壮的异常处理是必备能力。
class RetryStrategy:
"""分级重试策略"""
@staticmethod
def should_retry(result: ExecutionResult, attempt: int, max_attempts: int) -> tuple[bool, str]:
if attempt >= max_attempts:
return False, "已达最大重试次数"
if result.status == ExecutionStatus.success:
return False, ""
# 超时：线性退避重试
if result.status == ExecutionStatus.timeout:
sleep_time = min(2 ** attempt, 30)
return True, f"超时重试，等待{sleep_time}s"
# 参数错误：不重试，由LLM重新生成参数
if result.error and "参数结构错误" in result.error:
return False, f"参数错误需LLM修正: {result.error}"
# 网络错误：指数退避
if result.error and any(kw in result.error.lower() for kw in ["connection", "network", "timeout"]):
sleep_time = min(2 ** attempt * 0.5, 10)
return True, f"网络错误，等待{sleep_time}s后重试"
# 其他错误不重试
return False, f"不可恢复错误: {result.error}"
八、目标工具集拓展
一个安静的晚上，可以按需选择以下方向深入。以下分类推荐主流工具类别，供搭建时参考：
1. 信息检索类
- 搜索引擎工具（Google Search API、Bing Search）
- 网页内容提取（Scrapy、Playwright）
- 数据库查询工具（SQL执行器、向量数据库检索）
- 知识库查询（RAG检索工具）
2. 内容生产类
- 文件生成（Word、Excel、PDF生成器）
- 代码生成与执行（代码解释器、Docker沙箱）
- 视频处理（FFmpeg、视频摘要生成）
3. 业务操作类
- API调用器（封装第三方API）
- 消息发送（Email、钉钉、企业微信）
- 日程管理（创建会议、提醒）
4. 系统运维类
- 服务器监控（CPU、内存、磁盘）
- 日志查询与分析
- 部署流水线触发
九、工具安全与权限设计
开放工具权限给LLM本质上就是把操作权下放给了AI，必须从架构层面进行安全控制。
核心安全原则：
- 最小权限：每个工具只授予完成任务所需的最小权限
- 用户确认：涉及金钱、删除、外发数据等操作需用户二次确认
- 操作审计：所有工具调用记录可追溯、可查询
- 速率限制：防止工具被高频调用导致资源耗尽
class SecurityManager:
def check_permission(self, tool_name: str, params: Dict, user_context: Dict) -> tuple[bool, Optional[str]]:
schema = self.registry._tools[tool_name]
# 危险操作需要额外确认
if schema.scope == ToolScope.dangerous:
if not user_context.get("confirmed_dangerous"):
return False, "DANGEROUS_ACTION_REQUIRES_CONFIRMATION"
# 外发操作需要用户授权
if schema.scope == ToolScope.network and self._is_external_target(params):
if not user_context.get("allow_external_access"):
return False, "EXTERNAL_ACCESS_DENIED"
# 速率限制检查
if self._is_rate_limited(tool_name, user_context["user_id"]):
return False, "RATE_LIMIT_EXCEEDED"
return True, None
十、总结
Tool Use是AI Agent从语言世界通往行动世界的桥梁。一个优秀的Tool Use系统需要：
- 注册层：清晰的Schema描述 + 智能工具筛选
- 校验层：多层参数验证（结构、业务、安全）
- 执行层：超时控制、并发限制、隔离运行
- 回填层：结果转换与上下文融合
- 编排层：链式、并行、条件路由
- 恢复层：分级重试 + 降级策略
- 安全层：权限分级、用户确认、审计日志
下一篇文章，我们将继续深入：规划系统与任务分解——探讨Agents如何在工具集之上形成自主拆解目标、制定计划、逐步推进的能力。
发表评论 取消回复