函数调用

演示如何为 LLM 定义工具、解析模型返回的工具调用、本地执行工具并把结果回传给模型， 完成一次完整的 ReAct 式闭环。所有逻辑通过依赖注入可在测试中用伪造客户端验证。 对应教程：第3卷《OpenAI API》第3章 函数调用。

文件路径: examples/volume-03-openai-api\function_calling.py

---

源代码

"""函数调用（Function Calling / Tool Calling）示例。

演示如何为 LLM 定义工具、解析模型返回的工具调用、本地执行工具并把结果回传给模型，
完成一次完整的 ReAct 式闭环。所有逻辑通过依赖注入可在测试中用伪造客户端验证。

对应教程：第3卷《OpenAI API》第3章 函数调用。
"""

from __future__ import annotations

import json
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Protocol


class ChatClient(Protocol):
    """支持工具调用的聊天补全客户端协议。"""

    @property
    def chat(self) -> Any: ...


@dataclass
class ToolCall:
    """一次工具调用请求。

    Attributes:
        id: 调用 ID（用于回传结果时对应）。
        name: 工具名称。
        arguments: 工具参数（已解析的字典）。
    """

    id: str
    name: str
    arguments: dict[str, Any]

    @classmethod
    def from_raw(cls, raw: Any) -> ToolCall:
        """从模型返回的原始对象构造 ``ToolCall``。

        Args:
            raw: 形如 ``{"id": ..., "function": {"name": ..., "arguments": "..."}}``
                的字典或对象。

        Returns:
            ``ToolCall`` 实例。

        Raises:
            ValueError: 参数 JSON 非法时。
        """
        if isinstance(raw, dict):
            call_id = raw.get("id", "")
            function = raw.get("function", {})
            name = function.get("name", "")
            raw_args = function.get("arguments", "{}")
        else:
            call_id = getattr(raw, "id", "")
            function = getattr(raw, "function", None)
            name = getattr(function, "name", "")
            raw_args = getattr(function, "arguments", "{}")
        try:
            arguments = json.loads(raw_args) if isinstance(raw_args, str) else dict(raw_args)
        except json.JSONDecodeError as exc:
            raise ValueError(f"工具参数 JSON 非法: {raw_args!r}") from exc
        return cls(id=call_id, name=name, arguments=arguments)


@dataclass
class ToolDef:
    """工具定义。

    Attributes:
        name: 工具名称。
        description: 工具描述。
        parameters: JSON Schema 参数定义。
        func: 实际执行的 Python 函数。
    """

    name: str
    description: str
    parameters: dict[str, Any]
    func: Callable[..., Any]

    def to_openai_schema(self) -> dict[str, Any]:
        """转换为 OpenAI 工具调用 schema。

        Returns:
        OpenAI ``tools`` 字段所需的字典。
        """
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }


@dataclass
class ToolRegistry:
    """工具注册表。"""

    _tools: dict[str, ToolDef] = field(default_factory=dict)

    def register(self, tool: ToolDef) -> None:
        """注册工具。

        Args:
        tool: 工具定义。
        """
        self._tools[tool.name] = tool

    def schemas(self) -> list[dict[str, Any]]:
        """返回所有工具的 OpenAI schema。

        Returns:
        schema 列表。
        """
        return [t.to_openai_schema() for t in self._tools.values()]

    def dispatch(self, call: ToolCall) -> str:
        """执行一次工具调用。

        Args:
        call: 工具调用请求。

        Returns:
        工具执行结果（字符串）。

        Raises:
        KeyError: 工具未注册时。
        """
        tool = self._tools.get(call.name)
        if tool is None:
            raise KeyError(f"未注册的工具: {call.name}")
        result = tool.func(**call.arguments)
        return str(result)


def extract_tool_calls(response: Any) -> list[ToolCall]:
    """从模型响应中提取工具调用列表。

    Args:
    response: 聊天补全响应对象。

    Returns:
    ``ToolCall`` 列表；若无工具调用返回空列表。
    """
    choices = getattr(response, "choices", None)
    if not choices:
        return []
    message = getattr(choices[0], "message", None)
    if message is None:
        return []
    raw_calls = getattr(message, "tool_calls", None)
    if not raw_calls:
        return []
    return [ToolCall.from_raw(c) for c in raw_calls]


def get_response_text(response: Any) -> str:
    """提取模型回复文本。"""
    choices = getattr(response, "choices", None)
    if not choices:
        return ""
    message = getattr(choices[0], "message", None)
    return getattr(message, "content", "") or ""


