用契约路由在生产前验证免费模型端点

免费模型端点会在你不知情的情况下发生变化。响应可能看起来正确，但仍然违反集成契约。按契约路由，而不是凭感觉路由。本教程构建了一个小型路由器，在提升流量之前对基线端点和候选端点进行比较。

你不需要付费计算资源来进行冒烟测试。本地模拟和 Python 标准库就够了。

1. 先定义契约

选择一个你的应用已经依赖的工具调用。在这个例子中，模型必须返回一条审查评论。契约有五条规则。

• message.content 非空。
• arguments 可解析为 JSON。
• arguments 对象包含 file、line 和 comment。
• line 是大于零的整数。
• 看起来正常但没有返回工具调用的模型会违反契约。返回有效 JSON 但缺少 line 的模型也会违反契约。你检查的是形状，不是语气。

2. 构建契约路由器

将此文件保存为 canary_router.py。

#!/usr/bin/env python3
import json
import os
import sys
import time
import urllib.error
import urllib.request

REQUIRED_KEYS = ('file', 'line', 'comment')

def post_chat(url, key, payload):
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode(),
        headers={
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + key,
        },
        method='POST',
    )
    with urllib.request.urlopen(req, timeout=20) as resp:
        return resp.status, json.loads(resp.read().decode())

def extract_text(data):
    try:
        return data['choices'][0]['message']['content']
    except (KeyError, IndexError, TypeError):
        return ''

def extract_tool_args(data):
    try:
        calls = data['choices'][0]['message'].get('tool_calls') or []
    except (KeyError, IndexError, TypeError):
        return []
    out = []
    for call in calls:
        fn = call.get('function') or {}
        raw = fn.get('arguments', '{}')
        try:
            out.append(json.loads(raw))
        except (json.JSONDecodeError, TypeError):
            out.append({'__invalid_json__': True})
    return out

def validate(content, tool_args):
    if not content.strip():
        return False, 'empty content'
    if not tool_args:
        return False, 'no tool call'
    for args in tool_args:
        if args.get('__invalid_json__'):
            return False, 'invalid tool JSON'
        for key in REQUIRED_KEYS:
            if key not in args:
                return False, 'missing ' + key
        line = args.get('line')
        if not isinstance(line, int) or line < 1:
            return False, 'bad line'
    return True, 'ok'

def check(url, key, payload):
    started = time.time()
    try:
        status, data = post_chat(url, key, payload)
        content = extract_text(data)
        tool_args = extract_tool_args(data)
        ok, reason = validate(content, tool_args)
        return ok, reason, round(time.time() - started, 3), status
    except Exception as exc:
        return False, repr(exc), round(time.time() - started, 3), None

def main():
    if len(sys.argv) != 2:
        print('usage: python3 canary_router.py request.json')
        sys.exit(2)
    baseline = os.environ['BASELINE_URL']
    candidate = os.environ['CANDIDATE_URL']
    key = os.environ.get('API_KEY', '')
    payload = json.load(open(sys.argv[1]))
    b = check(baseline, key, payload)
    c = check(candidate, key, payload)
    print('endpoint  ok  reason  status  latency_s')
    print('baseline ', b[0], b[1], b[3], b[2])
    print('candidate', c[0], c[1], c[3], c[2])
    if b[0] and c[0]:
        print('VERDICT: PROMOTE')
    elif b[0] and not c[0]:
        print('VERDICT: FALLBACK')
    elif not b[0] and c[0]:
        print('VERDICT: QUARANTINE')
    else:
        print('VERDICT: STOP')

if __name__ == '__main__':
    main()

该路由器只使用 Python 标准库。它不调用任何供应商 SDK。这使得这个关卡易于审计。

3. 创建一个请求 fixture

用一个小 Python 代码片段编写 request.json。

python3 - <<'PY'
import json
json.dump({
    'model': 'candidate',
    'messages': [
        {'role': 'user', 'content': 'Post one review comment for the diff.'}
    ],
    'tools': [
        {
            'type': 'function',
            'function': {
                'name': 'post_review',
                'parameters': {
                    'type': 'object',
                    'properties': {
                        'file': {'type': 'string'},
                        'line': {'type': 'integer'},
                        'comment': {'type': 'string'}
                    },
                    'required': ['file', 'line', 'comment']
                }
            }
        }
    ]
}, open('request.json', 'w'), indent=2)
PY

然后用环境变量运行路由器。

export BASELINE_URL='https://baseline.example/v1/chat/completions'
export CANDIDATE_URL='https://candidate.example/v1/chat/completions'
export API_KEY='your-key'
python3 canary_router.py request.json

输出显示每个端点、通过/失败原因、HTTP 状态码和延迟。判决结果映射到一个动作。

4. 用本地模拟进行冒烟测试

在验证路由器之前不要花费真实的 token。将以下内容保存为 mock_server.py。

import json
from http.server import BaseHTTPRequestHandler, HTTPServer

def make_response(ok):
    if ok:
        return {
            'choices': [{
                'message': {
                    'content': 'ok',
                    'tool_calls': [{
                        'function': {
                            'name': 'post_review',
                            'arguments': json.dumps({
                                'file': 'app.py',
                                'line': 10,
                                'comment': 'check this path'
                            })
                        }
                    }]
                }
            }]
        }
    return {'choices': [{'message': {'content': '', 'tool_calls': []}}]}

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get('Content-Length', '0'))
        self.rfile.read(length)
        body = json.dumps(make_response(self.path == '/pass')).encode()
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

HTTPServer(('127.0.0.1', 8000), Handler).serve_forever()

运行模拟，然后将两个 URL 都指向它。

python3 mock_server.py &
export API_KEY=''
export BASELINE_URL='http://127.0.0.1:8000/pass'
export CANDIDATE_URL='http://127.0.0.1:8000/fail'
python3 canary_router.py request.json

预期判决是 FALLBACK。停止模拟并将候选切换到 /pass 以获得 PROMOTE。这证明了在你涉及真实端点之前，路由器是 fail closed 的。

5. 使用免费模型端点作为候选

披露：本文是 MonkeyCode 产品推广的一部分。在本教程中，我将 MonkeyCode 的免费模型访问和免费服务器选项视为运营商提供的可用性声明。路由器不依赖特定的模型名称、配额或硬件。

将候选指向一个免费的 OpenAI 兼容聊天补全 URL。将基线指向你当前的端点。将模拟作为 CI 中的回归 fixture 保留。

for fixture in fixtures/*.json; do
  BASELINE_URL=$BASELINE_URL CANDIDATE_URL=$CANDIDATE_URL API_KEY=$API_KEY python3 canary_router.py $fixture
done

每个 fixture 运行三到五次。免费模型端点是非确定性的。记录通过率，而不是单次通过。

契约检查是结构性的。它们不能证明评论是有用的。

对于非确定性模型，一次运行是不够的。

速率限制和临时中断看起来像候选失败。检查 status 和 latency 列。

路由器不执行工具调用。保持它们干运行。

恶意端点可以返回有效 JSON 但带有错误数据。这不是安全边界。

谁不应该使用这种方法

如果你需要来自免费基础设施的严格延迟 SLO，请跳过本文。不要发送私人代码或个人数据。不要将单次契约运行作为语义质量的证明。需要精确输出的团队应该运行带有人工审核的黄金集。

契约路由将不稳定的免费端点转变为受控实验。先构建契约。用本地模拟进行冒烟测试。然后比较真实的候选。用判决结果说话，而不是凭兴奋。

如果你想使用 MonkeyCode 的免费服务器选项，先从 /pass 模拟和一个 fixture 开始，再发送任何真实请求。