协议转换与流式输出 深度解析

11def anthropic_to_openai(body: dict, remote_model: str) -> dict:
12    messages = []
13    system = body.get("system")
14    if system:
15        messages.append({"role": "system", "content": content_text(system)})
16
17    for msg in body.get("messages", []):
18        role = msg.get("role", "user")
19        content = msg.get("content", "")
20        if isinstance(content, str):
21            messages.append({"role": role, "content": content})
22            continue
23
24        parts = []
25        tool_calls = []
26        for block in content:
27            kind = block.get("type")
28            if kind == "text":
29                parts.append({"type": "text", "text": block.get("text", "")})
30            elif kind == "image":
31                src = block.get("source", {})
32                if src.get("type") == "base64":
33                    parts.append({
34                        "type": "image_url",
35                        "image_url": {
36                            "url": f"data:{src.get('media_type','image/png')};base64,{src.get('data','')}"
37                        }
38                    })
39            elif kind == "tool_use":
40                tool_calls.append({
41                    "id": block.get("id"),
42                    "type": "function",
43                    "function": {
44                        "name": block.get("name"),
45                        "arguments": json.dumps(block.get("input", {}))
46                    }
47                })
48            elif kind == "tool_result":
49                messages.append({
50                    "role": "tool",
51                    "tool_call_id": block.get("tool_use_id"),
52                    "content": content_text(block.get("content", ""))
53                })
54
55        item = {"role": role, "content": parts}
56        if tool_calls:
57            item["tool_calls"] = tool_calls
58        messages.append(item)
59
60    out = {
61        "model": remote_model,
62        "messages": messages,
63        "stream": bool(body.get("stream")),
64        "max_tokens": body.get("max_tokens", 4096),
65    }
66    for key in ("temperature", "top_p", "stop"):
67        if key in body:
68            out[key] = body[key]
69
70    if body.get("tools"):
71        out["tools"] = [{
72            "type": "function",
73            "function": {
74                "name": t["name"],
75                "description": t.get("description", ""),
76                "parameters": t.get("input_schema", {}),
77            }
78        } for t in body["tools"]]
79
80    if body.get("tool_choice"):
81        out["tool_choice"] = body["tool_choice"]
82
83    return out

23class OpenAIChatAdapter(ProviderAdapter):
24    async def complete(self, ctx, payload):
25        headers = {"Content-Type": "application/json"}
26        if ctx.api_key:
27            headers["Authorization"] = f"Bearer {ctx.api_key}"
28        async with httpx.AsyncClient(timeout=ctx.timeout) as client:
29            response = await client.post(
30                ctx.base_url.rstrip("/") + "/chat/completions",
31                headers=headers,
32                json=payload,
33            )
34        if response.status_code >= 400:
35            raise RuntimeError(f"upstream HTTP {response.status_code}: {response.text[:1000]}")
36        return response.json()
37
38    async def stream(self, ctx, payload):
39        headers = {"Content-Type": "application/json"}
40        if ctx.api_key:
41            headers["Authorization"] = f"Bearer {ctx.api_key}"
42        payload = dict(payload)
43        payload["stream"] = True
44        async with httpx.AsyncClient(timeout=ctx.timeout) as client:
45            async with client.stream(
46                "POST",
47                ctx.base_url.rstrip("/") + "/chat/completions",
48                headers=headers,
49                json=payload,
50            ) as response:
51                if response.status_code >= 400:
52                    text = (await response.aread()).decode(errors="replace")
53                    raise RuntimeError(f"upstream HTTP {response.status_code}: {text[:1000]}")
54                async for line in response.aiter_lines():
55                    if not line.startswith("data:"):
56                        continue
57                    value = line[5:].strip()
58                    if value == "[DONE]":
59                        return
60                    try:
61                        import json
62                        yield json.loads(value)
63                    except ValueError:
64                        continue

1import json
2
3def sse(event: str, data: dict) -> str:
4    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
5
6def openai_delta_to_anthropic(delta: dict, index: int = 0):
7    events = []
8    if delta.get("content"):
9        events.append(sse("content_block_delta", {
10            "type": "content_block_delta",
11            "index": index,
12            "delta": {"type": "text_delta", "text": delta["content"]},
13        }))
14    # Preserve provider reasoning when it is exposed in a common field.
15    reasoning = delta.get("reasoning_content") or delta.get("reasoning")
16    if reasoning:
17        events.append(sse("content_block_delta", {
18            "type": "content_block_delta",
19            "index": index,
20            "delta": {"type": "thinking_delta", "thinking": reasoning},
21        }))
22    return events