从零搭建ChatGPT应用：技术选型与核心实现详解

3次阅读

没有评论

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

当前开发者接入 ChatGPT API 时普遍面临三大痛点：

上下文管理复杂 ：多轮对话需要维护历史消息，而 GPT 模型本身是无状态的
流式响应处理困难 ：直接返回完整结果会导致用户等待时间过长
生产环境稳定性 ：API 调用存在速率限制和错误处理等问题

典型应用场景包括：

智能客服系统
内容创作助手
编程辅助工具
教育领域的问答应用

优势：

官方 SDK 支持完善
适合数据处理密集型任务
生态中有成熟的异步框架 (如 FastAPI)

劣势：

高并发场景需要额外优化
类型系统不如 TS 严格

优势：

天然非阻塞 IO 适合聊天场景
与前端技术栈统一
社区有成熟的 WebSocket 支持

劣势：

处理复杂业务逻辑时代码易混乱
类型系统依赖 TS 补充

Python 版本：

import openai

openai.api_key = 'your-api-key'

response = openai.ChatCompletion.create(
  model="gpt-3.5-turbo",
  messages=[{"role": "system", "content": "你是一个有帮助的助手"},
    {"role": "user", "content": "你好！"}
  ],
  temperature=0.7  # 控制回复随机性
)

Node.js 版本：

const {Configuration, OpenAIApi} = require('openai');

const configuration = new Configuration({apiKey: process.env.OPENAI_API_KEY,});

const openai = new OpenAIApi(configuration);

const response = await openai.createChatCompletion({
  model: "gpt-3.5-turbo",
  messages: [{ role: "system", content: "你是一个有帮助的助手"},
    {role: "user", content: "你好！"}
  ]
});

推荐采用环形缓冲区实现：

graph LR
    A[新用户消息] --> B{是否超过 token 限制?}
    B -- 是 --> C[移除最早的历史消息]
    B -- 否 --> D[保留完整历史]
    C --> E[将消息加入对话上下文]
    D --> E

Python 实现示例：

from collections import deque

class Conversation:
    def __init__(self, max_tokens=4096):
        self.history = deque()
        self.max_tokens = max_tokens
        self.current_tokens = 0

    def add_message(self, role, content):
        msg = {"role": role, "content": content}
        token_count = len(content) // 4  # 粗略估算

        while self.current_tokens + token_count > self.max_tokens and len(self.history) > 0:
            removed = self.history.popleft()
            self.current_tokens -= len(removed["content"]) // 4

        self.history.append(msg)
        self.current_tokens += token_count

Python 使用 SSE(Server-Sent Events) 示例：

from flask import Response

@app.route('/stream-chat')
def stream_chat():
    def generate():
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=conversation.history,
            stream=True
        )

        for chunk in response:
            if content := chunk.choices[0].delta.get('content'):
                yield f"data: {content}\n\n"

    return Response(generate(), mimetype='text/event-stream')

推荐方案：