1482 API Comparison Openai Anthropic B5351691

OpenAI 与 Anthropic API 差异对比调研

本文档详细对比 OpenAI 和 Anthropic (Claude) API 在消息格式、工具调用、多模态、响应格式和流式输出等方面的差异。

1. 消息格式 (Messages Format)

1.1 OpenAI Messages 结构

OpenAI 使用 messages 数组，每个消息包含 role 和 content 字段。

基本结构：

{
  "model": "gpt-4",
  "messages": [
    {
      "role": "system",
      "content": "You are a helpful assistant."
    },
    {
      "role": "user",
      "content": "Hello!"
    },
    {
      "role": "assistant",
      "content": "Hi there! How can I help you?"
    }
  ],
  "max_tokens": 1000
}

支持的角色：

• system: 系统提示词（可选，但建议放在第一条）
• user: 用户消息
• assistant: 助手回复
• tool: 工具调用结果（用于 function calling）

多模态内容格式：

{
  "role": "user",
  "content": [
    {
      "type": "text",
      "text": "What's in this image?"
    },
    {
      "type": "image_url",
      "image_url": {
        "url": "data:image/jpeg;base64,/9j/4AAQSkZJRg..."
      }
    }
  ]
}

1.2 Anthropic Messages 结构

Anthropic 使用 messages 数组，但结构略有不同。

基本结构：

{
  "model": "claude-3-5-sonnet-20241022",
  "max_tokens": 1024,
  "system": "You are a helpful assistant.",
  "messages": [
    {
      "role": "user",
      "content": "Hello!"
    },
    {
      "role": "assistant",
      "content": "Hi there! How can I help you?"
    }
  ]
}

关键差异：

• system 是独立参数，不在 messages 数组中
• messages 中只包含 user 和 assistant 角色
• 消息必须交替出现（user → assistant → user → assistant）

多模态内容格式：

{
  "role": "user",
  "content": [
    {
      "type": "text",
      "text": "What's in this image?"
    },
    {
      "type": "image",
      "source": {
        "type": "base64",
        "media_type": "image/jpeg",
        "data": "/9j/4AAQSkZJRg..."
      }
    }
  ]
}

1.3 System Prompt 处理差异

OpenAI 示例：

{
  "messages": [
    {
      "role": "system",
      "content": "You are a helpful assistant."
    },
    {
      "role": "user",
      "content": "Hello"
    }
  ]
}

Anthropic 示例：

{
  "system": "You are a helpful assistant.",
  "messages": [
    {
      "role": "user",
      "content": "Hello"
    }
  ]
}

1.4 格式转换示例

Anthropic → OpenAI：

def anthropic_to_openai(anthropic_request):
    """将 Anthropic 格式转换为 OpenAI 格式"""
    openai_messages = []
    
    # 将 system 参数转换为第一条 system 消息
    if "system" in anthropic_request:
        openai_messages.append({
            "role": "system",
            "content": anthropic_request["system"]
        })
    
    # 转换 messages（需要处理 content 数组）
    for msg in anthropic_request.get("messages", []):
        role = msg["role"]
        content = msg["content"]
        
        # 如果 content 是数组，需要转换格式
        if isinstance(content, list):
            converted_content = []
            for part in content:
                if part.get("type") == "text":
                    converted_content.append({
                        "type": "text",
                        "text": part.get("text", "")
                    })
                elif part.get("type") == "image":
                    # 转换图片格式
                    source = part.get("source", {})
                    if source.get("type") == "base64":
                        converted_content.append({
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:{source.get('media_type')};base64,{source.get('data')}"
                            }
                        })
            content = converted_content if converted_content else content[0].get("text", "") if content else ""
        
        openai_messages.append({
            "role": role,
            "content": content
        })
    
    return {
        "messages": openai_messages,
        "max_tokens": anthropic_request.get("max_tokens", 1024)
    }

OpenAI → Anthropic：

