Day 13：工具调用机制深入 - 从手动解析到标准化

今日目标

1. 解决昨天遇到的参数传递问题
2. 理解工具调用的两种模式：手动解析 vs Function Calling
3. 实现健壮的工具执行层
4. 学习 OpenAI Function Calling 的标准用法

---

一、问题分析：为什么会报错？

昨天的错误：

get_current_time() takes 0 positional arguments but 1 was given

根本原因：我们用统一的方式调用所有工具，但不同工具的参数签名不同。

你的代码可能是这样

这暴露了手动解析模式的一个核心问题：参数处理不够灵活。

result = tools[action](action_input)  # 所有工具都传参数，但 get_current_time 不需要参数

---

二、解决方案 1：改进手动解析模式

2.1 使用 inspect 动态检查参数

import inspect
from typing import Callable, Any

def execute_tool(tool_func: Callable, action_input: str) -> str:
    """
    智能执行工具，根据函数签名决定是否传参
    """
    sig = inspect.signature(tool_func)
    params = sig.parameters
    
    if len(params) == 0:
        # 无参数函数，如 get_current_time
        return tool_func()
    else:
        # 有参数函数
        # 检查 action_input 是否为空或无效
        if not action_input or action_input.strip() in ["无", "None", ""]:
            return "错误：该工具需要输入参数"
        return tool_func(action_input)

2.2 完整的工具注册和执行系统

import inspect
import json
from datetime import datetime
from typing import Callable, Dict, Any, Optional
from dataclasses import dataclass

@dataclass
class Tool:
    """工具定义"""
    name: str
    description: str
    func: Callable
    parameters: Dict[str, Any]  # 参数说明
    
    def execute(self, input_str: str) -> str:
        """执行工具"""
        sig = inspect.signature(self.func)
        
        try:
            if len(sig.parameters) == 0:
                return str(self.func())
            else:
                # 尝试解析 JSON 格式的输入
                try:
                    params = json.loads(input_str)
                    if isinstance(params, dict):
                        return str(self.func(**params))
                except json.JSONDecodeError:
                    # 不是 JSON，作为单个字符串参数传入
                    return str(self.func(input_str))
        except Exception as e:
            return f"工具执行错误: {str(e)}"


class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self.tools: Dict[str, Tool] = {}
    
    def register(self, name: str, description: str, 
                 parameters: Optional[Dict] = None):
        """装饰器方式注册工具"""
        def decorator(func: Callable):
            self.tools[name] = Tool(
                name=name,
                description=description,
                func=func,
                parameters=parameters or {}
            )
            return func
        return decorator
    
    def get(self, name: str) -> Optional[Tool]:
        return self.tools.get(name)
    
    def execute(self, name: str, input_str: str) -> str:
        tool = self.get(name)
        if not tool:
            return f"未知工具: {name}"
        return tool.execute(input_str)
    
    def get_tools_description(self) -> str:
        """生成工具描述文本，用于 prompt"""
        lines = ["可用工具："]
        for name, tool in self.tools.items():
            lines.append(f"\n- {name}: {tool.description}")
            if tool.parameters:
                lines.append(f"  参数: {json.dumps(tool.parameters, ensure_ascii=False)}")
        return "\n".join(lines)


# 创建全局工具注册中心
registry = ToolRegistry()

# 注册工具
@registry.register(
    name="get_current_time",
    description="获取当前时间",
    parameters={}  # 无参数
)
def get_current_time():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


@registry.register(
    name="calculator",
    description="计算数学表达式",
    parameters={"expression": "数学表达式，如 '2 + 3 * 4'"}
)
def calculator(expression: str):
    """安全的计算器"""
    allowed_chars = set('0123456789+-*/.() ')
    if not all(c in allowed_chars for c in expression):
        return "错误：表达式包含非法字符"
    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"


