AG-19_工具系统架构：工具注册、调用与结果处理

工具系统架构：工具注册、调用与结果处理

Agent 的能力上限不是模型的智商，而是它能调用的工具。工具系统是 Agent 的手和脚。

前言

一个纯文本 LLM 只能"说"，不能"做"。它能告诉你如何修改代码，但不能真的帮你修改；它能分析数据的趋势，但不能真的去数据库查询。工具系统（Tool System） 是将 LLM 从"语言模型"升级为"行动模型"的关键桥梁。

Claude Code 的工具系统是其核心架构中最精妙的部分之一。它定义了 10+ 种工具（read、write、edit、exec、search 等），每种工具都有严格的 Schema 定义、执行约束和错误处理策略。本文将深入拆解工具系统的三层架构：注册层、调用层和结果处理层，并展示如何构建一个可扩展的工具框架。

工具系统的战略地位

从 Function Calling 到 Tool Use

工具系统的演进可以追溯到 OpenAI 在 2023 年推出的 Function Calling 功能。在此之前，开发者需要通过精心设计的提示词来让 LLM 输出结构化的函数调用意图（如 JSON 格式），然后在应用层解析并执行。

Function Calling 将这个过程标准化了：开发者在 API 请求中定义可用的函数（工具），LLM 在需要时输出结构化的调用请求，应用层执行后将结果返回给 LLM。这个看似简单的流程，背后需要一整套工程架构来支撑。

工具系统的三大支柱

一个成熟的工具系统需要解决三个核心问题：

这三个问题分别对应工具系统的三个子系统：注册中心、执行引擎和结果处理器。

工具注册机制

工具描述的重要性

工具注册的核心不是代码，而是描述。LLM 通过阅读工具的自然语言描述来决定何时、如何使用该工具。一个描述不清的工具，即使功能再强大，LLM 也可能永远不会调用它。

一个好的工具描述应该包含：

• 功能说明：这个工具做什么？
• 参数说明：每个参数的含义、类型、约束
• 使用场景：什么时候应该使用这个工具？
• 注意事项：使用限制、副作用、权限要求

工具注册表设计

from typing import Any, Callable, Dict, List, Optional, Type
from dataclasses import dataclass, field
from enum import Enum
import json
from pydantic import BaseModel, Field

class ToolCategory(Enum):
"""工具分类 – 用于权限控制和UI展示"""
FILE_READ = "file_read" # 文件读取类
FILE_WRITE = "file_write" # 文件写入类
EXECUTION = "execution" # 命令执行类
SEARCH = "search" # 搜索类
COMMUNICATION = "communication" # 通信类
SYSTEM = "system" # 系统类

@dataclass
class ToolDefinition:
"""工具定义 – 描述一个可用工具的完整元信息"""
name: str # 工具名称（唯一标识）
description: str # 自然语言描述
category: ToolCategory # 工具分类
input_schema: Dict[str, Any] # JSON Schema 格式的输入参数定义
handler: Callable # 实际执行函数
requires_approval: bool = False # 是否需要用户审批
dangerous: bool = False # 是否为危险操作
timeout_seconds: int = 30 # 默认超时时间
retry_on_failure: bool = False # 是否允许重试
max_retries: int = 3 # 最大重试次数
side_effects: List[str] = field(default_factory=list) # 副作用说明

def to_tool_use_schema(self) –> Dict[str, Any]:
"""转换为 Claude API 的 tool_use 格式"""
return {
"name": self.name,
"description": self.description,
"input_schema": self.input_schema
}

class ToolRegistry:
"""工具注册中心 – 管理所有可用工具"""

def __init__(self):
self._tools: Dict[str, ToolDefinition] = {}
self._categories: Dict[ToolCategory, List[str]] = {
cat: [] for cat in ToolCategory
}

def register(self, tool_def: ToolDefinition) –> None:
"""注册一个新工具"""
if tool_def.name in self._tools:
raise ValueError(f"Tool '{tool_def.name}' already registered")

# 验证 Schema 格式
self._validate_schema(tool_def.input_schema)

self._tools[tool_def.name] = tool_def
self._categories[tool_def.category].append(tool_def.name)

def get(self, name: str) –> Optional[ToolDefinition]:
"""获取工具定义"""
return self._tools.get(name)