def openai_to_anthropic(openai_request):
    """将 OpenAI 格式转换为 Anthropic 格式"""
    anthropic_request = {
        "messages": [],
        "max_tokens": openai_request.get("max_tokens", 1024)
    }
    
    # 提取 system 消息
    messages = openai_request.get("messages", [])
    system_messages = [msg for msg in messages if msg.get("role") == "system"]
    if system_messages:
        anthropic_request["system"] = system_messages[0].get("content", "")
        messages = [msg for msg in messages if msg.get("role") != "system"]
    
    # 转换其他消息
    for msg in messages:
        role = msg.get("role")
        if role not in ("user", "assistant"):
            continue  # Anthropic 只支持 user 和 assistant
        
        content = msg.get("content")
        
        # 如果 content 是数组，需要转换格式
        if isinstance(content, list):
            converted_content = []
            for part in content:
                if part.get("type") == "text":
                    converted_content.append({
                        "type": "text",
                        "text": part.get("text", "")
                    })
                elif part.get("type") == "image_url":
                    # 转换图片格式
                    image_url = part.get("image_url", {}).get("url", "")
                    if image_url.startswith("data:"):
                        # 解析 data URL: data:image/jpeg;base64,xxx
                        parts = image_url.split(",", 1)
                        if len(parts) == 2:
                            header = parts[0]
                            data = parts[1]
                            media_type = header.split(";")[0].split(":")[1]
                            converted_content.append({
                                "type": "image",
                                "source": {
                                    "type": "base64",
                                    "media_type": media_type,
                                    "data": data
                                }
                            })
            content = converted_content if converted_content else content
        else:
            # 简单文本内容
            content = content if isinstance(content, list) else [{"type": "text", "text": str(content)}]
        
        anthropic_request["messages"].append({
            "role": role,
            "content": content
        })
    
    return anthropic_request

2. 工具调用 (Function Calling / Tool Use)

2.1 OpenAI Tools 定义格式

工具定义：

{
  "model": "gpt-4",
  "messages": [
    {
      "role": "user",
      "content": "What's the weather in San Francisco?"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "Get the current weather in a given location",
        "parameters": {
          "type": "object",
          "properties": {
            "location": {
              "type": "string",
              "description": "The city and state, e.g. San Francisco, CA"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"],
              "description": "The unit of temperature"
            }
          },
          "required": ["location"]
        }
      }
    }
  ],
  "tool_choice": "auto"
}

工具调用响应格式：

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1677652288,
  "model": "gpt-4",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_abc123",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"location\": \"San Francisco, CA\", \"unit\": \"fahrenheit\"}"
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ],
  "usage": {
    "prompt_tokens": 82,
    "completion_tokens": 18,
    "total_tokens": 100
  }
}

工具结果返回格式：

{
  "role": "tool",
  "tool_call_id": "call_abc123",
  "content": "72 degrees and sunny"
}

2.2 Anthropic Tools 定义格式

工具定义：

{
  "model": "claude-3-5-sonnet-20241022",
  "max_tokens": 1024,
  "messages": [
    {
      "role": "user",
      "content": "What's the weather in San Francisco?"
    }
  ],
  "tools": [
    {
      "name": "get_weather",
      "description": "Get the current weather in a given location",
      "input_schema": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "The city and state, e.g. San Francisco, CA"
          },
          "unit": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "description": "The unit of temperature"
          }
        },
        "required": ["location"]
      }
    }
  ]
}

工具调用响应格式（在 content blocks 中）：

{
  "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
  "type": "message",
  "role": "assistant",
  "content": [
    {
      "type": "tool_use",
      "id": "toolu_01A09q90qw90lq917835lq9",
      "name": "get_weather",
      "input": {
        "location": "San Francisco, CA",
        "unit": "fahrenheit"
      }
    }
  ],
  "model": "claude-3-5-sonnet-20241022",
  "stop_reason": "tool_use",
  "stop_sequence": null,
  "usage": {
    "input_tokens": 25,
    "output_tokens": 4
  }
}

工具结果返回格式：