@registry.register(
    name="search",
    description="搜索信息（模拟）",
    parameters={"query": "搜索关键词"}
)
def search(query: str):
    # 模拟搜索结果
    mock_results = {
        "天气": "北京今日天气：晴，气温 15-25°C",
        "新闻": "今日头条：AI 技术持续突破",
        "python": "Python 是一种流行的编程语言",
    }
    for key, value in mock_results.items():
        if key in query.lower():
            return value
    return f"未找到与 '{query}' 相关的结果"


# 测试
if __name__ == "__main__":
    print(registry.get_tools_description())
    print("\n" + "="*50 + "\n")
    
    # 测试执行
    print("测试 get_current_time:", registry.execute("get_current_time", ""))
    print("测试 calculator:", registry.execute("calculator", "2 + 3 * 4"))
    print("测试 search:", registry.execute("search", "python"))

---

三、解决方案 2：使用 Function Calling（推荐）

手动解析 LLM 输出容易出错。现代 LLM API 提供了 Function Calling 功能，让模型以结构化方式返回工具调用意图。

3.1 Function Calling 的优势

3.2 OpenAI Function Calling 实现

import os
import json
from openai import OpenAI
from datetime import datetime
from typing import Callable, Dict, Any, List

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 定义工具的 JSON Schema（OpenAI 格式）
tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前的日期和时间",
            "parameters": {
                "type": "object",
                "properties": {},  # 无参数
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "计算数学表达式，支持加减乘除和括号",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "要计算的数学表达式，如 '2 + 3 * 4'"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 '北京'、'上海'"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# 工具实现
def get_current_time() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def calculator(expression: str) -> str:
    allowed_chars = set('0123456789+-*/.() ')
    if not all(c in allowed_chars for c in expression):
        return "错误：表达式包含非法字符"
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

def get_weather(city: str) -> str:
    # 模拟天气 API
    weather_data = {
        "北京": "晴，15-25°C，空气质量良好",
        "上海": "多云，18-23°C，湿度较高",
        "广州": "小雨，22-28°C，记得带伞",
    }
    return weather_data.get(city, f"暂无 {city} 的天气数据")

# 工具函数映射
tool_functions = {
    "get_current_time": get_current_time,
    "calculator": calculator,
    "get_weather": get_weather,
}


def execute_function_call(tool_call) -> str:
    """执行 function call"""
    func_name = tool_call.function.name
    func = tool_functions.get(func_name)
    
    if not func:
        return f"未知函数: {func_name}"
    
    # 解析参数
    try:
        args = json.loads(tool_call.function.arguments)
    except json.JSONDecodeError:
        args = {}
    
    # 执行函数
    try:
        return func(**args)
    except Exception as e:
        return f"执行错误: {str(e)}"


def chat_with_tools(user_message: str) -> str:
    """
    带工具调用的对话
    """
    messages = [{"role": "user", "content": user_message}]
    
    # 第一次调用：让模型决定是否使用工具
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools_schema,
        tool_choice="auto"  # 让模型自己决定
    )
    
    assistant_message = response.choices[0].message
    
    # 检查是否需要调用工具
    if assistant_message.tool_calls:
        # 有工具调用
        messages.append(assistant_message)  # 添加助手的回复
        
        # 执行所有工具调用
        for tool_call in assistant_message.tool_calls:
            print(f"[调用工具] {tool_call.function.name}({tool_call.function.arguments})")
            
            result = execute_function_call(tool_call)
            print(f"[工具结果] {result}")
            
            # 添加工具结果到消息
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
        
        # 第二次调用：让模型基于工具结果生成最终回复
        final_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages
        )
        return final_response.choices[0].message.content
    
    else:
        # 不需要工具，直接返回
        return assistant_message.content


# 测试
if __name__ == "__main__":
    test_queries = [
        "现在几点了？",
        "帮我算一下 (15 + 27) * 3 - 18 / 2",
        "北京今天天气怎么样？",
        "你好，介绍一下你自己",  # 不需要工具
    ]
    
    for query in test_queries:
        print(f"\n{'='*50}")
        print(f"用户: {query}")
        print(f"{'='*50}")
        response = chat_with_tools(query)
        print(f"助手: {response}")

