工具调用与 Function Calling 实战

import json
from openai import OpenAI

client = OpenAI()

def execute_function_call(tool_call, available_functions):
    """解析并执行工具调用"""
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)

    function = available_functions.get(function_name)
    if not function:
        return json.dumps({"error": f"未知函数: {function_name}"})

    try:
        result = function(**arguments)
        return json.dumps(result, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": str(e)})

# 注册可用函数
def get_weather(city: str, unit: str = "celsius"):
    # 实际实现中调用天气 API
    return {"city": city, "temperature": 22, "unit": unit}

available_functions = {
    "get_weather": get_weather
}

# 多轮工具调用循环
messages = [
    {"role": "system", "content": "你是一个天气助手"},
    {"role": "user", "content": "北京和上海今天哪个更暖和？"}
]

for _ in range(5):  # 最多 5 轮工具调用
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )

    message = response.choices[0].message

    if not message.tool_calls:
        # 没有工具调用，直接输出
        print(message.content)
        break

    messages.append(message)

    for tool_call in message.tool_calls:
        result = execute_function_call(tool_call, available_functions)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result
        })

from functools import wraps

class ToolRegistry:
    """工具注册与管理"""

    def __init__(self):
        self._tools = {}
        self._schemas = []

    def register(self, name: str = None):
        """装饰器：注册工具"""
        def decorator(func):
            tool_name = name or func.__name__
            self._tools[tool_name] = func
            self._schemas.append(self._build_schema(func))
            return func
        return decorator

    def _build_schema(self, func) -> Dict:
        """从类型注解自动生成 Schema"""
        import inspect
        sig = inspect.signature(func)
        properties = {}
        required = []

        for param_name, param in sig.parameters.items():
            if param_name == "self":
                continue
            properties[param_name] = {
                "type": self._type_to_str(param.annotation),
                "description": f"参数 {param_name}"
            }
            if param.default is inspect.Parameter.empty:
                required.append(param_name)

        return {
            "type": "function",
            "function": {
                "name": func.__name__,
                "description": func.__doc__ or "",
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required
                }
            }
        }

    def _type_to_str(self, t) -> str:
        mapping = {str: "string", int: "integer", float: "number", bool: "boolean"}
        return mapping.get(t, "string")

# 使用示例
registry = ToolRegistry()

@registry.register()
def search(query: str, max_results: int = 5) -> List[Dict]:
    """搜索互联网"""
    ...

@registry.register()
def calculate(expression: str) -> float:
    """执行数学计算"""
    return eval(expression)  # 注意：生产环境需要使用安全计算引擎