{
  "role": "user",
  "content": [
    {
      "type": "tool_result",
      "tool_use_id": "toolu_01A09q90qw90lq917835lq9",
      "content": "72 degrees and sunny"
    }
  ]
}

2.3 关键差异对比

2.4 格式转换示例

OpenAI Tools → Anthropic Tools：

def openai_tools_to_anthropic(openai_tools):
    """将 OpenAI tools 格式转换为 Anthropic 格式"""
    anthropic_tools = []
    
    for tool in openai_tools:
        if tool.get("type") == "function":
            function = tool.get("function", {})
            anthropic_tools.append({
                "name": function.get("name", ""),
                "description": function.get("description", ""),
                "input_schema": function.get("parameters", {})
            })
    
    return anthropic_tools

Anthropic Tools → OpenAI Tools：

def anthropic_tools_to_openai(anthropic_tools):
    """将 Anthropic tools 格式转换为 OpenAI 格式"""
    openai_tools = []
    
    for tool in anthropic_tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool.get("name", ""),
                "description": tool.get("description", ""),
                "parameters": tool.get("input_schema", {})
            }
        })
    
    return openai_tools

工具调用响应转换：

OpenAI → Anthropic：

def openai_tool_calls_to_anthropic(openai_response):
    """将 OpenAI tool_calls 转换为 Anthropic tool_use blocks"""
    tool_use_blocks = []
    
    message = openai_response.choices[0].message
    if message.tool_calls:
        for tool_call in message.tool_calls:
            import json
            arguments = json.loads(tool_call.function.arguments) if isinstance(tool_call.function.arguments, str) else tool_call.function.arguments
            
            tool_use_blocks.append({
                "type": "tool_use",
                "id": tool_call.id,
                "name": tool_call.function.name,
                "input": arguments
            })
    
    return tool_use_blocks

Anthropic → OpenAI：

def anthropic_tool_use_to_openai(anthropic_response):
    """将 Anthropic tool_use blocks 转换为 OpenAI tool_calls"""
    tool_calls = []
    
    for block in anthropic_response.content:
        if block.type == "tool_use":
            import json
            tool_calls.append({
                "id": block.id,
                "type": "function",
                "function": {
                    "name": block.name,
                    "arguments": json.dumps(block.input)
                }
            })
    
    return tool_calls

工具结果转换：

OpenAI Tool Result → Anthropic Tool Result：

def openai_tool_result_to_anthropic(tool_message):
    """将 OpenAI tool 消息转换为 Anthropic tool_result"""
    return {
        "role": "user",
        "content": [
            {
                "type": "tool_result",
                "tool_use_id": tool_message.get("tool_call_id"),
                "content": tool_message.get("content", "")
            }
        ]
    }

Anthropic Tool Result → OpenAI Tool Result：

def anthropic_tool_result_to_openai(anthropic_message):
    """将 Anthropic tool_result 转换为 OpenAI tool 消息"""
    tool_results = []
    
    for block in anthropic_message.get("content", []):
        if block.get("type") == "tool_result":
            tool_results.append({
                "role": "tool",
                "tool_call_id": block.get("tool_use_id"),
                "content": block.get("content", "")
            })
    
    return tool_results

3. 多模态/图片 (Vision)

3.1 OpenAI 图片输入格式

Base64 编码：

{
  "role": "user",
  "content": [
    {
      "type": "text",
      "text": "What's in this image?"
    },
    {
      "type": "image_url",
      "image_url": {
        "url": "data:image/jpeg;base64,/9j/4AAQSkZJRg..."
      }
    }
  ]
}

URL 方式：

{
  "role": "user",
  "content": [
    {
      "type": "text",
      "text": "What's in this image?"
    },
    {
      "type": "image_url",
      "image_url": {
        "url": "https://example.com/image.jpg"
      }
    }
  ]
}

支持的格式：

• JPEG
• PNG
• GIF
• WebP

3.2 Anthropic 图片输入格式

Base64 编码：

