Function Calling：从模型输出到安全的工具执行

get_weather(city="beijing")

tools = [
    {
        "type": "function",
        "name": "get_current_weather",
        "description": (
            "Get the current weather for a city. "
            "Use this tool when the user asks about current weather, "
            "temperature, humidity, or wind. Do not use it for historical weather."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country, for example: Beijing, China",
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit",
                },
            },
            "required": ["location", "unit"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]

import json
import os
from openai import OpenAI

client = OpenAI()
MODEL = os.environ["OPENAI_MODEL"]  # 使用当前支持工具调用的模型

TOOLS = [
    {
        "type": "function",
        "name": "get_current_weather",
        "description": "Get the current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country, e.g. Beijing, China",
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                },
            },
            "required": ["location", "unit"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


def get_current_weather(location: str, unit: str) -> dict:
    # 真实项目应在这里调用天气服务。
    return {
        "location": location,
        "temperature": 26,
        "unit": unit,
        "condition": "cloudy",
    }


def validate_business_rules(arguments: dict) -> None:
    # strict mode 不会替你检查城市是否真实存在。
    if len(arguments["location"].strip()) < 2:
        raise ValueError("location is invalid")


def execute_function(name: str, arguments: dict) -> dict:
    # 工具名称也必须使用服务端白名单，不能动态执行任意函数。
    if name == "get_current_weather":
        validate_business_rules(arguments)
        return get_current_weather(**arguments)

    raise ValueError(f"unknown tool: {name}")


def execute_tool_call(item) -> dict:
    try:
        arguments = json.loads(item.arguments)
        return {
            "ok": True,
            "data": execute_function(item.name, arguments),
        }
    except (json.JSONDecodeError, ValueError) as error:
        return {
            "ok": False,
            "error": {
                "code": "INVALID_TOOL_ARGUMENTS",
                "message": str(error),
                "retryable": False,
            },
        }
    except Exception:
        # 详细异常只写入服务端日志，不返回堆栈、密钥或内部实现。
        return {
            "ok": False,
            "error": {
                "code": "TOOL_EXECUTION_FAILED",
                "message": "The tool failed unexpectedly",
                "retryable": True,
            },
        }


def run_agent(user_message: str, max_tool_rounds: int = 8) -> str:
    input_items = [{"role": "user", "content": user_message}]

    for _ in range(max_tool_rounds):
        response = client.responses.create(
            model=MODEL,
            input=input_items,
            tools=TOOLS,
        )

        # 保留模型的完整输出，包括可能存在的 reasoning items。
        input_items += response.output
        tool_calls = [
            item for item in response.output
            if item.type == "function_call"
        ]

        # 没有工具调用，说明模型已经给出最终回答。
        if not tool_calls:
            return response.output_text

        # 一次响应可能包含多个工具调用，不能只处理第一个。
        for item in tool_calls:
            result = execute_tool_call(item)
            input_items.append(
                {
                    "type": "function_call_output",
                    "call_id": item.call_id,
                    "output": json.dumps(result, ensure_ascii=False),
                }
            )

    raise RuntimeError("maximum tool call rounds exceeded")


print(run_agent("北京现在天气怎么样？使用摄氏度。"))

{
  "type": "object",
  "properties": {
    "date": {
      "type": ["string", "null"],
      "description": "ISO 8601 date, or null when not specified"
    }
  },
  "required": ["date"],
  "additionalProperties": false
}

{"location": "beijing1", "unit": "celsius"}

response = client.responses.create(
    model=MODEL,
    input=input_items,
    tools=TOOLS,
    parallel_tool_calls=False,
)

execute_sql(sql)
run_shell(command)
send_http_request(url, method, body)

{"sql": "DROP TABLE orders"}

get_order_status(order_id)
list_customer_orders(customer_id, page_size)
cancel_order(order_id, reason, idempotency_key)

{
  "ok": false,
  "error": {
    "code": "WEATHER_SERVICE_TIMEOUT",
    "message": "Weather service did not respond within 3 seconds",
    "retryable": true
  }
}

search
query
create_info
do_action

get_current_weather
get_order_status
create_support_ticket
search_github_trending_repositories

This tool queries weather.

Get the current weather for a city. Use this tool when the user asks
about current temperature, humidity, or wind. Do not use it for
historical weather, climate statistics, or travel recommendations.

{"unit": "用户随便填写的字符串"}

{"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}}

search_order
query_order
get_order

get_order_by_id(order_id)
search_orders(customer_id, status, page_token)

模型理解任务
    ↓
生成工具调用请求
    ↓
Runtime 校验、鉴权并执行
    ↓
返回工具执行结果
    ↓
模型继续推理并回答