ChatGPT首页开发实战：从零搭建高可用对话界面的技术解析

1次阅读

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

开发 ChatGPT 风格的对话界面看似简单，但实际落地时会遇到几个特有挑战：

消息排序问题 ：用户快速连续发送消息时，如何保证服务端返回顺序与前端渲染顺序一致。常见错误是直接依赖客户端时间戳排序，这在分布式系统中极易出现乱序。
加载状态管理 ：当 OpenAI API 响应较慢时，需要同时处理：
当前消息的加载动画
历史消息的可交互状态
新消息输入框的禁用逻辑
历史对话持久化 ：超过 GPT-3.5 的 4k token 限制时，传统方案直接截断会导致上下文丢失。需要智能压缩历史对话而不破坏语义连贯性。

方案	延迟	复杂度	断线恢复	适用场景
SSE	中	低	部分支持	服务端单向推送
WebSocket	低	高	支持	双向实时通信
Polling	高	低	支持	简单场景 / 兼容性要求高

决策树 ：
1. 是否需要双向通信？→ 是 → WebSocket
2. 是否需要支持 IE11？→ 是 → Polling
3. 其他情况 → SSE

interface Message {
  id: string; // UUIDv7 避免冲突
  content: string;
  role: 'user' | 'assistant';
  status: 'pending' | 'streaming' | 'completed';
}

const ChatList = () => {const [messages, setMessages] = useRecoilState(messageState);

  // 虚拟滚动优化
  return (<List height={600} itemSize={80} itemCount={messages.length}>
      {({index, style}) => (
        <MessageItem 
          style={style}
          message={messages[index]} 
        />
      )}
    </List>
  );
};

@app.websocket("/chat")
async def chat_stream(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            user_input = await websocket.receive_text()

            # 记录开始时间用于超时控制
            start_time = time.time()

            async for chunk in openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": user_input}],
                stream=True,
                timeout=10  # 设置超时
            ):
                if time.time() - start_time > 15:  # 双重超时检查
                    raise WebSocketDisconnect()

                await websocket.send_text(chunk.choices[0].delta.get("content", "")
                )
    except asyncio.TimeoutError:
        logging.error("API 响应超时")
    except WebSocketDisconnect:
        logging.info("客户端主动断开")

class TokenBucket:
    def __init__(self, capacity: int, fill_rate: float):
        self.capacity = capacity  # 桶容量
        self.tokens = capacity    # 当前令牌数
        self.last_fill = time.time()
        self.fill_rate = fill_rate  # 令牌 / 秒

    def consume(self, tokens=1) -> bool:
        now = time.time()
        elapsed = now - self.last_fill

        # 计算新增令牌
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.fill_rate
        )
        self.last_fill = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False