用 OpenRouter Provider Routing 限定数据收集策略

同一模型可能由多家供应商托管，数据是否用于训练的政策不同。OpenRouter 的 provider 对象可以指定 order、allow_fallbacks、data_collection。 这篇做 provider_client.py + mock：当 data_collection=deny 时跳过会收集数据的供应商。

先看清成品

• mock：order=[openai,together] 且 deny 时落到 together。
• 若只剩会收集数据的供应商且不允许 fallback → 404。2 项 pytest 通过。

合规提醒：使用自己的 OpenRouter 账号与 API Key，不共享；按各模型页面的许可与数据政策选用模型。价格以 openrouter.ai/models 为准，本文不写、不编造价格；本地预算脚本的单价需你自行传入。

运行说明：没有可用的 OpenRouter 付费 Key。真实运行过：客户端对本地 mock 的 fallback / provider 路由、预算脚本的 dry-usage，以及对 https://openrouter.ai 使用无效密钥得到 HTTP 401（2026-10-23，UTC+8）。未产生计费调用。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

93-openrouter-B-src/
├── provider_client.py
├── mock_provider.py
└── test_93b.py

第 1 步：客户端

#!/usr/bin/env python3
"""OpenRouter provider routing: order + data_collection (official provider object)."""
from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request


def chat(*, base_url, api_key, model, message, provider):
    url = f"{base_url.rstrip('/')}/api/v1/chat/completions"
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": message}],
        "provider": provider,
    }
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode(),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("message", nargs="?", default="ping")
    p.add_argument("--base", default=os.environ.get("OPENROUTER_BASE_URL", "https://openrouter.ai"))
    p.add_argument("--key", default=os.environ.get("OPENROUTER_API_KEY", ""))
    p.add_argument("--model", default="meta-llama/llama-3.1-8b-instruct")
    p.add_argument("--deny-collection", action="store_true")
    p.add_argument("--order", default="together,fireworks")
    args = p.parse_args(argv)
    if not args.key:
        print("设置 OPENROUTER_API_KEY", file=sys.stderr)
        return 2
    provider = {
        "order": [x.strip() for x in args.order.split(",") if x.strip()],
        "allow_fallbacks": True,
        "data_collection": "deny" if args.deny_collection else "allow",
    }
    r = chat(base_url=args.base, api_key=args.key, model=args.model, message=args.message, provider=provider)
    print("provider:", r.get("provider"))
    print(r["choices"][0]["message"]["content"])
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

API_KEY = "or-test-key"


class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        if self.path != "/api/v1/chat/completions":
            self.send_error(404)
            return
        if self.headers.get("Authorization") != f"Bearer {API_KEY}":
            self.send_response(401); self.end_headers(); return
        body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0)) or b"0"))
        provider = body.get("provider") or {}
        # Simulate: if data_collection=deny and order prefers a collecting provider only -> 404-like
        order = provider.get("order") or []
        allow_fb = provider.get("allow_fallbacks", True)
        dc = provider.get("data_collection", "allow")
        collecting = {"openai", "anthropic"}
        zdr_ok = {"together", "fireworks"}
        candidates = order or ["together", "openai"]
        chosen = None
        for p in candidates:
            if dc == "deny" and p in collecting:
                continue
            chosen = p
            break
        if not chosen and allow_fb and dc != "deny":
            chosen = "openai"
        if not chosen:
            self.send_response(404)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"error":{"message":"No providers meet data_collection=deny"}}')
            return
        resp = {
            "id": "gen-p",
            "model": body.get("model", "meta-llama/llama-3.1-8b-instruct"),
            "provider": chosen,
            "choices": [{"message": {"role": "assistant", "content": f"via:{chosen}"}}],
            "usage": {"prompt_tokens": 8, "completion_tokens": 3, "total_tokens": 11},
        }
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(resp).encode())


if __name__ == "__main__":
    import os
    HTTPServer(("127.0.0.1", int(os.environ.get("PORT", "3094"))), H).serve_forever()

第 3 步：测试

import threading
from http.server import HTTPServer
import mock_provider
from provider_client import chat

def serve():
    httpd = HTTPServer(("127.0.0.1", 0), mock_provider.H)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd, httpd.server_address[1]

def test_deny_skips_collecting():
    httpd, port = serve()
    try:
        r = chat(base_url=f"http://127.0.0.1:{port}", api_key=mock_provider.API_KEY,
                 model="m", message="x",
                 provider={"order": ["openai", "together"], "data_collection": "deny", "allow_fallbacks": True})
        assert r["provider"] == "together"
    finally:
        httpd.shutdown()

def test_deny_none_left():
    httpd, port = serve()
    try:
        try:
            chat(base_url=f"http://127.0.0.1:{port}", api_key=mock_provider.API_KEY,
                 model="m", message="x",
                 provider={"order": ["openai", "anthropic"], "data_collection": "deny", "allow_fallbacks": False})
            assert False
        except SystemExit as e:
            assert "404" in str(e)
    finally:
        httpd.shutdown()

第 4 步：运行

# 2026-10-23 12:37 UTC+8
{
  "id": "gen-p",
  "model": "m",
  "provider": "together",
  "choices": [
    {
      "message": {
        "role": "assistant",
        "content": "via:together"
      }
    }
  ],
  "usage": {
    "prompt_tokens": 8,
    "completion_tokens": 3,
    "total_tokens": 11
  }
}

$ python -m pytest -q
2 passed

怎么检查结果？

• deny 模式下 provider 不应是会收集数据的那家（mock 里为 openai）。
• 无可用供应商时应失败而不是默默 allow。

调试示例

deny 后列表为空。 测试 allow_fallbacks=False 且 order 全是 collecting 供应商时，mock 返回 404。生产里这比静默落到不合规供应商更安全：宁可失败，也不要违反数据政策。

常见问题

模板规则

• 敏感数据默认 data_collection=deny。
• 记录响应里的 provider。
• 政策以官网为准。

项目资源

选模型不只看价格，还要看数据怎么处理。