{
  "role": "user",
  "content": [
    {
      "type": "text",
      "text": "What's in this image?"
    },
    {
      "type": "image",
      "source": {
        "type": "base64",
        "media_type": "image/jpeg",
        "data": "/9j/4AAQSkZJRg..."
      }
    }
  ]
}

支持的格式：

• JPEG
• PNG
• GIF
• WebP
• PDF（Anthropic 特有）

PDF 支持示例：

{
  "role": "user",
  "content": [
    {
      "type": "text",
      "text": "Summarize this PDF"
    },
    {
      "type": "image",
      "source": {
        "type": "base64",
        "media_type": "application/pdf",
        "data": "JVBERi0xLjQKJeLjz9MK..."
      }
    }
  ]
}

3.3 关键差异

3.4 格式转换示例

OpenAI Image → Anthropic Image：

def openai_image_to_anthropic(image_block):
    """将 OpenAI image_url 转换为 Anthropic image"""
    image_url = image_block.get("image_url", {}).get("url", "")
    
    if image_url.startswith("data:"):
        # 解析 data URL
        parts = image_url.split(",", 1)
        if len(parts) == 2:
            header = parts[0]  # data:image/jpeg;base64
            data = parts[1]
            
            # 提取 media_type
            media_type = header.split(";")[0].split(":")[1]  # image/jpeg
            
            return {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": media_type,
                    "data": data
                }
            }
    else:
        # URL 方式 - Anthropic 不支持，需要先下载并转换为 base64
        raise ValueError("Anthropic API does not support image URLs. Please convert to base64 first.")
    
    return None

Anthropic Image → OpenAI Image：

def anthropic_image_to_openai(image_block):
    """将 Anthropic image 转换为 OpenAI image_url"""
    source = image_block.get("source", {})
    
    if source.get("type") == "base64":
        media_type = source.get("media_type", "image/jpeg")
        data = source.get("data", "")
        
        return {
            "type": "image_url",
            "image_url": {
                "url": f"data:{media_type};base64,{data}"
            }
        }
    
    return None

4. 响应格式 (Response Format)

4.1 OpenAI 响应格式

基本响应：

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1677652288,
  "model": "gpt-4",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "Hello! How can I help you today?"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 10,
    "completion_tokens": 8,
    "total_tokens": 18
  }
}

带工具调用的响应：

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1677652288,
  "model": "gpt-4",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_abc123",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"location\": \"San Francisco\"}"
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ],
  "usage": {
    "prompt_tokens": 82,
    "completion_tokens": 18,
    "total_tokens": 100
  }
}

Finish Reasons：

• stop: 正常完成
• length: 达到 max_tokens 限制
• tool_calls: 需要调用工具
• content_filter: 内容被过滤
• function_call: 旧版 function calling（已废弃）

4.2 Anthropic 响应格式

基本响应：

{
  "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
  "type": "message",
  "role": "assistant",
  "content": [
    {
      "type": "text",
      "text": "Hello! How can I help you today?"
    }
  ],
  "model": "claude-3-5-sonnet-20241022",
  "stop_reason": "end_turn",
  "stop_sequence": null,
  "usage": {
    "input_tokens": 10,
    "output_tokens": 8
  }
}

带工具调用的响应：

{
  "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
  "type": "message",
  "role": "assistant",
  "content": [
    {
      "type": "text",
      "text": "I'll check the weather for you."
    },
    {
      "type": "tool_use",
      "id": "toolu_01A09q90qw90lq917835lq9",
      "name": "get_weather",
      "input": {
        "location": "San Francisco"
      }
    }
  ],
  "model": "claude-3-5-sonnet-20241022",
  "stop_reason": "tool_use",
  "stop_sequence": null,
  "usage": {
    "input_tokens": 25,
    "output_tokens": 4
  }
}

Stop Reasons：

• end_turn: 正常完成一轮对话
• max_tokens: 达到 max_tokens 限制
• tool_use: 需要调用工具
• stop_sequence: 遇到停止序列

