ChatGPT API 集成实战：从零构建智能对话软件的技术解析

1次阅读

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

在客服系统中集成 ChatGPT 可以降低 70% 人工坐席成本，同时提升首次解决率；教育软件通过对话式交互实现个性化学习路径推荐，提高用户留存；企业级应用借助其多语言能力快速实现全球化部署。这些价值背后，都需要扎实的技术实现作为支撑。

对比维度	OpenAI 官方 SDK	直接 HTTP 调用
开发效率	✅ 封装完善，快速集成	❌ 需自行处理序列化 / 反序列化
灵活性	❌ 高级功能受限	✅ 可完全自定义请求逻辑
协议支持	❌ 仅 REST	✅ 支持 WebSocket/SSE
错误处理	✅ 内置重试机制	❌ 需自行实现
版本更新	❌ 依赖库更新	✅ 即时生效

对于需要精细控制的中大型项目，推荐使用 HTTP 直连方案。

import asyncio
from aiohttp import ClientSession

async def chat_completion_with_retry(messages, max_retries=3):
    base_delay = 1  # 初始延迟 1 秒
    async with ClientSession() as session:
        for attempt in range(max_retries):
            try:
                async with session.post(
                    'https://api.openai.com/v1/chat/completions',
                    headers={'Authorization': f'Bearer {API_KEY}'},
                    json={'model': 'gpt-4', 'messages': messages},
                    timeout=30
                ) as resp:
                    return await resp.json()
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(base_delay * (2 ** attempt))

import redis
from pickle import dumps, loads

r = redis.Redis(host='localhost', port=6379)

def save_conversation(user_id, messages, ttl=3600):
    r.setex(f'conv:{user_id}', ttl, dumps(messages))

def load_conversation(user_id):
    data = r.get(f'conv:{user_id}')
    return loads(data) if data else []

async def stream_response(prompt):
    async with ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            headers={'Authorization': f'Bearer {API_KEY}'},
            json={
                'model': 'gpt-4',
                'messages': [{'role': 'user', 'content': prompt}],
                'stream': True
            },
            timeout=30
        ) as resp:
            async for chunk in resp.content:
                if chunk:
                    yield chunk.decode()

TCP 连接复用要点：保持单个 ClientSession 实例复用，避免每次请求建立新连接。

调用方式	QPS	平均延迟	错误率
同步	12	850ms	0.3%
异步	48	210ms	1.2%

设置 max_tokens=512 硬限制
对历史消息使用 gpt-3.5-turbo 压缩
实现自动摘要生成对话快照

import re

SENSITIVE_PATTERNS = [r'\b\d{4}[-]?\d{4}[-]?\d{4}[-]?\d{4}\b',  # 信用卡号
    r'\b\d{3}-?\d{2}-?\d{4}\b'  # SSN
]

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

用户 ID 哈希化存储
对话内容加密后存入冷存储
实现 30 天自动清除机制

from pydantic import BaseModel
from typing import AsyncGenerator
import jwt

class Message(BaseModel):
    role: str
    content: str

async def generate_token() -> str:
    payload = {'iss': 'your_service_id', 'exp': datetime.utcnow() + timedelta(minutes=30)}
    return jwt.encode(payload, SECRET_KEY, algorithm='HS256')

async def chat_stream(messages: list[Message]) -> AsyncGenerator[str, None]:
    token = await generate_token()
    async with ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            headers={'Authorization': f'Bearer {token}',
                'Content-Type': 'application/json'
            },
            json={
                'model': 'gpt-4',
                'messages': [m.dict() for m in messages],
                'stream': True,
                'temperature': 0.7
            }
        ) as resp:
            async for line in resp.content:
                yield line.decode()