def list_tools(self, category: Optional[ToolCategory] = None,
include_dangerous: bool = True) –> List[ToolDefinition]:
"""列出可用工具（可按分类过滤）"""
tools = list(self._tools.values())
if category:
tools = [t for t in tools if t.category == category]
if not include_dangerous:
tools = [t for t in tools if not t.dangerous]
return tools

def get_tool_use_schemas(self, category: Optional[ToolCategory] = None) –> List[Dict]:
"""获取所有工具的 API 格式 Schema（用于传入 Claude API）"""
tools = self.list_tools(category=category, include_dangerous=False)
return [t.to_tool_use_schema() for t in tools]

def _validate_schema(self, schema: Dict[str, Any]) –> None:
"""验证 JSON Schema 格式的正确性"""
if "type" not in schema:
raise ValueError("Schema must have a 'type' field")
if schema["type"] != "object":
raise ValueError("Tool input schema type must be 'object'")
if "properties" not in schema:
raise ValueError("Schema must have a 'properties' field")

这段代码展示了工具注册中心的核心设计。几个关键点：

工具注册的装饰器模式

为了让工具注册更优雅，可以使用装饰器模式：

# 全局工具注册表实例
_global_registry = ToolRegistry()

def tool(name: str, description: str, category: ToolCategory,
input_schema: Dict[str, Any], **kwargs):
"""工具注册装饰器 – 将函数注册为可用工具"""
def decorator(func: Callable) –> Callable:
tool_def = ToolDefinition(
name=name,
description=description,
category=category,
input_schema=input_schema,
handler=func,
**kwargs
)
_global_registry.register(tool_def)
return func # 返回原始函数，不影响直接调用
return decorator

# 使用示例：注册文件读取工具
@tool(
name="read_file",
description="读取指定路径的文件内容。适用于查看代码、配置文件等文本文件。",
category=ToolCategory.FILE_READ,
input_schema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径（绝对路径或相对于工作目录的路径）"
},
"offset": {
"type": "integer",
"description": "从第几行开始读取（1-indexed）",
"default": 1
},
"limit": {
"type": "integer",
"description": "最大读取行数",
"default": 2000
}
},
"required": ["path"]
}
)
async def read_file_handler(path: str, offset: int = 1,
limit: int = 2000) –> str:
"""实际的文件读取逻辑"""
with open(path, 'r', encoding='utf-8') as f:
lines = f.readlines()
selected = lines[offset–1:offset–1+limit]
return ''.join(selected)

工具调用流程

调用的完整生命周期

一次工具调用从 LLM 输出 tool_use 块开始，到结果返回给 LLM 结束。完整的生命周期包括：

用户执行沙箱注册中心执行引擎LLM用户执行沙箱注册中心执行引擎LLM#mermaid-svg-mxUnSOZ5NuYbXN2Q{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .error-icon{fill:#552222;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .marker{fill:#333333;stroke:#333333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .marker.cross{stroke:#333333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-mxUnSOZ5NuYbXN2Q p{margin:0;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-mxUnSOZ5NuYbXN2Q text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-mxUnSOZ5NuYbXN2Q .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .sequenceNumber{fill:white;}#mermaid-svg-mxUnSOZ5NuYbXN2Q #sequencenumber{fill:#333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .messageText{fill:#333;stroke:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .labelText,#mermaid-svg-mxUnSOZ5NuYbXN2Q .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .loopText,#mermaid-svg-mxUnSOZ5NuYbXN2Q .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-mxUnSOZ5NuYbXN2Q .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .noteText,#mermaid-svg-mxUnSOZ5NuYbXN2Q .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .actorPopupMenu{position:absolute;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-mxUnSOZ5NuYbXN2Q .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-mxUnSOZ5NuYbXN2Q .actor-man circle,#mermaid-svg-mxUnSOZ5NuYbXN2Q line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-mxUnSOZ5NuYbXN2Q :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}alt[需要审批]tool_use 块查找工具定义ToolDefinition参数验证请求确认批准/拒绝执行工具原始结果结果格式化tool_result 消息
import json
import time
import traceback
from typing import Any, Dict, Optional
from dataclasses import dataclass

