原文: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.arguments 是JSON 编码的字符串,不是对象——那些“好心”返回已解析对象的端点,会坑坏所有对它调用 json.loads 的客户端。第二,流式模式下,工具调用以片段形式到达,你要按 index 字段重新拼装,id 和 function.name 通常只在第一个片段中出现。如果一个端点每个片段都重发 id,或者只有一个调用时省略 index,那么你的朴素累加器能正常跑,但一旦模型发出两个并行调用就会失败。还要检查 finish_reason 是 tool_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_p、presence_penalty,以及 max_tokens 与 max_completion_tokens 的关系也一样。
6. `stop` 序列。 端点是否遵守 stop 字符串数组?返回内容里是包含停止序列还是剔除停止序列(参考实现是剔除)?finish_reason 是否返回 "stop"?相当多的兼容垫片(shim)是在完整生成完之后再做截断来实现 stop,这意味着你会为根本没看见的 token 付费——usage 数字会暴露这一点。
7. 用量统计(usage accounting)。 非流式响应应携带 usage.prompt_tokens、usage.completion_tokens、usage.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_chat 或 error_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)