3.3 运行结果示例

==================================================
用户: 现在几点了？
==================================================
[调用工具] get_current_time({})
[工具结果] 2024-01-15 14:32:18
助手: 现在是 2024年1月15日 14:32:18。

==================================================
用户: 帮我算一下 (15 + 27) * 3 - 18 / 2
==================================================
[调用工具] calculator({"expression": "(15 + 27) * 3 - 18 / 2"})
[工具结果] 117.0
助手: 计算结果是 117。

==================================================
用户: 北京今天天气怎么样？
==================================================
[调用工具] get_weather({"city": "北京"})
[工具结果] 晴，15-25°C，空气质量良好
助手: 北京今天天气晴朗，气温在 15-25°C 之间，空气质量良好，是个适合外出的好天气！

==================================================
用户: 你好，介绍一下你自己
==================================================
助手: 你好！我是一个 AI 助手，可以帮你查询时间、计算数学表达式、查看天气等...

---

四、并行工具调用

有时一个问题需要同时调用多个工具：

def chat_with_parallel_tools(user_message: str) -> str:
    """
    支持并行工具调用的对话
    """
    messages = [{"role": "user", "content": user_message}]
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools_schema,
        tool_choice="auto",
        parallel_tool_calls=True  # 允许并行调用
    )
    
    assistant_message = response.choices[0].message
    
    if assistant_message.tool_calls:
        messages.append(assistant_message)
        
        print(f"[并行调用 {len(assistant_message.tool_calls)} 个工具]")
        
        # 可以用 asyncio 或线程池并行执行
        # 这里简单顺序执行演示
        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            func_args = tool_call.function.arguments
            print(f"  - {func_name}({func_args})")
            
            result = execute_function_call(tool_call)
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
        
        final_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages
        )
        return final_response.choices[0].message.content
    
    return assistant_message.content


# 测试并行调用
print(chat_with_parallel_tools("现在几点了？北京和上海的天气分别怎么样？"))

输出：

[并行调用 3 个工具]
  - get_current_time({})
  - get_weather({"city": "北京"})
  - get_weather({"city": "上海"})
助手: 现在是 14:35:22。北京今天晴朗，15-25°C；上海多云，18-23°C，湿度较高。

---

五、Claude 的 Tool Use

Claude 也支持类似的功能，语法稍有不同：

import anthropic

client = anthropic.Anthropic()

# Claude 的工具定义格式
claude_tools = [
    {
        "name": "get_current_time",
        "description": "获取当前的日期和时间",
        "input_schema": {
            "type": "object",
            "properties": {},
            "required": []
        }
    },
    {
        "name": "calculator",
        "description": "计算数学表达式",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式"
                }
            },
            "required": ["expression"]
        }
    }
]


def chat_with_claude_tools(user_message: str) -> str:
    """Claude 的工具调用"""
    messages = [{"role": "user", "content": user_message}]
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        tools=claude_tools,
        messages=messages
    )
    
    # 处理响应
    final_text = []
    tool_results = []
    
    for block in response.content:
        if block.type == "text":
            final_text.append(block.text)
        elif block.type == "tool_use":
            # 执行工具
            tool_name = block.name
            tool_input = block.input
            print(f"[调用工具] {tool_name}({tool_input})")
            
            # 执行
            if tool_name == "get_current_time":
                result = get_current_time()
            elif tool_name == "calculator":
                result = calculator(tool_input.get("expression", ""))
            else:
                result = "未知工具"
            
            print(f"[工具结果] {result}")
            
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result
            })
    
    # 如果有工具调用，需要继续对话
    if tool_results:
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
        
        final_response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=claude_tools,
            messages=messages
        )
        
        for block in final_response.content:
            if block.type == "text":
                final_text.append(block.text)
    
    return "\n".join(final_text)

---

六、统一工具接口设计

为了兼容不同模型，我们可以设计统一的工具接口：

