大模型协议转换指南

由于 OpenAI API 格式已成为事实标准，许多应用和框架都基于该格式开发。本文介绍如何在不同厂商 API 之间进行协议转换。

为什么需要协议转换？

在实际开发中，你可能需要同时使用多个大模型厂商的服务。协议转换让你可以用统一的接口调用不同的模型。

协议转换的核心价值：

flowchart LR
    A[应用代码] --> B[统一接口]
    B --> C[协议转换层]
    C --> D[OpenAI]
    C --> E[Claude]
    C --> F[Gemini]
    C --> G[国产模型]

应用代码只需要对接一种格式（通常是 OpenAI 格式），协议转换层负责适配不同厂商。

转换方案对比

选择建议：

格式对比速查

理解各厂商格式差异是协议转换的基础。

请求结构对比

关键差异说明：

1. 系统指令位置：OpenAI 放在 messages 数组中，Claude 是独立参数，Gemini 是嵌套对象
2. max_tokens：Claude 必需，OpenAI 可选
3. 流式端点：Gemini 使用独立端点，其他用参数控制
4. 参数命名：Gemini 使用 camelCase，其他用 snake_case

消息格式对比

内容结构差异：

# OpenAI: content 可以是字符串或数组
{"role": "user", "content": "Hello"}
{"role": "user", "content": [{"type": "text", "text": "Hello"}]}

# Claude: content 可以是字符串或数组
{"role": "user", "content": "Hello"}
{"role": "user", "content": [{"type": "text", "text": "Hello"}]}

# Gemini: 必须使用 parts 数组
{"role": "user", "parts": [{"text": "Hello"}]}

响应结构对比

提取内容的代码：

def extract_content(response, provider):
    if provider == "openai":
        return response["choices"][0]["message"]["content"]
    elif provider == "claude":
        return response["content"][0]["text"]
    elif provider == "gemini":
        return response["candidates"][0]["content"]["parts"][0]["text"]

手动转换实现

手动转换虽然繁琐，但能让你完全理解格式差异，也适合简单场景。

OpenAI → Claude

def openai_to_claude(openai_request: dict) -> dict:
    """将 OpenAI 格式转换为 Claude 格式"""
    claude_request = {
        "model": map_model(openai_request["model"]),
        "max_tokens": openai_request.get("max_tokens", 4096),  # Claude 必需
        "messages": [],
    }

    # 提取系统消息（Claude 需要放在顶层）
    system_messages = []
    for msg in openai_request["messages"]:
        if msg["role"] == "system":
            system_messages.append(msg["content"])
        else:
            claude_request["messages"].append({
                "role": msg["role"],
                "content": msg["content"]
            })

    if system_messages:
        claude_request["system"] = "\n".join(system_messages)

    # 转换可选参数
    if "temperature" in openai_request:
        claude_request["temperature"] = openai_request["temperature"]

    if "stream" in openai_request:
        claude_request["stream"] = openai_request["stream"]

    # 转换工具（注意字段名差异）
    if "tools" in openai_request:
        claude_request["tools"] = [
            {
                "name": tool["function"]["name"],
                "description": tool["function"].get("description", ""),
                "input_schema": tool["function"]["parameters"]  # Claude 用 input_schema
            }
            for tool in openai_request["tools"]
        ]

    return claude_request

def map_model(openai_model: str) -> str:
    """模型名称映射"""
    mapping = {
        "gpt-4o": "claude-sonnet-4-20250514",
        "gpt-4-turbo": "claude-sonnet-4-20250514",
        "gpt-4": "claude-3-opus-20240229",
        "gpt-3.5-turbo": "claude-3-5-haiku-20241022",
    }
    return mapping.get(openai_model, "claude-sonnet-4-20250514")

Claude → OpenAI

