从零开始掌握Claude与ChatGPT代码交互：开发者实战指南

1次阅读

共计 3397 个字符，预计需要花费 9 分钟才能阅读完成。

在同时集成 Claude 和 ChatGPT API 时，开发者常遇到三类核心问题：

认证机制差异
Claude 使用 x-api-key 头部认证
ChatGPT 需要Authorization: Bearer+API Key
混合环境需要维护两套凭据系统
响应结构不兼容
Claude 返回 completion 字段
ChatGPT 返回choices[0].message.content
业务逻辑需要处理不同数据路径
上下文管理碎片化
Claude 支持自动对话历史管理
ChatGPT 需手动维护 messages 数组
多轮对话时状态同步困难

特性	Claude API	ChatGPT API
平均响应延迟	1200-1800ms	800-1200ms
单次 token 限制	9000	4096
多轮对话支持	自动上下文	需手动维护 messages
错误码体系	自定义 4xx/5xx	HTTP 标准码
流式响应	支持	支持

from typing import Literal
import httpx

APIType = Literal['claude', 'chatgpt']

class AuthHandler:
    def __init__(self, claude_key: str, openai_key: str):
        self._keys = {
            'claude': claude_key,
            'chatgpt': openai_key
        }

    def get_headers(self, api_type: APIType) -> dict:
        if api_type == 'claude':
            return {'x-api-key': self._keys[api_type]}
        return {'Authorization': f'Bearer {self._keys[api_type]}',
            'Content-Type': 'application/json'
        }

from functools import lru_cache
import json
from datetime import timedelta

class AIAdapter:
    def __init__(self, auth: AuthHandler):
        self.auth = auth
        self.client = httpx.AsyncClient(timeout=30.0)

    @lru_cache(maxsize=100)
    async def _call_api(
        self,
        api_type: APIType,
        prompt: str,
        context: str = ""
    ) -> str:
        endpoint = {
            'claude': 'https://api.anthropic.com/v1/complete',
            'chatgpt': 'https://api.openai.com/v1/chat/completions'
        }[api_type]

        payload = self._build_payload(api_type, prompt, context)

        for attempt in range(3):  # 重试机制
            try:
                resp = await self.client.post(
                    endpoint,
                    headers=self.auth.get_headers(api_type),
                    json=payload
                )
                resp.raise_for_status()
                return self._parse_response(api_type, resp.json())
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429:  # Rate limit
                    await asyncio.sleep(2 ** attempt)  # 指数退避
                    continue
                raise

    def _build_payload(self, api_type: APIType, prompt: str, context: str) -> dict:
        if api_type == 'claude':
            return {"prompt": f"{context}\n\nHuman: {prompt}\n\nAssistant:",
                "max_tokens_to_sample": 1000
            }
        return {
            "model": "gpt-3.5-turbo",
            "messages": [{"role": "system", "content": context},
                {"role": "user", "content": prompt}
            ]
        }

    def _parse_response(self, api_type: APIType, data: dict) -> str:
        if api_type == 'claude':
            return data["completion"].strip()
        return data["choices"][0]["message"]["content"].strip()

Claude：每分钟 100 请求，建议实现令牌桶算法
ChatGPT：RPM（每分钟请求数）根据账户等级变化，需动态获取

from ratelimit import limits, sleep_and_retry

# 装饰器实现速率控制
@sleep_and_retry
@limits(calls=90, period=60)  # 保留 10% 余量
def call_with_retry():
    # API 调用代码
    pass

import re

SENSITIVE_PATTERNS = [r'(?i)password\s*[:=]\s*[\"\']?\w+',
    r'\b(?:api|secret)_?key\b\s*[:=]\s*[\"\']?\w+'
]

def sanitize_input(text: str) -> str:
    for pattern in SENSITIVE_PATTERNS:
        text = re.sub(pattern, '[REDACTED]', text)
    return text

from datetime import datetime

class ConversationSession:
    def __init__(self, ttl: int = 1800):  # 默认 30 分钟
        self._cache = {}
        self.ttl = ttl

    def add(self, session_id: str, context: str):
        self._cache[session_id] = {
            'context': context,
            'timestamp': datetime.now().timestamp()
        }

    def get(self, session_id: str) -> str | None:
        if session_id not in self._cache:
            return None

        entry = self._cache[session_id]
        if (datetime.now().timestamp() - entry['timestamp']) > self.ttl:
            del self._cache[session_id]
            return None

        return entry['context']

使用相同 prompt（约 200 tokens）测试：

指标	Claude	ChatGPT
平均响应时间	1580ms	920ms
消耗 tokens	240	210
超时发生率	1.2%	0.7%

任务：实现自动切换的 fallback 机制，当主 API 超时或返回错误时自动切换到备用 API。需考虑：

错误状态码的识别（503/429/500）
切换时的上下文格式转换
失败请求的自动重试队列
流量切换的熔断机制

提示代码结构：

class FallbackRouter:
    def __init__(self, primary: APIType, fallback: APIType):
        self.primary = primary
        self.fallback = fallback
        self._circuit_breaker = False

    async def send_request(self, prompt: str) -> str:
        # 实现你的解决方案
        pass