from abc import ABC, abstractmethod
from typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum


class ModelProvider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    MANUAL = "manual"  # 手动解析模式


@dataclass
class ToolDefinition:
    """统一的工具定义"""
    name: str
    description: str
    parameters: Dict[str, Any]  # JSON Schema 格式
    func: callable
    
    def to_openai_format(self) -> Dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": self.parameters,
                    "required": list(self.parameters.keys())
                }
            }
        }
    
    def to_claude_format(self) -> Dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": {
                "type": "object",
                "properties": self.parameters,
                "required": list(self.parameters.keys())
            }
        }
    
    def to_prompt_format(self) -> str:
        """用于手动解析模式的 prompt 描述"""
        params_desc = ", ".join([
            f"{k}: {v.get('description', '')}" 
            for k, v in self.parameters.items()
        ])
        return f"- {self.name}: {self.description}" + \
               (f" (参数: {params_desc})" if params_desc else "")


class ToolManager:
    """统一工具管理器"""
    
    def __init__(self):
        self.tools: Dict[str, ToolDefinition] = {}
    
    def register(self, name: str, description: str, 
                 parameters: Dict[str, Any] = None):
        """装饰器注册工具"""
        def decorator(func):
            self.tools[name] = ToolDefinition(
                name=name,
                description=description,
                parameters=parameters or {},
                func=func
            )
            return func
        return decorator
    
    def get_tools_for_provider(self, provider: ModelProvider) -> List[Dict]:
        """获取指定提供商格式的工具列表"""
        if provider == ModelProvider.OPENAI:
            return [t.to_openai_format() for t in self.tools.values()]
        elif provider == ModelProvider.ANTHROPIC:
            return [t.to_claude_format() for t in self.tools.values()]
        else:
            return [t.to_prompt_format() for t in self.tools.values()]
    
    def execute(self, name: str, **kwargs) -> str:
        """执行工具"""
        tool = self.tools.get(name)
        if not tool:
            return f"未知工具: {name}"
        try:
            return str(tool.func(**kwargs))
        except Exception as e:
            return f"执行错误: {str(e)}"


# 使用示例
tool_manager = ToolManager()

@tool_manager.register(
    name="search_web",
    description="搜索互联网获取信息",
    parameters={
        "query": {"type": "string", "description": "搜索关键词"},
        "num_results": {"type": "integer", "description": "返回结果数量"}
    }
)
def search_web(query: str, num_results: int = 5) -> str:
    return f"搜索 '{query}' 的前 {num_results} 条结果..."


# 获取不同格式
print("OpenAI 格式:")
print(json.dumps(tool_manager.get_tools_for_provider(ModelProvider.OPENAI), indent=2, ensure_ascii=False))

print("\nClaude 格式:")
print(json.dumps(tool_manager.get_tools_for_provider(ModelProvider.ANTHROPIC), indent=2, ensure_ascii=False))

print("\nPrompt 格式:")
print(tool_manager.get_tools_for_provider(ModelProvider.MANUAL))

---

七、今日练习

任务 1：修复你的代码

用 inspect 或条件判断修复昨天的参数传递问题。

任务 2：实现 Function Calling 版本

把你的 Agent 改成使用 OpenAI Function Calling（如果用的是 Claude 就用 Tool Use）。

任务 3：添加更多工具

实现以下工具并注册：

• read_file(path): 读取本地文件
• write_file(path, content): 写入文件
• run_python(code): 执行 Python 代码（注意安全性）

任务 4（选做）：工具链

让 Agent 解决这个问题：“把当前时间和北京天气写入 report.txt 文件”
这需要 Agent 连续调用多个工具：get_current_time → get_weather → write_file

---

八、总结

明天预告：Day 14 我们将实现完整的 ReAct Agent 循环，加入错误恢复、最大步数限制、上下文管理等生产级特性。

---

有问题随时问，特别是如果你在修复昨天代码时遇到困难，把代码贴出来我帮你看。