def claude_to_openai_response(claude_response: dict) -> dict:
    """将 Claude 响应转换为 OpenAI 格式"""
    # 提取文本内容和工具调用
    content = ""
    tool_calls = []

    for block in claude_response.get("content", []):
        if block["type"] == "text":
            content += block["text"]
        elif block["type"] == "tool_use":
            tool_calls.append({
                "id": block["id"],
                "type": "function",
                "function": {
                    "name": block["name"],
                    "arguments": json.dumps(block["input"])  # Claude 是对象，OpenAI 要字符串
                }
            })

    # 构建 OpenAI 格式响应
    message = {"role": "assistant", "content": content or None}
    if tool_calls:
        message["tool_calls"] = tool_calls

    return {
        "id": claude_response["id"],
        "object": "chat.completion",
        "created": int(time.time()),
        "model": claude_response["model"],
        "choices": [{
            "index": 0,
            "message": message,
            "finish_reason": map_stop_reason(claude_response["stop_reason"])
        }],
        "usage": {
            "prompt_tokens": claude_response["usage"]["input_tokens"],
            "completion_tokens": claude_response["usage"]["output_tokens"],
            "total_tokens": (
                claude_response["usage"]["input_tokens"] +
                claude_response["usage"]["output_tokens"]
            )
        }
    }

def map_stop_reason(claude_reason: str) -> str:
    """停止原因映射"""
    mapping = {
        "end_turn": "stop",
        "max_tokens": "length",
        "tool_use": "tool_calls",
        "stop_sequence": "stop"
    }
    return mapping.get(claude_reason, "stop")

OpenAI → Gemini

def openai_to_gemini(openai_request: dict) -> dict:
    """将 OpenAI 格式转换为 Gemini 格式"""
    gemini_request = {
        "contents": [],
        "generationConfig": {}
    }

    # 转换消息
    for msg in openai_request["messages"]:
        if msg["role"] == "system":
            gemini_request["systemInstruction"] = {
                "parts": [{"text": msg["content"]}]
            }
        else:
            role = "model" if msg["role"] == "assistant" else "user"
            gemini_request["contents"].append({
                "role": role,
                "parts": [{"text": msg["content"]}]
            })

    # 转换生成配置
    if "temperature" in openai_request:
        gemini_request["generationConfig"]["temperature"] = openai_request["temperature"]

    if "max_tokens" in openai_request:
        gemini_request["generationConfig"]["maxOutputTokens"] = openai_request["max_tokens"]

    if "top_p" in openai_request:
        gemini_request["generationConfig"]["topP"] = openai_request["top_p"]

    # 转换工具
    if "tools" in openai_request:
        gemini_request["tools"] = [{
            "functionDeclarations": [
                {
                    "name": tool["function"]["name"],
                    "description": tool["function"].get("description", ""),
                    "parameters": tool["function"]["parameters"]
                }
                for tool in openai_request["tools"]
            ]
        }]

    return gemini_request

Gemini → OpenAI

def gemini_to_openai_response(gemini_response: dict) -> dict:
    """将 Gemini 响应转换为 OpenAI 格式"""
    candidate = gemini_response["candidates"][0]
    content = candidate["content"]

    # 提取文本和工具调用
    text_content = ""
    tool_calls = []

    for part in content.get("parts", []):
        if "text" in part:
            text_content += part["text"]
        elif "functionCall" in part:
            tool_calls.append({
                "id": f"call_{uuid.uuid4().hex[:8]}",
                "type": "function",
                "function": {
                    "name": part["functionCall"]["name"],
                    "arguments": json.dumps(part["functionCall"]["args"])
                }
            })

    message = {"role": "assistant", "content": text_content or None}
    if tool_calls:
        message["tool_calls"] = tool_calls

    # 映射结束原因
    finish_reason_map = {
        "STOP": "stop",
        "MAX_TOKENS": "length",
        "SAFETY": "content_filter"
    }

    usage = gemini_response.get("usageMetadata", {})

    return {
        "id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": gemini_response.get("modelVersion", "gemini"),
        "choices": [{
            "index": 0,
            "message": message,
            "finish_reason": finish_reason_map.get(
                candidate.get("finishReason", "STOP"), "stop"
            )
        }],
        "usage": {
            "prompt_tokens": usage.get("promptTokenCount", 0),
            "completion_tokens": usage.get("candidatesTokenCount", 0),
            "total_tokens": usage.get("totalTokenCount", 0)
        }
    }

