「OpenAI 兼容」不是布尔值,而是光谱:11 项一致性检查清单

原文:https://dev.to/seven7763/openai-compatible-is-a-spectrum-not-a-boolean-heres-an-11-check-conformance-suite-2mhj(作者 @seven7763)

两个端点都宣称自己“OpenAI 兼容”。把应用指向第一个,一切正常;指向第二个,一周内也一切正常。然后某次流式工具调用返回时,arguments 被拆到多个 chunk 里,拆法完全超出你累加器的预期;JSON 解析器在重试循环里抛出异常;重试循环又因为错误响应体缺少退避代码要读的字段,开始疯狂地打端点。

没有人骗你。“OpenAI 兼容”从来不是一个布尔值。它是一个覆盖面,每个实现只覆盖其中不同的子集。

兼容性网关的维护者会读到大量这类 bug 报告。下面的清单特意写成可以拿它来测任何端点;文末脚本不知道也不在乎你指向的是哪个端点。如果某个端点在某个检查项上表现很难看,那正是应有的输出。

九个兼容面

1. 流式增量(delta)。 参考行为很明确:第一个 chat.completion.chunk 携带 choices[0].delta.role = "assistant",通常内容为空;中间的 chunk 携带 delta.content 片段;最后一个携带内容的 chunk 带 finish_reason;流以一行字面量 data: [DONE] 结束。实现会在四个地方出现分歧。有的从不发送携带 role 的 chunk,这会让靠它打开消息的客户端挂掉。有的把 finish_reason 放在 delta 为空的末尾 chunk 上。有的干脆省略 [DONE] 直接关连接——对读到 EOF 就结束的客户端没问题,对阻塞等待哨兵的客户端则是致命的。

2. 工具/函数调用(tool / function calling)。 这里有两个坑。第一,function.argumentsJSON 编码的字符串,不是对象——那些“好心”返回已解析对象的端点,会坑坏所有对它调用 json.loads 的客户端。第二,流式模式下,工具调用以片段形式到达,你要按 index 字段重新拼装,idfunction.name 通常只在第一个片段中出现。如果一个端点每个片段都重发 id,或者只有一个调用时省略 index,那么你的朴素累加器能正常跑,但一旦模型发出两个并行调用就会失败。还要检查 finish_reasontool_calls 而不是 stop——智能体循环靠这个值分支。

3. `response_format`。 有三个层级,而且经常被混为一谈:完全不支持;{"type": "json_object"}(合法 JSON,形状随意);{"type": "json_schema", "json_schema": {..., "strict": true}}(按你的 schema 做约束解码)。危险的是中间那种情况:端点接受参数但忽略它。你会得到一段被代码围栏包着的普通文本,解析器每五十个请求挂一次,看起来还像是模型质量问题。

4. `logprobs` / `top_logprobs`。 这通常是代理层最先丢弃的字段,因为几乎没人注意。如果你靠比较 token 概率做分类,或者用 logprobs 做置信度门控,那它就是承重墙,应该显式测试。

5. `temperature` 与采样参数。 推理型模型在上游那里有时直接拒绝 temperature,有时接受但忽略,有时又真的按它生效。三种行为都说得过去;不知道自己面对的是哪一种才说不过去。top_ppresence_penalty,以及 max_tokensmax_completion_tokens 的关系也一样。

6. `stop` 序列。 端点是否遵守 stop 字符串数组?返回内容里是包含停止序列还是剔除停止序列(参考实现是剔除)?finish_reason 是否返回 "stop"?相当多的兼容垫片(shim)是在完整生成完之后再做截断来实现 stop,这意味着你会为根本没看见的 token 付费——usage 数字会暴露这一点。

7. 用量统计(usage accounting)。 非流式响应应携带 usage.prompt_tokensusage.completion_tokensusage.total_tokens。流式响应只有在你传 stream_options: {"include_usage": true} 时才会包含 usage,而且它会出现在最后一个 choices 数组为的 chunk 里——这种结构会让假设 choices[0] 一定存在的客户端崩溃。如果你按请求做成本分摊,还要检查缓存提示词 token 和推理 token 的明细是否还能拿到。

8. 错误响应体结构。 你写过的每个重试层都在解析这个字段,却没有人去测它。参考格式是 {"error": {"message": ..., "type": ..., "param": ..., "code": ...}},并配上语义匹配的 HTTP 状态码。现实中的变体:返回 200 但 body 里带 error 对象(对重试逻辑是致命的——你会高高兴兴地把一条错误字符串返回给用户);中间代理返回 HTML 错误页;本应 400 的地方返回 500,把永久性的客户端错误变成无限重试风暴。

9. `/v1/models` 的忠实度。 它存在吗?返回的是 {"object": "list", "data": [{"id": ...}]} 吗?以及最关键的一点:它列出的是你的 key 真正能调用的模型 ID 吗?如果目录端点返回的是厂商在售的所有模型,而不是你的 key 有权使用的模型,那还不如没有目录,因为你会在它之上构建 CI 检查。

脚本

