共计 4190 个字符,预计需要花费 11 分钟才能阅读完成。
技术背景:两大模型的 API 设计哲学
-
ChatGPT 的请求响应模式:OpenAI 采用典型的无状态设计,每个 API 请求都是独立事件。开发者需要自行维护对话历史,在 messages 数组中传递完整上下文。优点是灵活性高,缺点是长对话时 Token 消耗增长明显。

-
Claude 的会话连续性 :Anthropic 的 API 内置会话状态管理,通过
conversation_id自动关联上下文。在连续对话场景下,开发者只需传递最新用户输入,显著降低 Token 开销。代价是会话过期需要重新建立连接(默认 30 分钟闲置断开)。 -
生成策略差异 :ChatGPT 的 temperature 参数范围(0-2) 比 Claude(0-1)更宽,实际测试显示:
- 当 temperature=0.7 时,ChatGPT 输出的创意性更强
- Claude 在 temperature=0.3 时仍能保持较好的逻辑连贯性
接入准备:快速获取 API 密钥
- OpenAI 平台:
- 登录OpenAI 账户中心
- 点击 ”Create new secret key” 生成密钥
-
注意免费试用配额可能随时调整(2023 年 9 月后注册用户默认无免费额度)
-
Anthropic 控制台:
- 访问 Anthropic 官网 申请早期访问权限
- 通过邮件收到邀请后,在 Settings 页面创建 API Key
-
新账号默认提供 $10 等效的免费额度
-
环境变量配置建议:
# 推荐使用.env 文件管理密钥 echo 'OPENAI_KEY=sk-your-key-here' >> .env echo 'ANTHROPIC_KEY=sk-ant-your-key-here' >> .env
Python 实战:从基础调用到高级封装
基础请求封装(带异常处理)
import os
import aiohttp
from dotenv import load_dotenv
load_dotenv()
async def chat_completion(
provider: str,
messages: list,
temperature: float = 0.7
):
"""
通用 AI 对话请求封装
:param provider: 服务商['openai','anthropic']
:param messages: 消息历史,格式参考对应 API 文档
:param temperature: 生成多样性控制(0-1)
:return: 响应 JSON 或抛出异常
"""headers = {"Content-Type":"application/json","Authorization": f"Bearer {os.getenv(f'{provider.upper()}_KEY')}"
}
payload = {
"model": "gpt-3.5-turbo" if provider == 'openai' else "claude-2",
"messages": messages,
"temperature": temperature
}
async with aiohttp.ClientSession() as session:
try:
url = "https://api.openai.com/v1/chat/completions" if provider == 'openai' \
else "https://api.anthropic.com/v1/complete"
async with session.post(url, json=payload, headers=headers) as resp:
if resp.status != 200:
error = await resp.json()
raise Exception(f"API error: {error.get('error', {}).get('message')}")
return await resp.json()
except Exception as e:
print(f"请求失败: {str(e)}")
raise
Claude 会话状态管理类
class ClaudeConversation:
def __init__(self):
self.conversation_id = None
self.memory = [] # 本地备份完整对话历史
async def send_message(self, user_input: str):
"""智能处理会话延续的 Claude 交互"""
payload = {"prompt": f"\n\nHuman: {user_input}\n\nAssistant:",
"model": "claude-2",
"max_tokens_to_sample": 1000
}
if self.conversation_id:
payload["conversation_id"] = self.conversation_id
response = await chat_completion("anthropic", payload)
self.conversation_id = response.get("conversation_id")
self.memory.append({"user": user_input, "ai": response["completion"]})
return response
流式响应处理示例
async def stream_response(provider: str, prompt: str):
"""处理大文本的流式响应,降低内存占用"""
async with aiohttp.ClientSession() as session:
url = "https://api.openai.com/v1/chat/completions" if provider == 'openai' \
else "https://api.anthropic.com/v1/complete"
payload = {
"model": "gpt-3.5-turbo" if provider == 'openai' else "claude-2",
"messages": [{"role": "user", "content": prompt}],
"stream": True
}
async with session.post(url, json=payload, headers=headers) as resp:
async for line in resp.content:
if line.strip():
print(line.decode('utf-8'), end='', flush=True)
生产环境关键策略
速率限制规避方案
from collections import deque
import time
class RateLimiter:
"""令牌桶算法实现"""
def __init__(self, rate: int, capacity: int):
self.rate = rate # 每秒补充令牌数
self.capacity = capacity # 桶容量
self.tokens = capacity
self.last_check = time.time()
self.queue = deque()
async def acquire(self):
now = time.time()
elapsed = now - self.last_check
self.last_check = now
# 补充令牌
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate
)
if self.tokens < 1:
delay = (1 - self.tokens) / self.rate
await asyncio.sleep(delay)
self.tokens = 0
else:
self.tokens -= 1
敏感内容过滤
import re
def content_filter(text: str) -> bool:
"""基础关键词 + 正则过滤"""
block_patterns = [r'(暴力 | 仇恨言论 | 自残)[\s\S]* 指南',
r'如何制作(炸弹 | 毒品)',
r'\b(儿童色情 | 未成年人)\b.*(图片 | 视频)'
]
for pattern in block_patterns:
if re.search(pattern, text, re.IGNORECASE):
return False
return True
成本监控公式
# OpenAI 计费示例(GPT-3.5-turbo)总费用 = (输入 Token 数 * $0.0015/1K) + (输出 Token 数 * $0.002/1K)
# Claude 计费示例
总费用 = (输入 + 输出字符数) / 1000 * $0.0205
验证与测试
Postman 测试集要点
- 环境变量配置:
- 设置
OPENAI_KEY和ANTHROPIC_KEY -
添加 Pre-request Script 自动添加 Authorization 头
-
测试用例设计:
- 正常流测试:验证 200 响应和完整返回结构
- 错误流测试:模拟无效 API Key、超长输入等场景
- 性能测试:记录响应时间百分位数(P90 < 2s)
单元测试示例
import pytest
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_claude_conversation():
"""验证会话连续性功能"""
mock_response = {
"completion": "测试回复",
"conversation_id": "test_conv_id"
}
# 模拟 API 返回
with patch('aiohttp.ClientSession.post',
new=AsyncMock(return_value=AsyncMock(json=AsyncMock(return_value=mock_response)))):
conv = ClaudeConversation()
await conv.send_message("你好")
assert conv.conversation_id == "test_conv_id"
assert len(conv.memory) == 1
开放式思考题
-
在多轮对话场景中,如何平衡上下文长度与 Token 消耗的关系?是否存在最优的上下文窗口滑动策略?
-
当同时接入多个 AI 提供方时,应该如何设计抽象层来统一不同 API 的响应格式和错误处理?
-
针对行业垂直领域(如医疗、法律),除了基础的内容过滤外,还需要哪些额外的安全防护措施?
正文完
发表至: 未分类
近两天内

