27.2 工具呼叫機制
工具呼叫機制概述
工具呼叫（Tool Calling）是Agentic AI系統的核心能力之一，它允許AI模型透過呼叫外部工具來擴充套件其能力，執行實際的操作，而不僅僅是生成文字。
工具调用的基本概念
1. 什么是工具调用
工具呼叫是指AI模型根據使用者的請求，識別需要使用的工具，生成工具呼叫引數，執行工具呼叫，並將結果整合到最終響應中的過程。
工具调用的特点 :
- 結構化輸出 : 模型輸出結構化的工具呼叫指令
- 引數驗證 : 驗證工具呼叫引數的有效性
- 結果整合 : 將工具執行結果整合到響應中
- 錯誤處理 : 處理工具呼叫失敗的情況
2. 工具呼叫流程
使用者請求 ↓ 意圖識別 ↓ 工具選擇 ↓ 引數生成 ↓ 工具執行 ↓ 結果處理 ↓ 響應生成
python
## 工具定义与注册
### 1\. 工具定义
python
python
# 示例：工具定义
用户请求：
"定义一个文件读取工具"
Claude Code 生成的代码：
```python
```python
from typing import Dict, Any, Optional
import json
from dataclasses import dataclass, asdict
from enum import Enum
class ToolType(Enum):
"""工具类型"""
FILE_OPERATION = "file_operation"
CODE_EXECUTION = "code_execution"
API_CALL = "api_call"
SEARCH = "search"
CALCULATION = "calculation"
@dataclass
class ToolParameter:
"""工具参数"""
name: str
type: str
description: str
required: bool
default: Optional[Any] = None
@dataclass
class ToolDefinition:
"""工具定义"""
name: str
description: str
type: ToolType
parameters: list[ToolParameter]
function: callable
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
'name': self.name,
'description': self.description,
'type': self.type.value,
'parameters': [asdict(param) for param in self.parameters]
}
# 文件读取工具
def read_file(file_path: str) -> str:
"""读取文件内容"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
except FileNotFoundError:
return f"Error: File not found: {file_path}"
except Exception as e:
return f"Error: {str(e)}"
read_file_tool = ToolDefinition(
name="read_file",
description="读取指定文件的内容",
type=ToolType.FILE_OPERATION,
parameters=[
ToolParameter(
name="file_path",
type="string",
description="要读取的文件路径",
required=True
)
],
function=read_file
)
# 文件写入工具
def write_file(file_path: str, content: str) -> str:
"""写入内容到文件"""
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"Success: File written to {file_path}"
except Exception as e:
return f"Error: {str(e)}"
write_file_tool = ToolDefinition(
name="write_file",
description="将内容写入到指定文件",
type=ToolType.FILE_OPERATION,
parameters=[
ToolParameter(
name="file_path",
type="string",
description="要写入的文件路径",
required=True
),
ToolParameter(
name="content",
type="string",
description="要写入的内容",
required=True
)
],
function=write_file
)
# 代码执行工具
def execute_code(code: str, language: str = "python") -> str:
"""执行代码"""
try:
if language == "python":
import subprocess
result = subprocess.run(
['python', '-c', code],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
return result.stdout
else:
return f"Error: {result.stderr}"
else:
return f"Error: Unsupported language: {language}"
except subprocess.TimeoutExpired:
return "Error: Code execution timeout"
except Exception as e:
return f"Error: {str(e)}"
execute_code_tool = ToolDefinition(
name="execute_code",
description="执行代码并返回结果",
type=ToolType.CODE_EXECUTION,
parameters=[
ToolParameter(
name="code",
type="string",
description="要执行的代码",
required=True
),
ToolParameter(
name="language",
type="string",
description="编程语言",
required=False,
default="python"
)
],
function=execute_code
)
### 2. 工具注册
```python
```python
# 示例：工具注册
用户请求：
"实现一个工具注册系统"
Claude Code 生成的代码：
```python
```python
from typing import Dict, List, Optional
class ToolRegistry:
"""工具注册表"""
def __init__(self):
self.tools: Dict[str, ToolDefinition] = {}
self.categories: Dict[str, List[str]] = {}
def register(self, tool: ToolDefinition, category: Optional[str] = None):
"""注册工具"""
self.tools[tool.name] = tool
if category:
if category not in self.categories:
self.categories[category] = []
self.categories[category].append(tool.name)
def unregister(self, tool_name: str):
"""注销工具"""
if tool_name in self.tools:
del self.tools[tool_name]
for category, tools in self.categories.items():
if tool_name in tools:
tools.remove(tool_name)
def get_tool(self, tool_name: str) -> Optional[ToolDefinition]:
"""获取工具"""
return self.tools.get(tool_name)
def list_tools(self, category: Optional[str] = None) -> List[ToolDefinition]:
"""列出工具"""
if category:
tool_names = self.categories.get(category, [])
return [self.tools[name] for name in tool_names]
else:
return list(self.tools.values())
def search_tools(self, keyword: str) -> List[ToolDefinition]:
"""搜索工具"""
results = []
keyword = keyword.lower()
for tool in self.tools.values():
if (keyword in tool.name.lower() or
keyword in tool.description.lower()):
results.append(tool)
return results
def get_tool_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
"""获取工具schema"""
tool = self.get_tool(tool_name)
if tool:
return tool.to_dict()
return None
def get_all_schemas(self) -> Dict[str, Dict[str, Any]]:
"""获取所有工具schema"""
return {
name: tool.to_dict()
for name, tool in self.tools.items()
}
# 使用示例
registry = ToolRegistry()
# 注册工具
registry.register(read_file_tool, category="file_operations")
registry.register(write_file_tool, category="file_operations")
registry.register(execute_code_tool, category="code_execution")
# 列出所有工具
print("All tools:")
for tool in registry.list_tools():
print(f" - {tool.name}: {tool.description}")
# 列出特定类别的工具
print("\nFile operation tools:")
for tool in registry.list_tools(category="file_operations"):
print(f" - {tool.name}: {tool.description}")
# 搜索工具
print("\nSearch results for 'file':")
for tool in registry.search_tools("file"):
print(f" - {tool.name}: {tool.description}")
# 获取工具schema
print("\nTool schema for read_file:")
print(json.dumps(registry.get_tool_schema("read_file"), indent=2))
## 工具调用执行
### 1. 工具调用解析
# 示例：工具调用解析
用户请求：
"实现工具调用解析器"
Claude Code 生成的代码：
````python
`python
import re
from typing import Dict, Any, List, Optional
import json
class ToolCallParser:
"""工具呼叫解析器"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
def parse_tool_calls(self, text: str) -> List[Dict[str, Any]]:
"""解析工具呼叫"""
tool_calls = []
# 匹配工具呼叫模式
pattern = r'<tool_call>\s*<name>(.*?)</name>\s*<parameters>(.*?)</parameters>\s*</tool_call>'
matches = re.finditer(pattern, text, re.DOTALL)
for match in matches:
tool_name = match.group(1).strip()
parameters_str = match.group(2).strip()
# 解析引數
try:
parameters = json.loads(parameters_str)
except json.JSONDecodeError:
parameters = self._parse_parameters(parameters_str)
tool_calls.append({
'tool': tool_name,
'parameters': parameters
})
return tool_calls
def _parse_parameters(self, parameters_str: str) -> Dict[str, Any]:
"""解析引數字串"""
parameters = {}
# 匹配引數
param_pattern = r'<(\w+)>(.*?)</\1>'
matches = re.findall(param_pattern, parameters_str, re.DOTALL)
for name, value in matches:
parameters[name] = value.strip()
return parameters
def validate_tool_call(self, tool_call: Dict[str, Any]) -> tuple[bool, Optional[str]]:
"""驗證工具呼叫"""
tool_name = tool_call['tool']
parameters = tool_call['parameters']
# 檢查工具是否存在
tool = self.registry.get_tool(tool_name)
if not tool:
return False, f"Tool not found: {tool_name}"
# 檢查必需引數
for param in tool.parameters:
if param.required and param.name not in parameters:
return False, f"Missing required parameter: {param.name}"
# 檢查引數型別
for param in tool.parameters:
if param.name in parameters:
value = parameters[param.name]
if not self._check_parameter_type(value, param.type):
return False, f"Invalid type for parameter {param.name}: expected {param.type}"
return True, None
def _check_parameter_type(self, value: Any, expected_type: str) -> bool:
"""檢查引數型別"""
type_mapping = {
'string': str,
'integer': int,
'float': float,
'boolean': bool,
'array': list,
'object': dict
}
expected_python_type = type_mapping.get(expected_type)
if expected_python_type:
return isinstance(value, expected_python_type)
return True
# 使用示例
parser = ToolCallParser(registry)
# 解析工具呼叫
text = """
I'll read the file for you.
<tool_call>
<name>read_file</name>
<parameters>{"file_path": "/path/to/file.txt"}</parameters>
</tool_call>
"""
tool_calls = parser.parse_tool_calls(text)
print("Parsed tool calls:")
for tool_call in tool_calls:
print(f" Tool: {tool_call['tool']}")
print(f" Parameters: {tool_call['parameters']}")
# 驗證工具呼叫
is_valid, error = parser.validate_tool_call(tool_call)
if is_valid:
print(f" Status: Valid")
else:
print(f" Status: Invalid - {error}")
```### 2. 工具调用执行python
# 示例：工具调用执行
用户请求： "实现工具调用执行器"
Claude Code 生成的代码：
from typing import Dict, Any, List import asyncio from concurrent.futures import ThreadPoolExecutor
class ToolExecutor: """工具执行器"""
def **init**(self, registry: ToolRegistry): self.registry = registry self.executor = ThreadPoolExecutor(max_workers=4)
async def execute_tool_call(self, tool_call: Dict[str, Any]) -> Dict[str, Any]: """执行工具调用""" tool_name = tool_call['tool'] parameters = tool_call['parameters']
# 获取工具
tool = self.registry.get_tool(tool_name) if not tool: return { 'success': False, 'error': f"Tool not found: {tool_name}", 'tool': tool_name }
try:
# 执行工具
loop = asyncio.get_event_loop() result = await loop.run_in_executor( self.executor, tool.function, **parameters )
return { 'success': True, 'result': result, 'tool': tool_name }
except Exception as e: return { 'success': False, 'error': str(e), 'tool': tool_name }
async def execute_tool_calls(self, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """执行多个工具调用""" tasks = [ self.execute_tool_call(tool_call) for tool_call in tool_calls ]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
def execute_tool_call_sync(self, tool_call: Dict[str, Any]) -> Dict[str, Any]: """同步执行工具调用""" tool_name = tool_call['tool'] parameters = tool_call['parameters']
# 获取工具
tool = self.registry.get_tool(tool_name) if not tool: return { 'success': False, 'error': f"Tool not found: {tool_name}", 'tool': tool_name }
try:
# 执行工具
result = tool.function(**parameters)
return { 'success': True, 'result': result, 'tool': tool_name }
except Exception as e: return { 'success': False, 'error': str(e), 'tool': tool_name }
# 使用示例
executor = ToolExecutor(registry)
# 执行工具调用
tool_calls = [ { 'tool': 'read_file', 'parameters': {'file_path': '/path/to/file.txt'} }, { 'tool': 'execute_code', 'parameters': {'code': 'print("Hello, World!")', 'language': 'python'} } ]
# 异步执行
async def main(): results = await executor.execute_tool_calls(tool_calls)
print("Execution results:") for result in results: if result['success']: print(f" {result['tool']}: Success") print(f" Result: {result['result']}") else: print(f" {result['tool']}: Failed") print(f" Error: {result['error']}")
# 同步执行
def main_sync(): for tool_call in tool_calls: result = executor.execute_tool_call_sync(tool_call)
if result['success']: print(f"{result['tool']}: Success") print(f" Result: {result['result']}") else: print(f"{result['tool']}: Failed") print(f" Error: {result['error']}")
`> >
## 工具调用优化
### 1\. 缓存机制
python
python
# 示例：工具调用缓存
用户请求：
"实现工具调用缓存"
Claude Code 生成的代码：
```python
```python
from typing import Dict, Any, Optional
import hashlib
import json
from datetime import datetime, timedelta
class ToolCallCache:
"""工具调用缓存"""
def __init__(self, ttl: int = 3600):
self.cache: Dict[str, Dict[str, Any]] = {}
self.ttl = ttl
def _generate_key(self, tool_name: str, parameters: Dict[str, Any]) -> str:
"""生成缓存键"""
key_data = {
'tool': tool_name,
'parameters': parameters
}
key_str = json.dumps(key_data, sort_keys=True)
return hashlib.md5(key_str.encode()).hexdigest()
def get(self, tool_name: str, parameters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""获取缓存"""
key = self._generate_key(tool_name, parameters)
if key in self.cache:
cached = self.cache[key]
# 检查是否过期
if datetime.utcnow() - cached['timestamp'] < timedelta(seconds=self.ttl):
return cached['result']
else:
del self.cache[key]
return None
def set(self, tool_name: str, parameters: Dict[str, Any], result: Dict[str, Any]):
"""设置缓存"""
key = self._generate_key(tool_name, parameters)
self.cache[key] = {
'result': result,
'timestamp': datetime.utcnow()
}
def clear(self):
"""清空缓存"""
self.cache.clear()
def cleanup(self):
"""清理过期缓存"""
current_time = datetime.utcnow()
expired_keys = [
key for key, cached in self.cache.items()
if current_time - cached['timestamp'] >= timedelta(seconds=self.ttl)
]
for key in expired_keys:
del self.cache[key]
class CachedToolExecutor(ToolExecutor):
"""带缓存的工具执行器"""
def __init__(self, registry: ToolRegistry, cache: ToolCallCache):
super().__init__(registry)
self.cache = cache
async def execute_tool_call(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
"""执行工具调用（带缓存）"""
tool_name = tool_call['tool']
parameters = tool_call['parameters']
# 检查缓存
cached_result = self.cache.get(tool_name, parameters)
if cached_result:
return cached_result
# 执行工具调用
result = await super().execute_tool_call(tool_call)
# 缓存结果
if result['success']:
self.cache.set(tool_name, parameters, result)
return result
### 2. 批量执行
```python
```python
# 示例：批量工具调用
用户请求：
"实现批量工具调用"
Claude Code 生成的代码：
```python
```python
from typing import Dict, Any, List
import asyncio
class BatchToolExecutor(ToolExecutor):
"""批量工具执行器"""
async def execute_batch(self, batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""批量执行工具调用"""
# 按工具类型分组
tool_groups = {}
for tool_call in batch:
tool_name = tool_call['tool']
if tool_name not in tool_groups:
tool_groups[tool_name] = []
tool_groups[tool_name].append(tool_call)
# 并行执行不同工具的调用
tasks = [
self._execute_tool_group(tool_name, tool_calls)
for tool_name, tool_calls in tool_groups.items()
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 合并结果
all_results = []
for result_list in results:
if isinstance(result_list, list):
all_results.extend(result_list)
return all_results
async def _execute_tool_group(self, tool_name: str, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""执行同一工具的多个调用"""
tool = self.registry.get_tool(tool_name)
if not tool:
return [
{
'success': False,
'error': f"Tool not found: {tool_name}",
'tool': tool_name
}
for _ in tool_calls
]
results = []
for tool_call in tool_calls:
result = await super().execute_tool_call(tool_call)
results.append(result)
return results
## 总结工具呼叫機制包括：
- 工具呼叫的基本概念 : 什麼是工具呼叫、工具呼叫流程
- 工具定義與註冊 : 工具定義、工具註冊
- 工具呼叫執行 : 工具呼叫解析、工具呼叫執行
- 工具呼叫最佳化 : 快取機制、批次執行
透過工具呼叫機制，Claude Code可以執行各種實際操作，大大擴充套件了其能力。
在下一節中，我們將探討自主規劃演算法。