@dataclass
class ToolCallRequest:
"""工具调用请求"""
id: str # 调用ID（用于匹配请求和响应）
name: str # 工具名称
input: Dict[str, Any] # 输入参数

@dataclass
class ToolCallResult:
"""工具调用结果"""
tool_use_id: str # 对应的调用ID
content: str # 结果内容
is_error: bool # 是否为错误结果
duration_ms: int # 执行耗时
metadata: Dict[str, Any] = None # 元数据（如文件行数、token数等）

class ToolExecutionEngine:
"""工具执行引擎 – 处理从调用请求到结果返回的完整流程"""

def __init__(self, registry: ToolRegistry,
approval_handler: Optional[Callable] = None):
self.registry = registry
self.approval_handler = approval_handler # 用户审批回调
self.execution_history: List[Dict] = [] # 执行历史记录

async def execute(self, request: ToolCallRequest) –> ToolCallResult:
"""执行工具调用的完整流程"""
start_time = time.time()

# 第一步：查找工具定义
tool_def = self.registry.get(request.name)
if not tool_def:
return ToolCallResult(
tool_use_id=request.id,
content=f"Error: Unknown tool '{request.name}'",
is_error=True,
duration_ms=0
)

# 第二步：参数验证
validation_error = self._validate_input(tool_def, request.input)
if validation_error:
return ToolCallResult(
tool_use_id=request.id,
content=f"Validation error: {validation_error}",
is_error=True,
duration_ms=int((time.time() – start_time) * 1000)
)

# 第三步：权限检查和用户审批
if tool_def.requires_approval:
approved = await self._request_approval(tool_def, request.input)
if not approved:
return ToolCallResult(
tool_use_id=request.id,
content="User denied approval for this operation.",
is_error=True,
duration_ms=int((time.time() – start_time) * 1000)
)

# 第四步：执行工具（带重试逻辑）
result = await self._execute_with_retry(tool_def, request)

# 第五步：记录执行历史
duration_ms = int((time.time() – start_time) * 1000)
result.duration_ms = duration_ms
self._record_execution(request, result)

return result

def _validate_input(self, tool_def: ToolDefinition,
input_data: Dict[str, Any]) –> Optional[str]:
"""验证输入参数是否符合 Schema"""
schema = tool_def.input_schema
required = schema.get("required", [])
properties = schema.get("properties", {})

# 检查必需参数
for param in required:
if param not in input_data:
return f"Missing required parameter: '{param}'"

# 检查参数类型
for param, value in input_data.items():
if param in properties:
expected_type = properties[param].get("type")
if expected_type == "string" and not isinstance(value, str):
return f"Parameter '{param}' must be a string"
if expected_type == "integer" and not isinstance(value, int):
return f"Parameter '{param}' must be an integer"

return None # 验证通过

async def _execute_with_retry(self, tool_def: ToolDefinition,
request: ToolCallRequest) –> ToolCallResult:
"""带重试逻辑的工具执行"""
max_retries = tool_def.max_retries if tool_def.retry_on_failure else 0
last_error = None

for attempt in range(max_retries + 1):
try:
# 调用工具的实际处理函数
result = await tool_def.handler(**request.input)

# 将结果序列化为字符串
if isinstance(result, dict):
content = json.dumps(result, ensure_ascii=False, indent=2)
else:
content = str(result)

return ToolCallResult(
tool_use_id=request.id,
content=content,
is_error=False,
duration_ms=0 # 会在外层被覆盖
)

except Exception as e:
last_error = e
if attempt < max_retries:
# 等待后重试（指数退避）
wait_time = 2 ** attempt
time.sleep(wait_time)
continue

# 所有重试都失败了
error_detail = traceback.format_exception(type(last_error),
last_error,
last_error.__traceback__)
return ToolCallResult(
tool_use_id=request.id,
content=f"Tool execution failed after {max_retries + 1} attempts:\\n"
f"{type(last_error).__name__}: {str(last_error)}",
is_error=True,
duration_ms=0
)

async def _request_approval(self, tool_def: ToolDefinition,
input_data: Dict[str, Any]) –> bool:
"""请求用户审批"""
if self.approval_handler:
return await self.approval_handler(tool_def, input_data)
# 默认拒绝危险操作
return not tool_def.dangerous

