Function Calling 的微调基础与 tool_calls 回环

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order",
            "description": (
                "按订单号查询订单状态与支付信息，只读，不修改任何数据。"
                "order_id 必须是 18 位纯数字；如果用户给的是手机号或物流单号，"
                "先调用 search_orders 定位订单，不要用本工具试探。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "18 位订单号，纯数字，不带空格和横线",
                    },
                    "fields": {
                        "type": "array",
                        "items": {
                            "type": "string",
                            "enum": ["status", "paid_at", "amount", "refundable"],
                        },
                        "description": "需要返回的字段，缺省返回 status 和 refundable",
                    },
                },
                "required": ["order_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "create_refund",
            "description": (
                "对一笔已支付订单发起退款，有副作用，会产生真实资金流水。"
                "仅在订单状态为 paid 且 refundable 为 true 时调用；"
                "金额超过订单实付金额时不要调用，改为向用户确认。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "18 位订单号"},
                    "amount": {
                        "type": "number",
                        "description": "退款金额，单位元，最多两位小数",
                    },
                    "reason": {
                        "type": "string",
                        "enum": ["quality", "wrong_item", "late_delivery", "other"],
                    },
                },
                "required": ["order_id", "amount", "reason"],
            },
        },
    },
]

import json
from openai import OpenAI

client = OpenAI()
MODEL = "gpt-4o-mini"  # 换成账号下支持 tools 的模型


# order_service / refund_service 由你的业务层注入
def dispatch(name, args):
    if name == "get_order":
        return order_service.fetch(args["order_id"], args.get("fields"))
    if name == "create_refund":
        return refund_service.create(args["order_id"], args["amount"], args["reason"])
    raise KeyError(name)


def run_turn(messages, tools, max_rounds=6):
    for _ in range(max_rounds):
        resp = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content

        # 一轮可能返回多个调用，逐个执行，逐个配对
        for call in msg.tool_calls:
            try:
                args = json.loads(call.function.arguments or "{}")
            except json.JSONDecodeError as exc:
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": f"arguments 不是合法 JSON：{exc.msg}（位置 {exc.pos}）。"
                               f"只输出 JSON 对象，不要注释、不要尾逗号。",
                })
                continue

            try:
                result = dispatch(call.function.name, args)
                text = json.dumps(result, ensure_ascii=False)
            except Exception as exc:
                # 错误也要回灌，否则下一次请求会缺一条 tool 消息
                text = f"执行失败 {type(exc).__name__}: {exc}。可换参数重试或向用户说明。"

            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": text,
            })

    return None  # 轮次用尽，交给上层降级

import json
from jsonschema import Draft202012Validator

VALIDATORS = {
    "get_order": Draft202012Validator(TOOLS[0]["function"]["parameters"]),
    "create_refund": Draft202012Validator(TOOLS[1]["function"]["parameters"]),
}


def call_tool(call, registry):
    name = call.function.name

    if name not in registry:
        allowed = ", ".join(sorted(registry))
        return f"unknown_tool: 没有名为 {name} 的工具。可用工具：{allowed}。"

    try:
        args = json.loads(call.function.arguments or "{}")
    except json.JSONDecodeError as exc:
        return f"invalid_json: {exc.msg}（位置 {exc.pos}）。请重新生成完整的 JSON 对象。"

    errors = sorted(VALIDATORS[name].iter_errors(args), key=lambda e: list(e.path))
    if errors:
        detail = "; ".join(
            f"{'/'.join(map(str, e.path)) or ''}: {e.message}" for e in errors
        )
        return f"schema_error: {detail}。请按上述字段修正后重新调用。"

    try:
        return json.dumps(registry[name](**args), ensure_ascii=False, default=str)
    except TypeError as exc:
        return f"bad_arguments: {exc}"
    except Exception as exc:
        return f"tool_failed: {type(exc).__name__}: {exc}"