流式响应转换

Claude SSE → OpenAI SSE

async def convert_claude_stream_to_openai(claude_stream):
    """将 Claude 流式响应转换为 OpenAI 格式"""
    response_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"

    async for line in claude_stream:
        if not line.startswith("data: "):
            continue

        data = json.loads(line[6:])
        event_type = data.get("type")

        if event_type == "content_block_delta":
            delta = data.get("delta", {})
            if delta.get("type") == "text_delta":
                yield f"data: {json.dumps({
                    'id': response_id,
                    'object': 'chat.completion.chunk',
                    'choices': [{
                        'index': 0,
                        'delta': {'content': delta['text']},
                        'finish_reason': None
                    }]
                })}\n\n"

        elif event_type == "message_delta":
            stop_reason = data.get("delta", {}).get("stop_reason")
            yield f"data: {json.dumps({
                'id': response_id,
                'object': 'chat.completion.chunk',
                'choices': [{
                    'index': 0,
                    'delta': {},
                    'finish_reason': map_stop_reason(stop_reason)
                }]
            })}\n\n"

        elif event_type == "message_stop":
            yield "data: [DONE]\n\n"

Gemini Stream → OpenAI SSE

async def convert_gemini_stream_to_openai(gemini_stream):
    """将 Gemini 流式响应转换为 OpenAI 格式"""
    response_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"

    async for line in gemini_stream:
        if not line.strip():
            continue

        data = json.loads(line)
        candidate = data.get("candidates", [{}])[0]
        content = candidate.get("content", {})

        for part in content.get("parts", []):
            if "text" in part:
                yield f"data: {json.dumps({
                    'id': response_id,
                    'object': 'chat.completion.chunk',
                    'choices': [{
                        'index': 0,
                        'delta': {'content': part['text']},
                        'finish_reason': None
                    }]
                })}\n\n"

        if "finishReason" in candidate:
            yield f"data: {json.dumps({
                'id': response_id,
                'object': 'chat.completion.chunk',
                'choices': [{
                    'index': 0,
                    'delta': {},
                    'finish_reason': 'stop'
                }]
            })}\n\n"
            yield "data: [DONE]\n\n"

开源转换工具

LiteLLM

最流行的多模型代理，支持 100+ 模型提供商。

pip install litellm

from litellm import completion

# 统一接口调用不同模型
response = completion(
    model="claude-3-5-sonnet-20241022",  # 或 gpt-4o, gemini/gemini-pro
    messages=[{"role": "user", "content": "Hello!"}]
)
print(response.choices[0].message.content)

作为代理服务器

litellm --model claude-3-5-sonnet-20241022 --port 8000

# 使用 OpenAI 格式调用
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "claude-3-5-sonnet-20241022",
    "messages": [{"role": "user", "content": "Hello"}]
  }'

GitHub: https://github.com/BerriAI/litellm

One API

国内流行的 API 管理和分发系统。

docker run -d -p 3000:3000 \
  -e SQL_DSN="root:123456@tcp(localhost:3306)/oneapi" \
  justsong/one-api

特点：

• 支持多种模型提供商
• 统一 OpenAI 格式
• 支持负载均衡
• 支持配额管理

GitHub: https://github.com/songquanpeng/one-api

OpenRouter

商业 API 聚合服务，提供统一接口。

from openai import OpenAI

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key="your-openrouter-key"
)

response = client.chat.completions.create(
    model="anthropic/claude-3.5-sonnet",  # 或 google/gemini-pro
    messages=[{"role": "user", "content": "Hello!"}]
)

官网: https://openrouter.ai

Portkey

企业级 AI 网关。

from portkey_ai import Portkey

portkey = Portkey(
    api_key="your-portkey-key",
    virtual_key="your-anthropic-virtual-key"
)

response = portkey.chat.completions.create(
    model="claude-3-5-sonnet-20241022",
    messages=[{"role": "user", "content": "Hello!"}]
)

官网: https://portkey.ai

自建代理服务

FastAPI 实现

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI()

