Agent 与工具调用

用户问题
   ↓
模型决策：
  直接回答？→ 输出答案，结束
  需要工具？→ 输出「调用工具 X，参数 Y」
              ↓
           宿主代码实际执行工具，把结果返还给模型
              ↓
           模型基于工具结果继续决策（可能再调其他工具）
              ↓
           直到模型认为可以回答，输出最终答案

import json
import datetime
from anthropic import Anthropic

client = Anthropic()

# 1. 定义工具（用 JSON Schema 描述参数）
tools = [
    {
        "name": "get_current_time",
        "description": "返回当前的本地时间（ISO 格式）",
        "input_schema": {
            "type": "object",
            "properties": {},
            "required": [],
        },
    },
    {
        "name": "convert_units",
        "description": "在不同单位之间做数值换算，支持长度、重量、温度",
        "input_schema": {
            "type": "object",
            "properties": {
                "value":      {"type": "number",  "description": "要换算的数值"},
                "from_unit":  {"type": "string",  "description": "原单位，如 km、kg、celsius"},
                "to_unit":    {"type": "string",  "description": "目标单位，如 mile、lb、fahrenheit"},
            },
            "required": ["value", "from_unit", "to_unit"],
        },
    },
]


# 2. 工具的实际实现
def get_current_time() -> str:
    return datetime.datetime.now().isoformat()


def convert_units(value: float, from_unit: str, to_unit: str) -> str:
    table = {
        ("km", "mile"):         value * 0.621371,
        ("mile", "km"):         value * 1.60934,
        ("kg", "lb"):           value * 2.20462,
        ("lb", "kg"):           value / 2.20462,
        ("celsius", "fahrenheit"): value * 9 / 5 + 32,
        ("fahrenheit", "celsius"): (value - 32) * 5 / 9,
    }
    result = table.get((from_unit, to_unit))
    if result is None:
        return f"不支持 {from_unit} → {to_unit} 的换算"
    return f"{value} {from_unit} = {result:.4f} {to_unit}"


def run_tool(name: str, inputs: dict) -> str:
    if name == "get_current_time":
        return get_current_time()
    if name == "convert_units":
        return convert_units(**inputs)
    return f"未知工具: {name}"


# 3. Agent 执行循环
def agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-8",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        # 模型决定直接回答，循环结束
        if response.stop_reason == "end_turn":
            return next(b.text for b in response.content if b.type == "text")

        # 模型请求工具调用
        if response.stop_reason == "tool_use":
            # 先把模型的响应（含 tool_use 块）追加进消息历史
            messages.append({"role": "assistant", "content": response.content})

            # 依次执行所有工具请求，收集结果
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = run_tool(block.name, block.input)
                    tool_results.append({
                        "type":        "tool_result",
                        "tool_use_id": block.id,
                        "content":     result,
                    })

            # 把工具结果作为 user 消息回传，让模型继续
            messages.append({"role": "user", "content": tool_results})


# 测试
print(agent("现在几点了？另外帮我把 100 摄氏度换算成华氏度。"))