做 OpenAI / Anthropic 兼容层时，真正难的不是转发

# naive_proxy.py —— 看起来能用，流式和工具调用很快露馅
from flask import Flask, request, Response
import requests

app = Flask(__name__)
UPSTREAM = "https://api.openai.com"

@app.post("/v1/chat/completions")
def chat():
    r = requests.post(
        f"{UPSTREAM}/v1/chat/completions",
        headers={
            "Authorization": request.headers.get("Authorization", ""),
            "Content-Type": "application/json",
        },
        json=request.get_json(force=True),
        stream=True,
        timeout=120,
    )
    return Response(r.iter_content(chunk_size=1024), status=r.status_code,
                    content_type=r.headers.get("Content-Type", "application/json"))

data: {"id":"chatcmpl-x","object":"chat.completion.chunk","choices":[{"delta":{"content":"你"},"index":0}]}

data: {"id":"chatcmpl-x","object":"chat.completion.chunk","choices":[{"delta":{},"finish_reason":"stop"}]}

data: [DONE]

# sse_normalize.py
from __future__ import annotations
import json
from typing import Iterator

def openai_text_chunks(text: str, chunk_size: int = 8) -> Iterator[str]:
    """把完整文本伪造成 OpenAI SSE，便于本地联调客户端。"""
    cid = "chatcmpl-demo"
    for i in range(0, len(text), chunk_size):
        piece = text[i : i + chunk_size]
        payload = {
            "id": cid,
            "object": "chat.completion.chunk",
            "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}],
        }
        yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
    end = {
        "id": cid,
        "object": "chat.completion.chunk",
        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
    }
    yield f"data: {json.dumps(end)}\n\n"
    yield "data: [DONE]\n\n"

from openai import OpenAI

client = OpenAI(api_key="sk-test", base_url="http://127.0.0.1:8080/v1")
stream = client.chat.completions.create(
    model="demo",
    messages=[{"role": "user", "content": "hi"}],
    stream=True,
)
for event in stream:
    delta = event.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"search","arguments":"{\"q\":"}}]}}
{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"hugo\"}"}}]}}

# tool_call_buffer.py
from __future__ import annotations
import json
from dataclasses import dataclass, field

@dataclass
class ToolCallBuf:
    id: str | None = None
    name: str | None = None
    arguments: str = ""

@dataclass
class StreamToolAssembler:
    calls: dict[int, ToolCallBuf] = field(default_factory=dict)

    def feed(self, tool_calls: list[dict]) -> None:
        for tc in tool_calls:
            idx = tc.get("index", 0)
            buf = self.calls.setdefault(idx, ToolCallBuf())
            if tc.get("id"):
                buf.id = tc["id"]
            fn = tc.get("function") or {}
            if fn.get("name"):
                buf.name = fn["name"]
            if fn.get("arguments"):
                buf.arguments += fn["arguments"]

    def finished(self) -> list[dict]:
        out = []
        for idx, buf in sorted(self.calls.items()):
            # 参数必须是合法 JSON，否则不要冒充「兼容成功」
            json.loads(buf.arguments or "{}")
            out.append(
                {
                    "id": buf.id or f"call_{idx}",
                    "type": "function",
                    "function": {"name": buf.name, "arguments": buf.arguments},
                }
            )
        return out

# error_map.py
from dataclasses import dataclass

@dataclass
class PublicError:
    status: int
    code: str
    message: str
    retryable: bool

def map_upstream_error(status: int, body: str) -> PublicError:
    b = body.lower()
    if status in (401, 403):
        return PublicError(401, "unauthorized", "网关或上游鉴权失败", False)
    if status == 429 or "rate" in b:
        return PublicError(429, "rate_limited", "上游限流，请稍后或换模型", True)
    if status in (408, 504) or "timeout" in b:
        return PublicError(504, "timeout", "上游超时", True)
    if status in (529, 503) or "overload" in b:
        return PublicError(503, "upstream_unavailable", "上游过载", True)
    if "content" in b and ("filter" in b or "policy" in b):
        return PublicError(400, "content_filtered", "内容被安全策略拦截", False)
    return PublicError(502, "bad_gateway", "上游错误", True)

# mini_compat_app.py
from flask import Flask, request, Response, jsonify
import json
from sse_normalize import openai_text_chunks
from error_map import map_upstream_error

app = Flask(__name__)

@app.post("/v1/chat/completions")
def chat_completions():
    body = request.get_json(force=True) or {}
    messages = body.get("messages") or []
    stream = bool(body.get("stream"))
    # 这里应：鉴权虚拟 Key → 选上游 → 适配协议
    # 演示：固定回显最后一条 user 消息
    user = next((m["content"] for m in reversed(messages) if m.get("role") == "user"), "")
    answer = f"[compat-demo] 收到：{user[:200]}"

    if not stream:
        return jsonify(
            {
                "id": "chatcmpl-demo",
                "object": "chat.completion",
                "choices": [
                    {
                        "index": 0,
                        "message": {"role": "assistant", "content": answer},
                        "finish_reason": "stop",
                    }
                ],
                "usage": {"prompt_tokens": 10, "completion_tokens": 12, "total_tokens": 22},
            }
        )

    return Response(openai_text_chunks(answer), mimetype="text/event-stream")

@app.get("/healthz")
def healthz():
    return {"ok": True}