4.3 关键差异对比

4.4 Stop Reason / Finish Reason 映射

OPENAI_TO_ANTHROPIC_FINISH_REASON = {
    "stop": "end_turn",
    "length": "max_tokens",
    "tool_calls": "tool_use",
    "content_filter": "end_turn",  # 近似映射
    "function_call": "tool_use"  # 旧版 function calling
}

ANTHROPIC_TO_OPENAI_STOP_REASON = {
    "end_turn": "stop",
    "max_tokens": "length",
    "tool_use": "tool_calls",
    "stop_sequence": "stop"  # 近似映射
}

def convert_finish_reason(openai_reason):
    """将 OpenAI finish_reason 转换为 Anthropic stop_reason"""
    return OPENAI_TO_ANTHROPIC_FINISH_REASON.get(openai_reason, "end_turn")

def convert_stop_reason(anthropic_reason):
    """将 Anthropic stop_reason 转换为 OpenAI finish_reason"""
    return ANTHROPIC_TO_OPENAI_STOP_REASON.get(anthropic_reason, "stop")

4.5 响应格式转换示例

OpenAI Response → Anthropic Response：

def openai_response_to_anthropic(openai_response):
    """将 OpenAI 响应转换为 Anthropic 格式"""
    choice = openai_response.choices[0]
    message = choice.message
    
    # 构建 content blocks
    content = []
    
    # 添加文本内容
    if message.content:
        content.append({
            "type": "text",
            "text": message.content
        })
    
    # 添加工具调用
    if message.tool_calls:
        import json
        for tool_call in message.tool_calls:
            arguments = json.loads(tool_call.function.arguments) if isinstance(tool_call.function.arguments, str) else tool_call.function.arguments
            content.append({
                "type": "tool_use",
                "id": tool_call.id,
                "name": tool_call.function.name,
                "input": arguments
            })
    
    # 转换 finish_reason
    stop_reason = convert_finish_reason(choice.finish_reason)
    
    return {
        "id": openai_response.id,
        "type": "message",
        "role": "assistant",
        "content": content,
        "model": openai_response.model,
        "stop_reason": stop_reason,
        "stop_sequence": None,
        "usage": {
            "input_tokens": openai_response.usage.prompt_tokens,
            "output_tokens": openai_response.usage.completion_tokens
        }
    }

Anthropic Response → OpenAI Response：

def anthropic_response_to_openai(anthropic_response):
    """将 Anthropic 响应转换为 OpenAI 格式"""
    # 提取文本内容
    text_content = ""
    tool_calls = []
    
    for block in anthropic_response.content:
        if block.type == "text":
            text_content += block.text
        elif block.type == "tool_use":
            import json
            tool_calls.append({
                "id": block.id,
                "type": "function",
                "function": {
                    "name": block.name,
                    "arguments": json.dumps(block.input)
                }
            })
    
    # 构建 message
    message = {
        "role": "assistant",
        "content": text_content if text_content else None
    }
    
    if tool_calls:
        message["tool_calls"] = tool_calls
    
    # 转换 stop_reason
    finish_reason = convert_stop_reason(anthropic_response.stop_reason)
    
    return {
        "id": anthropic_response.id,
        "object": "chat.completion",
        "created": int(time.time()),
        "model": anthropic_response.model,
        "choices": [
            {
                "index": 0,
                "message": message,
                "finish_reason": finish_reason
            }
        ],
        "usage": {
            "prompt_tokens": anthropic_response.usage.input_tokens,
            "completion_tokens": anthropic_response.usage.output_tokens,
            "total_tokens": anthropic_response.usage.input_tokens + anthropic_response.usage.output_tokens
        }
    }

5. 流式输出 (Streaming)

5.1 OpenAI 流式格式

请求参数：

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True  # 启用流式输出
)

