工具调用

import os, json, urllib.request, urllib.error

BASE = os.environ["MOSSHUB_API_BASE"].rstrip("/")
KEY = os.environ["MOSSHUB_API_KEY"]
MODEL = "YOUR_MODEL_ID"

def chat(messages, **options):
    body = {"model": MODEL, "messages": messages, "stream": False, **options}
    request = urllib.request.Request(
        BASE + "/v1/chat/completions",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": "Bearer " + KEY,
                 "Content-Type": "application/json"}, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=60) as response:
            result = json.load(response)
    except urllib.error.HTTPError as error:
        raise RuntimeError(f"HTTP {error.code}: {error.read().decode('utf-8')}") from error
    choices = result.get("choices", [])
    if not choices or not isinstance(choices[0].get("message"), dict):
        raise RuntimeError("未返回有效消息，请检查错误信息与模型能力")
    if choices[0].get("finish_reason") == "length":
        raise RuntimeError("回答因输出上限而截断，请调整预算后重试")
    return choices[0]["message"]

tools = [{"type": "function", "function": {
    "name": "get_order_status", "description": "查询演示订单状态",
    "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}},
                   "required": ["order_id"], "additionalProperties": False},
}}]
messages = [{"role": "user", "content": "帮我查询订单 DEMO-1001 的状态。"}]
for _ in range(3):
    message = chat(messages, tools=tools)
    calls = message.get("tool_calls") or []
    if not calls:
        print(message.get("content") or "模型未提供文本回答")
        break
    messages.append({"role": "assistant", "content": message.get("content"), "tool_calls": calls})
    for call in calls:
        function = call.get("function", {})
        try:
            args = json.loads(function.get("arguments", "{}"))
        except (json.JSONDecodeError, TypeError):
            args = None
        if function.get("name") != "get_order_status":
            result = {"error": "未知工具"}
        elif (not isinstance(args, dict) or set(args) != {"order_id"}
              or not isinstance(args["order_id"], str)):
            result = {"error": "参数无效"}
        elif args["order_id"] != "DEMO-1001":
            result = {"error": "演示订单不存在"}
        else:
            result = {"order_id": "DEMO-1001", "status": "已发货", "demo": True}
        if not call.get("id"):
            raise RuntimeError("工具调用缺少 id，不能回传结果")
        messages.append({"role": "tool", "tool_call_id": call["id"],
                         "content": json.dumps(result, ensure_ascii=False)})
else:
    raise RuntimeError("达到工具调用次数上限，交由应用处理")