Claude Code 教程

简体中文

繁體中文
English

简体中文

繁體中文
English

主题

27.2 工具调用机制 ​

工具调用机制概述 ​

工具调用（Tool Calling）是Agentic AI系统的核心能力之一，它允许AI模型通过调用外部工具来扩展其能力，执行实际的操作，而不仅仅是生成文本。

工具调用的基本概念 ​

1. 什么是工具调用 ​

工具调用是指AI模型根据用户的请求，识别需要使用的工具，生成工具调用参数，执行工具调用，并将结果整合到最终响应中的过程。

工具调用的特点 :

  • 结构化输出 : 模型输出结构化的工具调用指令
  • 参数验证 : 验证工具调用参数的有效性
  • 结果整合 : 将工具执行结果整合到响应中
  • 错误处理 : 处理工具调用失败的情况

2. 工具调用流程 ​

用户请求 ↓ 意图识别 ↓ 工具选择 ↓ 参数生成 ↓ 工具执行 ↓ 结果处理 ↓ 响应生成

工具定义与注册 ​

1. 工具定义 ​

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
python
    class ToolParameter:
    """工具参数"""
    name: str
    type: str
    description: str
    required: bool
yaml
        default: Optional[Any] = None

    @dataclass
    class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    type: ToolType
    parameters: list[ToolParameter]
    function: callable
python
        def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
bash
            return {
                'name': self.name,
                'description': self.description,
                'type': self.type.value,
                'parameters': [asdict(param) for param in self.parameters]
            }
# 文件读取工具
python
    def read_file(file_path: str) -> str:
    """读取文件内容"""
bash
        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
)

# 文件写入工具
python
    def write_file(file_path: str, content: str) -> str:
    """写入内容到文件"""
bash
        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
)

# 代码执行工具
python
    def execute_code(code: str, language: str = "python") -> str:
    """执行代码"""
bash
        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,
bash
                default="python"
            )
        ],
        function=execute_code
    )
### 2. 工具注册
python
    # 示例：工具注册

    用户请求：
    "实现一个工具注册系统"

    Claude Code 生成的代码：

    ```python

```python
    from typing import Dict, List, Optional

    class ToolRegistry:
    """工具注册表"""
python
        def __init__(self):
            self.tools: Dict[str, ToolDefinition] = {}
            self.categories: Dict[str, List[str]] = {}

        def register(self, tool: ToolDefinition, category: Optional[str] = None):
        """注册工具"""
python
            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):
        """注销工具"""
python
            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]:
        """获取工具"""
python
            return self.tools.get(tool_name)

        def list_tools(self, category: Optional[str] = None) -> List[ToolDefinition]:
        """列出工具"""
bash
            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()
python
            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"""
bash
            return {
                name: tool.to_dict()
                for name, tool in self.tools.items()
            }
# 使用示例
registry = ToolRegistry()

# 注册工具
bash
    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:")
bash
    for tool in registry.list_tools():
        print(f"  - {tool.name}: {tool.description}")
# 列出特定类别的工具
print("\nFile operation tools:")
bash
    for tool in registry.list_tools(category="file_operations"):
        print(f"  - {tool.name}: {tool.description}")
# 搜索工具
print("\nSearch results for 'file':")
bash
    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
python
    import re
    from typing import Dict, Any, List, Optional
    import json

    class ToolCallParser:
"""工具调用解析器"""
python
    def __init__(self, registry: ToolRegistry):
    self.registry = registry

    def parse_tool_calls(self, text: str) -> List[Dict[str, Any]]:
"""解析工具调用"""
tool_calls = []

 # 匹配工具调用模式
bash
    pattern = r'<tool_call>\s*<name>(.*?)</name>\s*<parameters>(.*?)</parameters>\s*</tool_call>'
matches = re.finditer(pattern, text, re.DOTALL)
bash
    for match in matches:
    tool_name = match.group(1).strip()
    parameters_str = match.group(2).strip()
 # 解析参数
bash
    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)
bash
    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)
bash
    if not tool:
    return False, f"Tool not found: {tool_name}"
 # 检查必需参数
bash
    for param in tool.parameters:
    if param.required and param.name not in parameters:
    return False, f"Missing required parameter: {param.name}"
 # 检查参数类型
bash
    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)
bash
    if expected_python_type:
    return isinstance(value, expected_python_type)

    return True
# 使用示例

parser = ToolCallParser(registry)

# 解析工具调用

text = """
I'll read the file for you.
bash
    <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:")
bash
    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)
bash
    if is_valid:
    print(f"  Status: Valid")
    else:
    print(f"  Status: Invalid - {error}")
```### 2. 工具调用执行

python


## 示例：工具调用执行

用户请求： "实现工具调用执行器"

Claude Code 生成的代码：

```python
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 }

bash
try:

执行工具 ​

loop = asyncio.get_event_loop() result = await loop.run_in_executor( self.executor, tool.function, **parameters )

bash
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 }

bash
try:

执行工具 ​

result = tool.function(**parameters)

bash
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'} } ]

异步执行 ​

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']}")

同步执行 ​

python
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

    # 示例：工具调用缓存

    用户请求：
    "实现工具调用缓存"

    Claude Code 生成的代码：

    ```python

```python
    from typing import Dict, Any, Optional
    import hashlib
    import json
    from datetime import datetime, timedelta

    class ToolCallCache:
    """工具调用缓存"""
python
        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)
bash
            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)
python
            if key in self.cache:
                cached = self.cache[key]
            # 检查是否过期
python
                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)
python
            self.cache[key] = {
                'result': result,
                'timestamp': datetime.utcnow()
            }

        def clear(self):
        """清空缓存"""
python
            self.cache.clear()

        def cleanup(self):
        """清理过期缓存"""
        current_time = datetime.utcnow()
        expired_keys = [
            key for key, cached in self.cache.items()
python
                if current_time - cached['timestamp'] >= timedelta(seconds=self.ttl)
            ]

            for key in expired_keys:
                del self.cache[key]

    class CachedToolExecutor(ToolExecutor):
    """带缓存的工具执行器"""
python
        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)
bash
            if cached_result:
                return cached_result
        # 执行工具调用
        result = await super().execute_tool_call(tool_call)

        # 缓存结果
bash
            if result['success']:
                self.cache.set(tool_name, parameters, result)

            return result
### 2. 批量执行
python
    # 示例：批量工具调用

    用户请求：
    "实现批量工具调用"

    Claude Code 生成的代码：

    ```python

```python
    from typing import Dict, Any, List
    import asyncio

    class BatchToolExecutor(ToolExecutor):
    """批量工具执行器"""
python
        async def execute_batch(self, batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """批量执行工具调用"""
        # 按工具类型分组
        tool_groups = {}
bash
            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 = [
python
                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 = []
bash
            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)
bash
            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

工具调用机制包括：

  1. **工具调用的基本概念** : 什么是工具调用、工具调用流程
  2. **工具定义与注册** : 工具定义、工具注册
  3. **工具调用执行** : 工具调用解析、工具调用执行
  4. **工具调用优化** : 缓存机制、批量执行

通过工具调用机制，Claude Code可以执行各种实际操作，大大扩展了其能力。

在下一节中，我们将探讨自主规划算法。

基于 MIT 许可发布

Copyright © 2024-present