PROVIDERS = {
    "openai": {
        "base_url": "https://api.openai.com/v1",
        "auth_header": "Authorization",
        "auth_prefix": "Bearer "
    },
    "anthropic": {
        "base_url": "https://api.anthropic.com/v1",
        "auth_header": "x-api-key",
        "auth_prefix": ""
    },
    "gemini": {
        "base_url": "https://generativelanguage.googleapis.com/v1beta",
        "auth_header": None,  # 使用 URL 参数
        "auth_prefix": ""
    }
}

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    model = body.get("model", "")

    # 路由到对应提供商
    if model.startswith("gpt"):
        return await proxy_openai(body)
    elif model.startswith("claude"):
        return await proxy_claude(body)
    elif model.startswith("gemini"):
        return await proxy_gemini(body)
    else:
        return {"error": "Unknown model"}

async def proxy_claude(openai_request: dict):
    """代理到 Claude 并转换格式"""
    claude_request = openai_to_claude(openai_request)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.anthropic.com/v1/messages",
            json=claude_request,
            headers={
                "x-api-key": ANTHROPIC_API_KEY,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json"
            }
        )

    claude_response = response.json()
    return claude_to_openai_response(claude_response)

async def proxy_claude_stream(openai_request: dict):
    """流式代理到 Claude"""
    claude_request = openai_to_claude(openai_request)
    claude_request["stream"] = True

    async def generate():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                "https://api.anthropic.com/v1/messages",
                json=claude_request,
                headers={
                    "x-api-key": ANTHROPIC_API_KEY,
                    "anthropic-version": "2023-06-01"
                }
            ) as response:
                async for chunk in convert_claude_stream_to_openai(
                    response.aiter_lines()
                ):
                    yield chunk

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

工具调用转换

OpenAI Tool → Claude Tool

def convert_openai_tool_to_claude(openai_tool: dict) -> dict:
    """转换 OpenAI 工具定义为 Claude 格式"""
    return {
        "name": openai_tool["function"]["name"],
        "description": openai_tool["function"].get("description", ""),
        "input_schema": openai_tool["function"]["parameters"]
    }

def convert_openai_tool_call_to_claude(tool_call: dict) -> dict:
    """转换 OpenAI 工具调用为 Claude 格式"""
    return {
        "type": "tool_use",
        "id": tool_call["id"],
        "name": tool_call["function"]["name"],
        "input": json.loads(tool_call["function"]["arguments"])
    }

def convert_openai_tool_result_to_claude(tool_call_id: str, result: str) -> dict:
    """转换 OpenAI 工具结果为 Claude 格式"""
    return {
        "type": "tool_result",
        "tool_use_id": tool_call_id,
        "content": result
    }

Claude Tool → OpenAI Tool

def convert_claude_tool_to_openai(claude_tool: dict) -> dict:
    """转换 Claude 工具定义为 OpenAI 格式"""
    return {
        "type": "function",
        "function": {
            "name": claude_tool["name"],
            "description": claude_tool.get("description", ""),
            "parameters": claude_tool["input_schema"]
        }
    }

def convert_claude_tool_use_to_openai(tool_use: dict) -> dict:
    """转换 Claude 工具使用为 OpenAI 格式"""
    return {
        "id": tool_use["id"],
        "type": "function",
        "function": {
            "name": tool_use["name"],
            "arguments": json.dumps(tool_use["input"])
        }
    }

注意事项

⚠️ 协议转换不是万能的，需要注意以下差异。

功能差异

参数范围差异

转换时的常见问题

调试技巧

1. 记录原始请求/响应：方便排查转换问题
2. 使用 JSON Schema 验证：确保转换后格式正确
3. 单元测试覆盖：为每种转换编写测试用例

# 调试日志示例
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

def convert_with_logging(request):
    logger.debug(f"Original request: {request}")
    converted = openai_to_claude(request)
    logger.debug(f"Converted request: {converted}")
    return converted

推荐工具选择

参考资料

• LiteLLM 文档: https://docs.litellm.ai
• One API 文档: https://github.com/songquanpeng/one-api
• OpenRouter 文档: https://openrouter.ai/docs
• Portkey 文档: https://docs.portkey.ai