流式响应格式（SSE）：

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1694268190,"model":"gpt-4","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}]}

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1694268190,"model":"gpt-4","choices":[{"index":0,"delta":{"content":" there"},"finish_reason":null}]}

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1694268190,"model":"gpt-4","choices":[{"index":0,"delta":{"content":"!"},"finish_reason":"stop"}]}

data: [DONE]

Python SDK 使用：

stream = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="")

工具调用的流式响应：

# 流式响应中，工具调用会分多个 chunk 发送
for chunk in stream:
    if chunk.choices[0].delta.tool_calls:
        for tool_call_delta in chunk.choices[0].delta.tool_calls:
            # tool_call_delta 包含部分工具调用信息
            # 需要累积多个 chunk 才能组成完整的 tool_call
            pass

5.2 Anthropic 流式格式

请求参数：

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="")

流式事件类型：

• message_start: 消息开始
• content_block_start: 内容块开始（text 或 tool_use）
• content_block_delta: 内容增量（文本片段）
• content_block_stop: 内容块结束
• message_delta: 消息增量（usage 信息）
• message_stop: 消息结束

流式响应示例（SSE）：

event: message_start
data: {"type":"message_start","message":{"id":"msg_01XFDUDYJgAACzvnptvVoYEL","type":"message","role":"assistant","content":[],"model":"claude-3-5-sonnet-20241022","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" there"}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"!"}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":3}}

event: message_stop
data: {"type":"message_stop"}

Python SDK 使用：

# 方式 1: 使用 text_stream（最简单）
with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="")

# 方式 2: 使用事件流（更细粒度控制）
with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="")

工具调用的流式响应：

with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                # 工具调用开始
                tool_id = event.content_block.id
                tool_name = event.content_block.name
        elif event.type == "content_block_delta":
            if event.delta.type == "input_json_delta":
                # 工具参数增量
                partial_input = event.delta.partial_json

5.3 关键差异对比

5.4 流式格式转换示例

OpenAI Stream → Anthropic Stream（模拟）：

def openai_stream_to_anthropic_events(openai_stream):
    """将 OpenAI 流式响应转换为 Anthropic 事件格式（模拟）"""
    message_id = None
    content_blocks = []
    current_block_index = 0
    
    for chunk in openai_stream:
        if not message_id:
            message_id = chunk.id
            # 发送 message_start
            yield {
                "type": "message_start",
                "message": {
                    "id": message_id,
                    "type": "message",
                    "role": "assistant",
                    "content": [],
                    "model": chunk.model,
                    "stop_reason": None,
                    "usage": {"input_tokens": 0, "output_tokens": 0}
                }
            }
        
        choice = chunk.choices[0]
        delta = choice.delta
        
        # 处理文本内容
        if delta.content:
            # 如果是新的内容块，发送 content_block_start
            if current_block_index >= len(content_blocks):
                content_blocks.append({"type": "text", "text": ""})
                yield {
                    "type": "content_block_start",
                    "index": current_block_index,
                    "content_block": {"type": "text", "text": ""}
                }
            
            # 发送文本增量
            yield {
                "type": "content_block_delta",
                "index": current_block_index,
                "delta": {
                    "type": "text_delta",
                    "text": delta.content
                }
            }
            content_blocks[current_block_index]["text"] += delta.content
        
        # 处理工具调用
        if delta.tool_calls:
            # OpenAI 的工具调用需要累积多个 chunk
            # 这里简化处理
            pass
        
        # 处理完成
        if choice.finish_reason:
            # 结束当前内容块
            if current_block_index < len(content_blocks):
                yield {
                    "type": "content_block_stop",
                    "index": current_block_index
                }
            
            # 发送 message_delta 和 message_stop
            yield {
                "type": "message_delta",
                "delta": {
                    "stop_reason": convert_finish_reason(choice.finish_reason),
                    "stop_sequence": None
                },
                "usage": {
                    "output_tokens": chunk.usage.completion_tokens if hasattr(chunk, 'usage') else 0
                }
            }
            
            yield {"type": "message_stop"}

Anthropic Stream → OpenAI Stream（模拟）：

