ChatGPT技术解析：从API调用到生产环境最佳实践

1次阅读

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

开发者在集成 ChatGPT 时经常面临以下典型问题：

Token 限制 ：单个请求的 token 上限为 4096（包括输入和输出），长对话场景需谨慎管理上下文
对话连贯性 ：普通 API 调用无法自动维护对话历史，需自行实现上下文存储与拼接
响应延迟 ：复杂请求或高峰时段可能出现响应缓慢，影响用户体验

Completion API
适合单轮文本补全任务
输入为纯文本字符串
不支持对话角色标记
Chat API
专为多轮对话设计
支持 system/user/assistant 角色定义
内置对话状态理解能力

优势：
实现打字机式逐词输出效果
降低用户感知延迟
服务端可提前开始处理
劣势：
客户端需处理更复杂的响应解析
网络中断时恢复成本较高
不适合需要完整响应的批处理场景

from typing import List, Dict

def build_conversation_context(history: List[Dict[str, str]], 
    new_query: str,
    system_prompt: str = "You are a helpful assistant."
) -> List[Dict[str, str]]:
    """
    构建符合 Chat API 要求的消息数组
    :param history: 历史消息记录，格式 [{"role":"user","content":"..."},...]
    :param new_query: 用户新输入
    :param system_prompt: 系统角色设定
    :return: 完整上下文数组
    """messages = [{"role":"system","content": system_prompt}]

    # 保留最近 3 轮对话以避免超限
    for msg in history[-6:]:  # 假设每轮包含 user+assistant 两条
        messages.append(msg)

    messages.append({"role": "user", "content": new_query})
    return messages

⚠️ 历史上下文长度需控制在 4096 tokens 内，建议实现 token 计数器或使用 tiktoken 库精确计算

import aiohttp
from typing import AsyncGenerator

async def stream_chat_response(messages: List[Dict[str, str]]) -> AsyncGenerator[str, None]:
    """
    异步流式获取 ChatGPT 响应
    :param messages: 完整上下文消息
    :yield: 实时返回的文本片段
    """api_url ="https://api.openai.com/v1/chat/completions"headers = {"Authorization": f"Bearer {API_KEY}","Content-Type":"application/json"
    }
    payload = {
        "model": "gpt-3.5-turbo",
        "messages": messages,
        "stream": True,
        "temperature": 0.7
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(api_url, json=payload, headers=headers) as resp:
            if resp.status != 200:
                raise Exception(f"API 请求失败: {resp.status}")

            async for line in resp.content:
                if line.startswith(b"data: [DONE]"):
                    break
                if line.startswith(b"data: {"):
                    json_data = json.loads(line[6:])
                    if "choices" in json_data:
                        yield json_data["choices"][0]["delta"].get("content", "")