Python调用ChatGPT API实战指南：从认证到流式响应处理

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

背景痛点分析

调用大模型 API 时，开发者常遇到几个典型问题：

1. 认证配置错误：API 密钥过期或格式错误导致认证失败，缺乏自动刷新机制
2. 响应延迟高：同步阻塞式调用在长文本生成时造成 UI 卡顿
3. Token 消耗失控：未预估 prompt 的 token 消耗导致超额计费

技术方案对比

同步 vs 异步调用选择

• requests 同步方案 适合：
• 简单脚本任务

• 需要快速验证 API 功能的场景

• aiohttp 异步方案 适合：

• 高并发爬虫
• 需要实时流式输出的交互应用

核心代码实现

认证模块（带 JWT 刷新）

from datetime import datetime, timedelta
import jwt

class AuthManager:
    """处理 API 密钥刷新，自动生成 JWT 令牌"""
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._token_expiry = datetime.utcnow()

    @property
    def valid_token(self) -> str:
        """返回有效令牌，自动刷新过期令牌"""
        if datetime.utcnow() >= self._token_expiry:
            self._refresh_token()
        return self._current_token

    def _refresh_token(self):
        """使用 HS256 算法生成新令牌"""
        self._current_token = jwt.encode({"exp": datetime.utcnow() + timedelta(minutes=30)},
            self.api_key,
            algorithm="HS256"
        )
        self._token_expiry = datetime.utcnow() + timedelta(minutes=25)  # 提前 5 分钟刷新

流式响应处理

import json
from typing import Generator

def process_stream_response(response) -> Generator[str, None, None]:
    """
    处理 stream=True 的响应分块
    :param response: 原始响应对象
    :yield: 解析后的文本片段
    """buffer = b''
    for chunk in response.iter_content(chunk_size=1024):  # 平衡内存和网络效率
        if chunk:
            buffer += chunk
            while b'\n' in buffer:
                line, buffer = buffer.split(b'\n', 1)
                if line.startswith(b'data:'):
                    try:
                        data = json.loads(line[6:])
                        yield data['choices'][0]['delta'].get('content', '')
                    except json.JSONDecodeError:
                        continue

生产环境建议

监控指标设计

1. 关键指标：
2. API 调用平均耗时（p99/p95）
3. 每日 token 消耗量

4. 429 错误率

5. 实现示例：

   from prometheus_client import Counter, Histogram
   
   API_CALL_DURATION = Histogram('api_call_duration', 'API response time')
   TOKEN_USAGE = Counter('token_usage_total', 'Total tokens consumed')
   
   @API_CALL_DURATION.time()
   def make_api_call(prompt: str):
       # ... 调用逻辑...
       TOKEN_USAGE.inc(len(prompt.split()))  # 简化的 token 计数

延伸思考

1. 如何实现跨 API 调用的全链路追踪？
2. 当需要处理超长对话时，怎样优化 token 使用效率？
3. 在微服务架构中，如何设计 API 调用的熔断机制？

通过模块化设计、完善的错误处理和性能监控，可以构建稳定高效的 AI 服务调用层。建议在实际项目中根据具体需求选择同步 / 异步方案，并持续优化 token 使用策略。