仅用标准库,Python 3.8+,无需安装。脚本会运行十一项检查并输出一张结论表。

#!/usr/bin/env python3
"""compat_probe.py — 这个端点到底有多兼容 OpenAI?

    export LLM_BASE_URL=https://api.example.com/v1
    export LLM_API_KEY=sk-...
    python3 compat_probe.py --model <exact-model-id>
"""
import argparse, json, os, sys, urllib.error, urllib.request

BASE = os.environ.get("LLM_BASE_URL", "").rstrip("/")
KEY = os.environ.get("LLM_API_KEY", "")
OUT = []

def _open(path, payload=None, method="GET", timeout=90):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(BASE + path, data=data, method=method)
    req.add_header("Authorization", "Bearer " + KEY)
    if data is not None:
        req.add_header("Content-Type", "application/json")
    try:
        return urllib.request.urlopen(req, timeout=timeout)
    except urllib.error.HTTPError as exc:   # 仍然是可读的文件对象
        return exc

def call(path, payload=None, method="GET"):
    resp = _open(path, payload, method)
    raw = resp.read().decode("utf-8", "replace")
    try:
        return resp.getcode(), json.loads(raw)
    except ValueError:
        return resp.getcode(), raw

def stream(payload):
    payload = dict(payload, stream=True)
    resp = _open("/chat/completions", payload, "POST")
    chunks, saw_done = [], False
    for line in resp:
        line = line.decode("utf-8", "replace").strip()
        if not line.startswith("data:"):
            continue
        body = line[5:].strip()
        if body == "[DONE]":
            saw_done = True
            break
        try:
            chunks.append(json.loads(body))
        except ValueError:
            pass
    return chunks, saw_done