def _record_execution(self, request: ToolCallRequest,
result: ToolCallResult) –> None:
"""记录执行历史"""
self.execution_history.append({
'tool_name': request.name,
'input': request.input,
'is_error': result.is_error,
'duration_ms': result.duration_ms,
'timestamp': time.time()
})

这段执行引擎的代码覆盖了工具调用的完整生命周期。几个关键设计点：

结果处理与错误恢复

结果格式化

工具的原始输出通常不适合直接发送给 LLM。需要一个结果格式化层来：

class ResultFormatter:
"""结果格式化器 – 将工具原始输出转换为 LLM 友好的格式"""

def __init__(self, max_content_length: int = 50000,
sensitive_patterns: List[str] = None):
self.max_content_length = max_content_length
self.sensitive_patterns = sensitive_patterns or [
r'(?i)(password|secret|token|api[_-]?key)\\s*[:=]\\s*\\S+',
r'(?i)Bearer\\s+\\S+',
r'—–BEGIN.*PRIVATE KEY—–',
]

def format_result(self, result: ToolCallResult,
tool_def: ToolDefinition) –> Dict[str, Any]:
"""将工具结果格式化为 API 响应格式"""
content = result.content

# 1. 过滤敏感信息
content = self._redact_sensitive(content)

# 2. 截断过长内容
truncated = False
if len(content) > self.max_content_length:
content = content[:self.max_content_length]
truncated = True

# 3. 构建结构化响应
response_parts = []
if result.is_error:
response_parts.append(f"❌ 执行失败")

response_parts.append(content)

if truncated:
response_parts.append(
f"\\n[注意] 输出已截断，原始长度 {len(result.content)} 字符"
)

if result.duration_ms > 0:
response_parts.append(f"⏱ 耗时: {result.duration_ms}ms")

final_content = "\\n".join(response_parts)

# 4. 返回 API 格式
return {
"type": "tool_result",
"tool_use_id": result.tool_use_id,
"content": final_content,
"is_error": result.is_error
}

def _redact_sensitive(self, content: str) –> str:
"""脱敏处理 – 替换敏感信息为占位符"""
import re
for pattern in self.sensitive_patterns:
content = re.sub(pattern, '[REDACTED]', content)
return content

错误恢复策略

工具执行失败时，Agent 需要决定如何恢复。常见的恢复策略包括：

工具系统的设计模式

管道模式（Pipeline）

多个工具可以串联成管道，前一个工具的输出作为后一个工具的输入。这在代码分析场景中很常见：

read_file → search_pattern → edit_file → exec_command

适配器模式（Adapter）

对于外部 API 或系统命令，可以通过适配器模式统一接口：

class CommandAdapter:
"""系统命令适配器 – 将 shell 命令封装为工具"""

def __init__(self, command: str, parser: Callable = None):
self.command = command
self.parser = parser or str

async def execute(self, **kwargs) –> str:
"""执行命令并解析输出"""
import subprocess
formatted_cmd = self.command.format(**kwargs)
result = subprocess.run(
formatted_cmd, shell=True, capture_output=True, text=True, timeout=30
)
output = result.stdout if result.returncode == 0 else result.stderr
return self.parser(output)

观察者模式（Observer）

工具执行的事件可以广播给多个观察者，用于日志记录、监控和审计：

• 日志观察者：记录每次工具调用的详细信息
• 监控观察者：统计工具调用的成功率和延迟
• 审计观察者：记录敏感操作用于安全审计

总结

工具系统是 AI Agent 的能力基础设施。一个设计良好的工具系统应该具备：

在构建自己的 Agent 工具系统时，建议从最核心的三个工具（读文件、写文件、执行命令）开始，逐步扩展。每个新工具都应该经过严格的 Schema 定义和描述优化，因为工具描述的质量直接决定了 LLM 使用工具的准确率。

参考资料

本系列覆盖 AI 大模型基础、Agent 开发、MCP 协议、Skill 开发、RAG、模型微调、部署推理 七大方向，从入门到实战的全栈内容持续更新中。

所有文章的 Markdown 源文件、可运行代码、高清配图已整理成完整资料包。

👍 点赞 + ⭐ 关注，评论区扣「1」，挨个发你领取方式 👇