如何使用ChatGPT提升开发效率：从API集成到生产环境最佳实践

2次阅读

没有评论

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

最近在项目中接入了 ChatGPT API 后，我发现直接裸调接口会遇到几个典型问题：

流式响应处理困难：当需要处理长文本生成时，官方 API 的流式响应（stream=True）需要开发者自己维护分块接收逻辑，稍不注意就会导致数据不完整或连接超时
token 成本不可控：特别是处理用户生成内容（UGC）时，prompt 长度不可预测，容易因 max_tokens 设置不当产生意外费用（比如用户提交了 5000 字的文档）
输出结果随机性：temperature 参数对结果影响巨大，但开发文档中的说明比较抽象，需要反复实验才能找到适合业务场景的值

官方 Python SDK 确实封装了基础功能，但在生产环境中我们发现两个局限：

缺乏灵活的异步支持（同步调用在高并发场景会成为瓶颈）
错误处理机制较简单（比如遇到 429 状态码时不会自动重试）

因此我们选择基于 aiohttp 自建异步客户端，核心优势在于：

可定制化重试策略（特别是对 RateLimit 的智能处理）
方便集成到现有异步框架（如 FastAPI/Django Channels）
更精细的性能监控（每个请求的延迟统计）

以下是经过生产验证的 Python 实现（Python 3.10+）：

import os
import asyncio
from typing import Optional, Dict, Any
from aiohttp import ClientSession, ClientError
from pydantic import BaseModel, Field

class ChatCompletionRequest(BaseModel):
    model: str = "gpt-3.5-turbo"
    messages: list[Dict[str, str]]
    temperature: float = 0.7
    max_tokens: Optional[int] = None

class ChatGPTClient:
    def __init__(self):
        self.api_key = os.getenv("OPENAI_API_KEY")
        self.base_url = "https://api.openai.com/v1"
        self.max_retries = 3

    async def _request_with_retry(
        self, 
        session: ClientSession,
        payload: dict,
        retry_count: int = 0
    ) -> dict:
        """
        指数退避重试机制
        Exponential backoff retry mechanism
        """
        try:
            async with session.post(f"{self.base_url}/chat/completions",
                json=payload,
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=30
            ) as resp:
                if resp.status == 429:  # Rate limited
                    wait_time = min(2 ** retry_count, 60)  # Cap at 60s
                    await asyncio.sleep(wait_time)
                    return await self._request_with_retry(session, payload, retry_count + 1)

                resp.raise_for_status()
                return await resp.json()

        except (ClientError, asyncio.TimeoutError) as e:
            if retry_count < self.max_retries:
                return await self._request_with_retry(session, payload, retry_count + 1)
            raise

    async def create_chat_completion(self, request: ChatCompletionRequest) -> dict:
        async with ClientSession() as session:
            return await self._request_with_retry(
                session,
                request.model_dump(exclude_none=True)
            )

为了保证输出结构化，我们使用 Jinja2 模板：

from jinja2 import Environment, StrictUndefined

env = Environment(undefined=StrictUndefined)  # 强制变量声明
def render_prompt(template: str, **kwargs) -> list[dict[str, str]]:
    """
    示例模板:
    {% for item in items %}
    - 用户输入: {{item.user_input}}
    - 上下文: {{item.context}}
    {% endfor %}
    请按 JSON 格式回复
    """
    rendered = env.from_string(template).render(**kwargs)
    return [{"role": "user", "content": rendered}]

根据我们的经验，建议：