Tool Calling 与 Agent Loop 原理

"""最小 Agent Loop：OpenAI Responses API 版（基于 openai 3.26.1）"""
import json
from typing import Any, Callable

from openai import OpenAI
from openai.types.responses import FunctionToolParam

client = OpenAI()  # 默认从环境变量 OPENAI_API_KEY 读取密钥
MODEL = "gpt-6-astra"  # 示例模型名取自 2026-10 的官方文档，可替换


# 1) 工具的实现：就是普通的 Python 函数
def get_weather(city: str) -> str:
    fake_db = {"北京": "晴，25°C", "上海": "小雨，22°C"}
    return fake_db.get(city, "暂无数据")


TOOL_IMPLS: dict[str, Callable[..., str]] = {"get_weather": get_weather}

# 2) 工具的声明：告诉模型有哪些工具、参数是什么样（JSON Schema）
TOOLS: list[FunctionToolParam] = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "查询某个城市今天的天气",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "城市名，例如 北京"}},
            "required": ["city"],
            "additionalProperties": False,
        },
        "strict": True,  # 严格模式：模型生成的参数一定符合这份 Schema
    }
]


def run_agent(user_input: str, max_turns: int = 10) -> str:
    # 历史里既有我们写的 dict，也有 SDK 返回的输出对象，所以用 list[Any]
    history: list[Any] = [{"role": "user", "content": user_input}]
    for _ in range(max_turns):  # 安全阀：防止无限循环
        response = client.responses.create(
            model=MODEL,
            instructions="你是一个简洁的中文助手。",
            tools=TOOLS,
            input=history,
        )
        # 3) 把模型的全部输出（function_call、reasoning 等 item）原样放回历史
        history += response.output
        calls = [item for item in response.output if item.type == "function_call"]
        if not calls:  # 4) 没有工具调用，说明任务完成
            return response.output_text
        for call in calls:  # 5) 逐个执行工具，并按 call_id 回填结果
            args = json.loads(call.arguments)  # 注意：arguments 是 JSON 字符串
            result = TOOL_IMPLS[call.name](**args)
            history.append(
                {"type": "function_call_output", "call_id": call.call_id, "output": result}
            )
    raise RuntimeError("超过最大轮数仍未完成")


if __name__ == "__main__":
    print(run_agent("北京和上海今天天气怎么样？"))

"""最小 Agent Loop：Anthropic Messages API 版（基于 anthropic 1.12.1）"""
from typing import Callable

import anthropic
from anthropic.types import MessageParam, ToolParam, ToolResultBlockParam

client = anthropic.Anthropic()  # 默认从环境变量 ANTHROPIC_API_KEY 读取
MODEL = "claude-opus-5-5"


def get_weather(city: str) -> str:
    fake_db = {"北京": "晴，25°C", "上海": "小雨，22°C"}
    return fake_db.get(city, "暂无数据")


TOOL_IMPLS: dict[str, Callable[..., str]] = {"get_weather": get_weather}

TOOLS: list[ToolParam] = [
    {
        "name": "get_weather",
        "description": "查询某个城市今天的天气",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "城市名，例如 北京"}},
            "required": ["city"],
        },
    }
]


def run_agent(user_input: str, max_turns: int = 10) -> str:
    messages: list[MessageParam] = [{"role": "user", "content": user_input}]
    for _ in range(max_turns):
        response = client.messages.create(
            model=MODEL,
            max_tokens=16000,
            system="你是一个简洁的中文助手。",
            tools=TOOLS,
            messages=messages,
        )
        # API 无状态：把这一轮模型输出的完整 content（含 tool_use、thinking 块）原样追加
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "refusal":  # 安全分类器拒绝了请求
            return "（模型拒绝了该请求）"
        if response.stop_reason != "tool_use":  # end_turn、max_tokens 等：不再调用工具
            return "".join(b.text for b in response.content if b.type == "text")

        tool_results: list[ToolResultBlockParam] = []
        for block in response.content:
            if block.type == "tool_use":
                try:
                    output = TOOL_IMPLS[block.name](**block.input)  # input 已经是 dict
                    tool_results.append(
                        {"type": "tool_result", "tool_use_id": block.id, "content": output}
                    )
                except Exception as e:  # 把错误告诉模型，让它自己调整做法
                    tool_results.append(
                        {
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"工具出错：{e}",
                            "is_error": True,
                        }
                    )
        # 所有结果必须放在**同一条** user 消息里
        messages.append({"role": "user", "content": tool_results})
    raise RuntimeError("超过最大轮数仍未完成")


if __name__ == "__main__":
    print(run_agent("北京和上海今天天气怎么样？"))

history = [用户输入]
重复，最多 max_turns 次：
    response = 调用模型(history, tools)
    history.append(response 的完整输出)
    如果 response 里没有工具调用：
        返回最终文本
    对 response 里的每个工具调用：
        result = 执行工具(调用.name, 调用.参数)
        history.append(工具结果，并关联到这次调用的 id)
超出轮数则报错