def run_function_call_loop(
    client: ChatClient,
    user_query: str,
    registry: ToolRegistry,
    model: str = "gpt-4o-mini",
    max_rounds: int = 4,
) -> str:
    """执行完整的函数调用闭环。

    流程：发起请求 → 若返回工具调用则执行并把结果回传 → 直到模型给出最终文本。

    Args:
    client: 符合 ``ChatClient`` 协议的客户端。
    user_query: 用户问题。
    registry: 工具注册表。
    model: 模型名称。
    max_rounds: 最大往返轮数，防止死循环。

    Returns:
    模型最终回复文本。

    Raises:
    RuntimeError: 超过最大轮数仍未结束。
    """
    messages: list[dict[str, Any]] = [{"role": "user", "content": user_query}]
    tools = registry.schemas()

    for _ in range(max_rounds):
        response = client.chat.completions.create(  # type: ignore[union-attr]
            model=model,
            messages=messages,
            tools=tools,
        )
        tool_calls = extract_tool_calls(response)
        if not tool_calls:
            return get_response_text(response)

        # 把助手发起的工具调用加入历史
        messages.append(
            {
                "role": "assistant",
                "content": get_response_text(response),
                "tool_calls": [
                    {
                        "id": c.id,
                        "type": "function",
                        "function": {"name": c.name, "arguments": json.dumps(c.arguments)},
                    }
                    for c in tool_calls
                ],
            }
        )
        # 执行每个工具并把结果作为 tool 消息回传
        for call in tool_calls:
            result = registry.dispatch(call)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})

    raise RuntimeError("函数调用循环超过最大轮数仍未结束")


# ===== 内置示例工具 =====


def get_weather(city: str, unit: str = "celsius") -> str:
    """示例工具：获取城市天气（教学用，返回固定值）。

    Args:
        city: 城市名。
        unit: 温度单位。

    Returns:
        天气描述字符串。
    """
    fake_temp = {"北京": 18, "上海": 22, "广州": 28}.get(city, 20)
    if unit == "fahrenheit":
        fake_temp = fake_temp * 9 // 5 + 32
        return f"{city}：{fake_temp}°F，晴。"
    return f"{city}：{fake_temp}°C，晴。"


def calculate(expression: str) -> str:
    """示例工具：计算算术表达式。

    Args:
    expression: 算术表达式字符串。

    Returns:
    计算结果字符串。
    """
    allowed = set("0123456789+-*/(). ")
    if not expression or not set(expression) <= allowed:
        return "错误：表达式非法。"
    try:
        value = eval(expression, {"__builtins__": {}}, {})  # noqa: S307
    except ZeroDivisionError:
        return "错误：除零。"
    except Exception:  # noqa: BLE001
        return "错误：无法计算。"
    return f"{expression} = {value}"


def build_default_registry() -> ToolRegistry:
    """构造包含示例工具的注册表。

    Returns:
    注册了天气与计算工具的 ``ToolRegistry``。
    """
    registry = ToolRegistry()
    registry.register(
        ToolDef(
            name="get_weather",
            description="获取指定城市的当前天气",
            parameters={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["city"],
            },
            func=get_weather,
        )
    )
    registry.register(
        ToolDef(
            name="calculate",
            description="计算一个只含数字与四则运算符的算术表达式",
            parameters={
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "算术表达式"},
                },
                "required": ["expression"],
            },
            func=calculate,
        )
    )
    return registry


def main() -> None:
    """演示工具调用闭环（使用伪造客户端，不调用真实 API）。"""
    # 构造一个按脚本返回的伪造客户端
    import types

    class _FakeMessage:
        def __init__(self, content: str, tool_calls: list | None = None) -> None:
            self.content = content
            self.tool_calls = tool_calls

    class _FakeChoice:
        def __init__(self, message: _FakeMessage) -> None:
            self.message = message

    class _FakeResponse:
        def __init__(self, message: _FakeMessage) -> None:
            self.choices = [_FakeChoice(message)]

    class _FakeCompletions:
        def __init__(self, script: list[_FakeMessage]) -> None:
            self._script = script
            self._i = 0

        def create(self, **kwargs: object) -> _FakeResponse:
            msg = self._script[self._i]
            self._i += 1
            return _FakeResponse(msg)

    class _FakeChat:
        def __init__(self, script: list[_FakeMessage]) -> None:
            self.completions = _FakeCompletions(script)

    class _FakeClient:
        def __init__(self, script: list[_FakeMessage]) -> None:
            self.chat = _FakeChat(script)

    script = [
        _FakeMessage(
            content="",
            tool_calls=[
                types.SimpleNamespace(
                    id="call-1",
                    function=types.SimpleNamespace(
                        name="get_weather",
                        arguments=json.dumps({"city": "北京"}),
                    ),
                )
            ],
        ),
        _FakeMessage(content="北京目前 18°C，晴天，适合出行。"),
    ]
    client = _FakeClient(script)
    registry = build_default_registry()
    answer = run_function_call_loop(client, "北京天气怎么样？", registry)
    print(f"最终回复: {answer}")


if __name__ == "__main__":
    main()