面向开发者的ChatGPT：从API集成到生产环境最佳实践

4次阅读

没有评论

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

在集成 ChatGPT API 时，开发者常遇到以下几个典型问题：

API 响应不稳定 ：尤其是在高峰期，响应时间可能从几百毫秒飙升到几秒，影响用户体验。
长对话上下文丢失 ：默认情况下，ChatGPT 不会记住之前的对话内容，导致在多轮对话中上下文丢失。
Token 成本不可控 ：尤其是使用 gpt- 4 时，长对话或高频请求可能导致成本迅速攀升。

OpenAI 官方 SDK 和 LangChain 等封装库各有优劣：

OpenAI 官方 SDK：
优点：直接对接 API，性能最佳，功能最全。
缺点：需要手动处理许多细节，如重试机制、上下文管理等。
LangChain：
优点：提供了更高层次的抽象，简化了上下文管理和链式调用。
缺点：性能开销稍大，灵活性较低。

以下是一个 Python 示例，演示如何异步处理流式响应，并实现 backoff 重试机制：

import openai
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def stream_chat_completion(messages):
    response = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True
    )
    async for chunk in response:
        yield chunk["choices"][0]["delta"].get("content", "")

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def save_conversation(user_id, messages, ttl=3600):
    redis_client.setex(f"conversation:{user_id}", ttl, json.dumps(messages))

def load_conversation(user_id):
    data = redis_client.get(f"conversation:{user_id}")
    return json.loads(data) if data else []

通过实测数据，gpt-3.5-turbo 和 gpt- 4 的每秒请求数（RPS）对比如下：

gpt-3.5-turbo：约 20 RPS
gpt-4：约 5 RPS

temperature 参数控制输出的随机性：

低值（如 0.2）：输出更加确定性和一致，适合需要精确答案的场景。
高值（如 0.8）：输出更加随机和创造性，适合需要多样性的场景。

OpenAI 提供了 content moderation API，可以在请求前后对内容进行检查：

moderation_response = openai.Moderation.create(input="用户输入内容")
if moderation_response["results"][0]["flagged"]:
    print("内容包含敏感信息")

使用令牌桶算法控制请求速率：

from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=60, period=60)
def make_api_request():
    # API 请求代码

graph TD
    A[开始] --> B{是否需要上下文管理?}
    B -->| 是 | C[使用 Redis 缓存对话状态]
    B -->| 否 | D[直接调用 API]
    C --> E{是否需要流式响应?}
    D --> E
    E -->| 是 | F[使用异步流式处理]
    E -->| 否 | G[同步请求]