def record(name, ok, detail):
    OUT.append((name, "PASS" if ok is True else "FAIL" if ok is False else "PARTIAL", detail))

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--model", required=True)
    model = ap.parse_args().model
    ask = {"model": model, "messages": [{"role": "user", "content": "Say hi."}],
           "max_tokens": 24}

    # 1 — 模型目录
    code, body = call("/models")
    ids = [m.get("id") for m in body.get("data", [])] if isinstance(body, dict) else []
    record("models_endpoint", code == 200 and bool(ids),
           f"HTTP {code}, {len(ids)} ids, target listed: {model in ids}")

    # 2 — 基础补全 + 模型回显 + usage 统计
    code, body = call("/chat/completions", ask, "POST")
    msg = (body.get("choices") or [{}])[0].get("message", {}) if isinstance(body, dict) else {}
    usage = body.get("usage") if isinstance(body, dict) else None
    record("basic_chat", bool(msg.get("content")), f"HTTP {code}")
    record("model_echo", isinstance(body, dict) and body.get("model") == model,
           f"asked {model!r}, got {body.get('model')!r}" if isinstance(body, dict) else "n/a")
    record("usage_fields", bool(usage and usage.get("total_tokens") is not None), str(usage))

    # 3 — 流式返回形态
    chunks, done = stream(ask)
    role = any((c.get("choices") or [{}])[0].get("delta", {}).get("role") for c in chunks)
    fin = any((c.get("choices") or [{}])[0].get("finish_reason") for c in chunks)
    record("stream_shape", bool(chunks) and role and fin and done,
           f"{len(chunks)} chunks, role_chunk={role}, finish_reason={fin}, [DONE]={done}")

    # 4 — 流式返回中的 usage
    chunks, _ = stream(dict(ask, stream_options={"include_usage": True}))
    record("stream_usage", any(c.get("usage") for c in chunks),
           "final usage chunk present" if any(c.get("usage") for c in chunks) else "absent")

    # 5 — 工具调用
    tool = {"type": "function", "function": {"name": "get_weather", "parameters": {
        "type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}}
    code, body = call("/chat/completions", dict(
        ask, messages=[{"role": "user", "content": "Weather in Osaka? Use the tool."}],
        tools=[tool], tool_choice="auto"), "POST")
    choice = (body.get("choices") or [{}])[0] if isinstance(body, dict) else {}
    calls = choice.get("message", {}).get("tool_calls") or []
    args_is_str = bool(calls) and isinstance(calls[0].get("function", {}).get("arguments"), str)
    record("tool_calls", bool(calls) and args_is_str and choice.get("finish_reason") == "tool_calls",
           f"n={len(calls)} arguments_is_string={args_is_str} finish={choice.get('finish_reason')}")

    # 6 — JSON 模式
    code, body = call("/chat/completions", dict(
        ask, messages=[{"role": "user", "content": "Return {\"ok\": true} as JSON."}],
        response_format={"type": "json_object"}), "POST")
    text = ((body.get("choices") or [{}])[0].get("message", {}).get("content")
            if isinstance(body, dict) else "") or ""
    try:
        json.loads(text); parsed = True
    except ValueError:
        parsed = False
    record("json_object", code == 200 and parsed, f"HTTP {code}, parses={parsed}")

    # 7 — 严格 schema
    schema = {"type": "json_schema", "json_schema": {"name": "r", "strict": True, "schema": {
        "type": "object", "properties": {"ok": {"type": "boolean"}},
        "required": ["ok"], "additionalProperties": False}}}
    code, _ = call("/chat/completions", dict(ask, response_format=schema), "POST")
    record("json_schema_strict", code == 200, f"HTTP {code}")

# 8 — logprobs
    code, body = call("/chat/completions", dict(ask, logprobs=True, top_logprobs=3), "POST")
    lp = ((body.get("choices") or [{}])[0].get("logprobs") if isinstance(body, dict) else None)
    record("logprobs", bool(lp and lp.get("content")), f"HTTP {code}")

    # 9 — temperature:接受 / 拒绝 / 忽略是三种完全不同的世界
    code, _ = call("/chat/completions", dict(ask, temperature=0.5), "POST")
    record("temperature", None if code == 400 else code == 200,
           "rejected with 400 (reasoning model?)" if code == 400 else f"HTTP {code}")

    # 10 — 停止序列
    code, body = call("/chat/completions", dict(
        ask, messages=[{"role": "user", "content": "Count: one two three four five"}],
        stop=["three"], max_tokens=48), "POST")
    text = ((body.get("choices") or [{}])[0].get("message", {}).get("content")
            if isinstance(body, dict) else "") or ""
    record("stop_sequences", "three" not in text, f"stop string leaked into content: {'three' in text}")

    # 11 — 针对不存在的模型的错误结构
    code, body = call("/chat/completions", dict(ask, model="definitely-not-a-model-xyz"), "POST")
    shaped = isinstance(body, dict) and isinstance(body.get("error"), dict)
    record("error_shape", 400 <= code < 500 and shaped, f"HTTP {code}, error object: {shaped}")

    width = max(len(n) for n, _, _ in OUT)
    for name, verdict, detail in OUT:
        print(f"{name.ljust(width)}  {verdict:<7} {detail}")
    print("\nPARTIAL is not a failure — it is a behaviour you now have to design around.")
    return 0 if all(v != "FAIL" for _, v, _ in OUT) else 1

if __name__ == "__main__":
    sys.exit(main())


运行这个脚本有两点说明。它会消耗少量微小的补全请求,所以先用一个便宜的模型跑一遍,确认你的 base URL 和密钥正确。然后再用你计划上线的那个精确模型 ID 跑一次——因为同一端点在不同模型上的符合性各不相同,推理模型尤其会拒绝它们的非推理兄弟模型所接受的参数。

阅读输出

basic_chaterror_shape 出现 FAIL 是真正的问题。logprobs 出现 FAIL 只在你确实用到 logprobs 时才成问题。这张表的意义不是打分,而是让你有了一份书面记录,写明哪些接口面可以放心依赖;供应商一有变动,你就能重新运行并 diff。

由此可以养成三个习惯:

  • 把它纳入 CI。 每晚运行一次,把输出作为构建产物提交。兼容性回归天生是无声的——没有人会发一条变更日志说“我们停止转发 logprobs 了”。
  • 在生产环境断言模型回显,而不只在探针里。如果你请求的是一个 ID,响应的 model 字段却说是另一个,你希望这种事在发生当天就出现在日志里,而不是一周后你注意到质量下滑时才察觉。
  • 把 `PARTIAL` 当作设计输入。 如果 stream_usage 缺失,你的成本归因就需要一条非流式路径或本地 tokenizer 估算。这是现在花两小时就能定下来的事,否则日后就是月底对账时的不解之谜。

它无法告诉你的事

探针测试的是协议,而不是其背后的模型。一个端点可能通过全部十一项检查,却仍然把你路由到更小的模型、量化构建或截断的上下文窗口——两种情况下线上的数据格式都一样完美。那是另一项需要不同工具的调查,诚实的说法是:行为探针给你的是信号,而非证据——供应商之外没人能看到到底是哪些权重为你的请求提供了服务。我把这个问题的目录侧单独写了出来——固定模型 ID、检测漂移,以及当 model 字段不再匹配你所请求的模型时该记录什么——见 模型 ID 是一种依赖,请像对待依赖一样钉住它

但协议符合性是你能用一个下午、一屏脚本就解决的部分,而如果你跳过它,它也是会在凌晨三点把你惊醒的部分。


本文是“验证你的端点”系列文章之一。关于模型 ID 漂移的配套文章见 [模型 ID 是一种依赖,请像对待依赖一样钉住它](https://dev.to/seven7763/model-ids-are-a-dependency-pin-them-like-one-db0);如果你在第一次生产调用前正在评估第三方端点,[在将生产流量路由到别人的 LLM 端点之前要回答的十个问题](https://dev.to/seven7763/ten-questions-to-answer-before-you-route-production-traffic-through-someone-elses-llm-endpoint-oml) 是买方侧的检查清单。

原文:https://dev.to/seven7763/openai-compatible-is-a-spectrum-not-a-boolean-heres-an-11-check-conformance-suite-2mhj(作者 @seven7763)

发布评论
全部评论(0)