编写 API 代理：统一 Anthropic Messages、OpenAI Response、DeepSeek 格式

1. 统一中间格式设计

设计一个内部标准格式，能够表达三种 API 的公共能力：

字段	类型	说明
model	string	模型名称
messages	array	消息数组（user/assistant）
system	string (opt)	系统提示，独立字段便于转换
max_tokens	int	最大输出 token
temperature	float	随机性控制
top_p	float	核采样概率
top_k	int (opt)	Anthropic 专有
stop	list[str]	停止词
stream	bool	是否流式
tools	list	工具定义

2. 核心差异映射

内部格式	Anthropic Messages	OpenAI Response (Codex)	DeepSeek Chat
max_tokens	max_tokens	max_output_tokens	max_tokens
stop	stop_sequences	stop	stop
system	顶层参数	instructions	messages[0].role=system
top_k	支持	不支持	不支持
thinking	支持	不支持	不支持

3. 架构图

+-------------------------------------------------------------+
|                     Client Request                          |
|                POST /v1/unified/chat                       |
+---------------------------+---------------------------------+
                            |
                            v
+-------------------------------------------------------------+
|                   Unified API Layer                         |
|  +-------------+  +-------------+  +-------------+         |
|  |    Router   |  |  Validator  |  |  Translater |         |
+---------------------------+---------------------------------+
                            |
                            v
+-------------------------------------------------------------+
|                    Provider Adapters                        |
|  +----------+  +----------+  +----------+                  |
|  |Anthropic  |  |OpenAI Resp |  |DeepSeek   |              |
|  |          |  |(Codex)    |  |          |                  |
+---------------------------+---------------------------------+
                            |
                            v
+-------------------------------------------------------------+
|                   Provider APIs                             |
|  https://api.anthropic.com/v1/messages                     |
|  https://api.openai.com/v1/responses                       |
|  https://api.deepseek.com/v1/chat/completions              |
+-------------------------------------------------------------+

4. 路由配置 Schema

providers:
  - name: anthropic
    base_url: https://api.anthropic.com
    endpoint: /v1/messages
    models: [claude-*, claude-*]
    headers:
      x-api-key: ${ANTHROPIC_API_KEY}
      
  - name: openai_response
    base_url: https://api.openai.com/v1
    endpoint: /responses
    models: [o1-*, o3-*, computer-use-preview]
    headers:
      Authorization: Bearer ${OPENAI_API_KEY}
      
  - name: deepseek
    base_url: https://api.deepseek.com/v1
    endpoint: /chat/completions
    models: [deepseek-chat, deepseek-coder]
    headers:
      Authorization: Bearer ${DEEPSEEK_API_KEY}

5. 请求格式转换

def to_anthropic(req: UnifiedRequest) -> dict:
    body = {
        "model": req.model,
        "max_tokens": req.max_tokens,
        "messages": req.messages,
        "temperature": req.temperature,
        "top_p": req.top_p,
    }
    if req.system:
        body["system"] = req.system
    if req.stop:
        body["stop_sequences"] = req.stop
    if req.thinking:
        body["thinking"] = req.thinking
    return body

def to_openai_response(req: UnifiedRequest) -> dict:
    return {
        "model": req.model,
        "input": req.messages,
        "max_output_tokens": req.max_tokens,
        "temperature": req.temperature,
        "instructions": req.system,
        "stop": req.stop,
        "stream": req.stream,
        "tools": req.tools
    }

def to_deepseek(req: UnifiedRequest) -> dict:
    msgs = req.messages.copy()
    if req.system:
        msgs.insert(0, {"role": "system", "content": req.system})
    return {
        "model": req.model,
        "messages": msgs,
        "max_tokens": req.max_tokens,
        "temperature": req.temperature,
        "top_p": req.top_p,
        "stop": req.stop,
        "stream": req.stream
    }

6. 响应格式统一

def normalize_response(provider: str, resp: dict) -> dict:
    if provider == "anthropic":
        return {
            "id": resp.get("id"),
            "role": "assistant",
            "content": extract_text_content(resp.get("content", [])),
            "finish_reason": map_stop_reason(resp.get("stop_reason")),
            "usage": {
                "input_tokens": resp["usage"]["input_tokens"],
                "output_tokens": resp["usage"]["output_tokens"]
            }
        }
    else:
        choice = resp.get("choices", [{}])[0] if provider == "openai_chat" else resp.get("output", [{}])[0]
        return {
            "role": "assistant",
            "content": choice.get("text", "") or choice.get("content", ""),
            "finish_reason": choice.get("finish_reason") or choice.get("type"),
            "usage": {
                "input_tokens": resp["usage"]["prompt_tokens"] or resp["usage"]["input_tokens"],
                "output_tokens": resp["usage"]["completion_tokens"] or resp["usage"]["output_tokens"]
            }
        }

7. 模型路由逻辑