def anthropic_stream_to_openai_chunks(anthropic_stream):
    """将 Anthropic 流式响应转换为 OpenAI chunk 格式（模拟）"""
    message_id = None
    model = None
    accumulated_content = ""
    tool_calls_accumulator = {}
    
    for event in anthropic_stream:
        if event.type == "message_start":
            message_id = event.message.id
            model = event.message.model
        
        elif event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                # 开始新的工具调用
                tool_id = event.content_block.id
                tool_calls_accumulator[tool_id] = {
                    "id": tool_id,
                    "type": "function",
                    "function": {
                        "name": event.content_block.name,
                        "arguments": ""
                    }
                }
        
        elif event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                # 文本增量
                accumulated_content += event.delta.text
                yield {
                    "id": message_id,
                    "object": "chat.completion.chunk",
                    "created": int(time.time()),
                    "model": model,
                    "choices": [{
                        "index": 0,
                        "delta": {"content": event.delta.text},
                        "finish_reason": None
                    }]
                }
            elif event.delta.type == "input_json_delta":
                # 工具参数增量
                # OpenAI 格式需要累积完整的 arguments
                pass
        
        elif event.type == "content_block_stop":
            # 内容块结束
            pass
        
        elif event.type == "message_delta":
            # 发送最终 chunk（如果有剩余内容）
            if accumulated_content:
                yield {
                    "id": message_id,
                    "object": "chat.completion.chunk",
                    "created": int(time.time()),
                    "model": model,
                    "choices": [{
                        "index": 0,
                        "delta": {},
                        "finish_reason": convert_stop_reason(event.delta.stop_reason)
                    }]
                }
        
        elif event.type == "message_stop":
            # 发送 [DONE]
            yield "[DONE]"

6. 完整转换工具示例

以下是一个完整的转换工具类，包含所有格式转换功能：

import json
import time
from typing import Any, Dict, List, Optional

