做出第一个会调用工具的 Agent

用户问题 → 模型 → 回答

用户问题
  → 模型决定调用哪个工具，并生成参数
  → 你的程序检查工具名和参数
  → 你的程序执行工具
  → 工具结果回到模型
  → 模型生成最终回答

模型输出
  ├─ 普通回答：直接结束
  └─ 工具请求
       ├─ 工具名
       ├─ JSON 参数
       └─ 本次调用 ID

python3 -m venv .venv
source .venv/bin/activate
python -m pip install openai

export BIGAO_API_KEY="你的密钥"
export BIGAO_MODEL_ID="控制台中支持工具调用的模型ID"

DEMO_WEATHER = {
    "台北": {"condition": "多云", "temperature_c": 28},
    "上海": {"condition": "小雨", "temperature_c": 25},
    "北京": {"condition": "晴", "temperature_c": 30},
}


def get_weather(city: str) -> dict:
    normalized = city.strip()
    if not normalized or len(normalized) > 20:
        raise ValueError("城市名称不合法")

    weather = DEMO_WEATHER.get(normalized)
    if weather is None:
        return {"city": normalized, "found": False}

    return {
        "city": normalized,
        "found": True,
        **weather,
        "notice": "这是教程演示数据，不是实时天气",
    }

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气演示数据。只在用户询问天气时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市中文名，例如台北、上海或北京",
                    }
                },
                "required": ["city"],
                "additionalProperties": False,
            },
        },
    }
]

import json
import os

from openai import OpenAI


API_KEY = os.environ.get("BIGAO_API_KEY")
MODEL_ID = os.environ.get("BIGAO_MODEL_ID")

if not API_KEY or not MODEL_ID:
    raise RuntimeError("请先设置 BIGAO_API_KEY 和 BIGAO_MODEL_ID")

client = OpenAI(
    api_key=API_KEY,
    base_url="https://api.bigao.ai/v1",
)

DEMO_WEATHER = {
    "台北": {"condition": "多云", "temperature_c": 28},
    "上海": {"condition": "小雨", "temperature_c": 25},
    "北京": {"condition": "晴", "temperature_c": 30},
}


def get_weather(city: str) -> dict:
    normalized = city.strip()
    if not normalized or len(normalized) > 20:
        raise ValueError("城市名称不合法")
    weather = DEMO_WEATHER.get(normalized)
    if weather is None:
        return {"city": normalized, "found": False}
    return {
        "city": normalized,
        "found": True,
        **weather,
        "notice": "这是教程演示数据，不是实时天气",
    }


TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气演示数据。只在用户询问天气时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市中文名，例如台北、上海或北京",
                    }
                },
                "required": ["city"],
                "additionalProperties": False,
            },
        },
    }
]


def run_agent(user_input: str) -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "你是天气演示助手。只根据工具返回的数据回答，"
                "并明确提醒用户数据不是实时天气。"
            ),
        },
        {"role": "user", "content": user_input},
    ]

    first = client.chat.completions.create(
        model=MODEL_ID,
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
    )
    assistant_message = first.choices[0].message
    messages.append(assistant_message)

    if not assistant_message.tool_calls:
        return assistant_message.content or "模型没有返回内容"

    if len(assistant_message.tool_calls) > 1:
        raise RuntimeError("本教程每轮最多允许一次工具调用")

    tool_call = assistant_message.tool_calls[0]
    if tool_call.function.name != "get_weather":
        raise RuntimeError("模型请求了未授权工具")

    try:
        arguments = json.loads(tool_call.function.arguments)
        city = arguments["city"]
        tool_result = get_weather(city)
    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
        tool_result = {"ok": False, "error": str(error)}

    messages.append(
        {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(tool_result, ensure_ascii=False),
        }
    )

    final = client.chat.completions.create(
        model=MODEL_ID,
        messages=messages,
        tools=TOOLS,
    )
    return final.choices[0].message.content or "模型没有返回内容"


if __name__ == "__main__":
    question = input("你想查询哪个城市？\n> ").strip()
    print(run_agent(question))

python weather_agent.py

{
  "request_id": "demo-001",
  "state": "DONE",
  "model_calls": 2,
  "tool_calls": 1,
  "tool_name": "get_weather",
  "argument_validation": "passed",
  "tool_status": "success",
  "duration_ms": 842
}

def test_get_weather_known_city():
    result = get_weather("台北")
    assert result["found"] is True
    assert result["city"] == "台北"
    assert "temperature_c" in result


def test_get_weather_unknown_city():
    result = get_weather("广州")
    assert result == {"city": "广州", "found": False}


def test_get_weather_rejects_empty_city():
    try:
        get_weather("   ")
    except ValueError as error:
        assert "不合法" in str(error)
    else:
        raise AssertionError("空城市名应该被拒绝")

MAX_STEPS = 5

for step in range(MAX_STEPS):
    response = call_model(messages, tools=TOOLS)
    if not response.tool_calls:
        return response.content

    for call in response.tool_calls:
        validated = validate_call(call)
        result = execute_allowed_tool(validated)
        messages.append(tool_message(call.id, result))

raise RuntimeError("达到最大步骤，任务仍未完成")