def select_provider(model: str) -> tuple[str, str]:
    """返回 (provider_type, api_style)"""
    if model.startswith("claude"):
        return ("anthropic", "messages")
    elif model.startswith("o") or "computer" in model:
        return ("openai", "response")
    elif model.startswith("gpt"):
        return ("openai", "chat")
    elif "deepseek" in model:
        return ("deepseek", "chat")
    raise ValueError(f"Unknown model: {model}")

# 用法
provider_type, api_style = select_provider("o1-preview")
# provider_type = "openai", api_style = "response"

8. 完整代码实现

from typing import Optional
import httpx
import os

class UnifiedRequest:
    def __init__(self, model: str, messages: list, max_tokens: int = 4096,
                 temperature: float = 1.0, top_p: float = 1.0,
                 system: str = None, stop: list = None,
                 stream: bool = False, tools: list = None, thinking: dict = None):
        self.model = model
        self.messages = messages
        self.max_tokens = max_tokens
        self.temperature = temperature
        self.top_p = top_p
        self.system = system
        self.stop = stop or []
        self.stream = stream
        self.tools = tools or []
        self.thinking = thinking

class UnifiedProxy:
    def __init__(self):
        self.providers = {
            "anthropic": {"base_url": "https://api.anthropic.com"},
            "openai": {"base_url": "https://api.openai.com/v1"},
            "deepseek": {"base_url": "https://api.deepseek.com/v1"}
        }
        self.client = httpx.AsyncClient(timeout=300)

    def _select_provider(self, model: str) -> tuple:
        if model.startswith("claude"):
            return ("anthropic", "messages")
        elif model.startswith("o") or "computer" in model:
            return ("openai", "response")
        elif model.startswith("gpt"):
            return ("openai", "chat")
        elif "deepseek" in model:
            return ("deepseek", "chat")
        raise ValueError(f"Unknown model: {model}")

    def _to_anthropic(self, req: UnifiedRequest) -> dict:
        body = {
            "model": req.model,
            "max_tokens": req.max_tokens,
            "messages": req.messages,
            "temperature": req.temperature,
            "top_p": req.top_p,
        }
        if req.system:
            body["system"] = req.system
        if req.stop:
            body["stop_sequences"] = req.stop
        if req.thinking:
            body["thinking"] = req.thinking
        return body

    def _to_openai_response(self, req: UnifiedRequest) -> dict:
        return {
            "model": req.model,
            "input": req.messages,
            "max_output_tokens": req.max_tokens,
            "temperature": req.temperature,
            "instructions": req.system,
            "stop": req.stop,
            "stream": req.stream,
            "tools": req.tools
        }

    def _to_openai_chat(self, req: UnifiedRequest) -> dict:
        msgs = req.messages.copy()
        if req.system:
            msgs.insert(0, {"role": "system", "content": req.system})
        return {
            "model": req.model,
            "messages": msgs,
            "max_tokens": req.max_tokens,
            "temperature": req.temperature,
            "top_p": req.top_p,
            "stop": req.stop,
            "stream": req.stream
        }

    def _to_deepseek(self, req: UnifiedRequest) -> dict:
        msgs = req.messages.copy()
        if req.system:
            msgs.insert(0, {"role": "system", "content": req.system})
        return {
            "model": req.model,
            "messages": msgs,
            "max_tokens": req.max_tokens,
            "temperature": req.temperature,
            "top_p": req.top_p,
            "stop": req.stop,
            "stream": req.stream
        }

    async def chat(self, request: UnifiedRequest) -> dict:
        provider, api_style = self._select_provider(request.model)
        config = self.providers["anthropic"] if provider == "anthropic" else self.providers[provider]
        adapter = getattr(self, f"_to_{provider}")
        payload = adapter(request)
        url = f"{config['base_url']}{config['endpoint']}"
        headers = {"Authorization": f"Bearer {os.getenv(provider.upper() + '_API_KEY')}"}
        if provider == "anthropic":
            headers["x-api-key"] = os.getenv("ANTHROPIC_API_KEY")
            headers["anthropic-version"] = "2023-06-01"
        resp = await self.client.post(url, json=payload, headers=headers)
        raw = resp.json()
        return self._normalize_response(provider, raw)

9. 使用示例

proxy = UnifiedProxy()

request = UnifiedRequest(
    model="claude-3-opus-20240229",
    system="You are helpful",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    max_tokens=1000
)

response = await proxy.chat(request)
print(response["content"])

10. 流式支持

SSE 响应标准化需要处理 content_block_delta、message_stop（Anthropic）和 choices[].delta（OpenAI/DeepSeek）两种格式，统一输出 {type: "content"|"done", delta: string}。

---

来源：官方 SDK 与 API 文档

---

如果你觉得本文有用，请点赞，收藏，转发

你有什么问题，请留言，我来帮你解答。