class APIConverter:
    """OpenAI 和 Anthropic API 格式转换工具"""
    
    # Finish reason 映射
    OPENAI_TO_ANTHROPIC_FINISH_REASON = {
        "stop": "end_turn",
        "length": "max_tokens",
        "tool_calls": "tool_use",
        "content_filter": "end_turn",
        "function_call": "tool_use"
    }
    
    ANTHROPIC_TO_OPENAI_STOP_REASON = {
        "end_turn": "stop",
        "max_tokens": "length",
        "tool_use": "tool_calls",
        "stop_sequence": "stop"
    }
    
    @staticmethod
    def anthropic_to_openai_request(anthropic_request: Dict[str, Any]) -> Dict[str, Any]:
        """将 Anthropic 请求转换为 OpenAI 格式"""
        openai_messages = []
        
        # 转换 system
        if "system" in anthropic_request:
            openai_messages.append({
                "role": "system",
                "content": anthropic_request["system"]
            })
        
        # 转换 messages
        for msg in anthropic_request.get("messages", []):
            role = msg["role"]
            content = msg["content"]
            
            # 处理 content 数组
            if isinstance(content, list):
                converted_content = []
                for part in content:
                    if part.get("type") == "text":
                        converted_content.append({
                            "type": "text",
                            "text": part.get("text", "")
                        })
                    elif part.get("type") == "image":
                        source = part.get("source", {})
                        if source.get("type") == "base64":
                            converted_content.append({
                                "type": "image_url",
                                "image_url": {
                                    "url": f"data:{source.get('media_type')};base64,{source.get('data')}"
                                }
                            })
                content = converted_content if converted_content else content[0].get("text", "") if content else ""
            
            openai_messages.append({
                "role": role,
                "content": content
            })
        
        result = {
            "messages": openai_messages,
            "max_tokens": anthropic_request.get("max_tokens", 1024)
        }
        
        # 转换 tools
        if "tools" in anthropic_request:
            result["tools"] = APIConverter.anthropic_tools_to_openai(anthropic_request["tools"])
        
        return result
    
    @staticmethod
    def openai_to_anthropic_request(openai_request: Dict[str, Any]) -> Dict[str, Any]:
        """将 OpenAI 请求转换为 Anthropic 格式"""
        anthropic_request = {
            "messages": [],
            "max_tokens": openai_request.get("max_tokens", 1024)
        }
        
        # 提取 system 消息
        messages = openai_request.get("messages", [])
        system_messages = [msg for msg in messages if msg.get("role") == "system"]
        if system_messages:
            anthropic_request["system"] = system_messages[0].get("content", "")
            messages = [msg for msg in messages if msg.get("role") != "system"]
        
        # 转换其他消息
        for msg in messages:
            role = msg.get("role")
            if role not in ("user", "assistant"):
                continue
            
            content = msg.get("content")
            if isinstance(content, list):
                converted_content = []
                for part in content:
                    if part.get("type") == "text":
                        converted_content.append({
                            "type": "text",
                            "text": part.get("text", "")
                        })
                    elif part.get("type") == "image_url":
                        image_url = part.get("image_url", {}).get("url", "")
                        if image_url.startswith("data:"):
                            parts = image_url.split(",", 1)
                            if len(parts) == 2:
                                header = parts[0]
                                data = parts[1]
                                media_type = header.split(";")[0].split(":")[1]
                                converted_content.append({
                                    "type": "image",
                                    "source": {
                                        "type": "base64",
                                        "media_type": media_type,
                                        "data": data
                                    }
                                })
                content = converted_content if converted_content else content
            else:
                content = [{"type": "text", "text": str(content)}]
            
            anthropic_request["messages"].append({
                "role": role,
                "content": content
            })
        
        # 转换 tools
        if "tools" in openai_request:
            anthropic_request["tools"] = APIConverter.openai_tools_to_anthropic(openai_request["tools"])
        
        return anthropic_request
    
    @staticmethod
    def anthropic_tools_to_openai(anthropic_tools: List[Dict]) -> List[Dict]:
        """将 Anthropic tools 转换为 OpenAI 格式"""
        openai_tools = []
        for tool in anthropic_tools:
            openai_tools.append({
                "type": "function",
                "function": {
                    "name": tool.get("name", ""),
                    "description": tool.get("description", ""),
                    "parameters": tool.get("input_schema", {})
                }
            })
        return openai_tools
    
    @staticmethod
    def openai_tools_to_anthropic(openai_tools: List[Dict]) -> List[Dict]:
        """将 OpenAI tools 转换为 Anthropic 格式"""
        anthropic_tools = []
        for tool in openai_tools:
            if tool.get("type") == "function":
                function = tool.get("function", {})
                anthropic_tools.append({
                    "name": function.get("name", ""),
                    "description": function.get("description", ""),
                    "input_schema": function.get("parameters", {})
                })
        return anthropic_tools
    
    @staticmethod
    def convert_finish_reason(openai_reason: str) -> str:
        """转换 finish_reason"""
        return APIConverter.OPENAI_TO_ANTHROPIC_FINISH_REASON.get(openai_reason, "end_turn")
    
    @staticmethod
    def convert_stop_reason(anthropic_reason: str) -> str:
        """转换 stop_reason"""
        return APIConverter.ANTHROPIC_TO_OPENAI_STOP_REASON.get(anthropic_reason, "stop")

7. 总结

主要差异总结

1. 消息格式

   • OpenAI: system 在 messages 数组中
   • Anthropic: system 是独立参数

2. 工具调用

   • OpenAI: tool_calls 数组，参数是 JSON 字符串
   • Anthropic: tool_use blocks，参数是 JSON 对象

3. 多模态

   • OpenAI: 支持 URL 和 base64
   • Anthropic: 仅支持 base64，但支持 PDF

4. 响应格式

   • OpenAI: choices[0].message
   • Anthropic: content 数组（blocks）

5. 流式输出

   • OpenAI: 简单的 chunk 格式
   • Anthropic: 事件驱动的流式格式

使用建议

…(truncated)