ChatGPT网站源码解析：构建高效对话系统的关键技术与避坑指南

1次阅读

没有评论

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

在真实流量下，对话系统往往会面临几个关键的性能挑战：

长连接资源占用：每个活跃用户会话都需要保持长时间的连接，这会迅速消耗服务器资源
上下文 token 爆炸：随着对话轮次增加，上下文 token 数量呈指数级增长，导致 API 调用成本飙升
高并发响应延迟：当大量请求同时到达时，传统的同步处理模式会导致响应时间显著增加
状态管理复杂性：在多实例部署环境下，如何保持对话上下文的一致性成为难题

在 ChatGPT 类应用中，实时性要求催生了不同的通信方案：

传统轮询方案
实现简单，兼容性好
频繁的无效请求造成带宽浪费
平均延迟高（至少半个轮询周期）
Server-Sent Events (SSE)
单向服务器推送
自动重连机制
适合文本为主的流式响应
WebSocket
全双工通信
更低的协议开销
适合需要双向交互的复杂场景

在 ChatGPT 的实际应用中，SSE 因其简单性和对 HTTP 协议的兼容性成为主流选择。

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def generate_chunks(prompt: str):
    # 模拟 AI 生成过程
    for word in ["思考", "中", "...", "最终", "回答"]:
        yield f"data: {word}\n\n"
        await asyncio.sleep(0.5)

@app.post("/chat")
async def chat_endpoint(request: Request):
    data = await request.json()
    return StreamingResponse(generate_chunks(data["prompt"]),
        media_type="text/event-stream"
    )

关键点说明：

使用 StreamingResponse 实现分块传输
generate_chunks是异步生成器函数
每个数据块遵循 SSE 格式规范

实现简单，零延迟
无法应对服务重启
不适合分布式部署

示例数据结构：

sessions = {
    "session_id": {"history": ["msg1", "msg2"],
        "token_count": 42
    }
}

支持跨实例共享状态
数据持久化保障
需要处理序列化 / 反序列化

示例结构：

import redis
import json

r = redis.Redis()

def save_session(session_id, data):
    r.set(f"chat:{session_id}", json.dumps(data), ex=86400)

def load_session(session_id):
    data = r.get(f"chat:{session_id}")
    return json.loads(data) if data else None

from transformers import GPT2Tokenizer

tokenizer = GPT2Tokenizer.from_pretrained("gpt2")

def count_tokens(text: str) -> int:
    return len(tokenizer.encode(text))

class TokenLimiter:
    def __init__(self, max_tokens=4096):
        self.max_tokens = max_tokens

    def check(self, messages: list[str]) -> bool:
        total = sum(count_tokens(msg) for msg in messages)
        return total <= self.max_tokens

时间复杂度分析：
– count_tokens: O(n) 其中 n 是文本长度
– check: O(mn) m 是消息数量

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int = 100):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key: str) -> dict|None:
        if key not in self.cache:
            return None
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key: str, value: dict) -> None:
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

时间复杂度：
– 所有操作均为 O(1)

import re

banned_patterns = [
    r"信用卡号",
    r"\d{3}-\d{2}-\d{4}"  # SSN 格式
]

def sanitize_input(text: str) -> str:
    for pattern in banned_patterns:
        if re.search(pattern, text):
            raise ValueError("包含敏感信息")
    return text

import httpx
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 call_openai(prompt: str):
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            "https://api.openai.com/v1/chat/completions",
            json={"messages": [{"role": "user", "content": prompt}]}
        )
        response.raise_for_status()
        return response.json()

在构建生产级对话系统时，开发者需要在实时性、资源消耗和用户体验之间找到平衡点。当对话 session 持续 24 小时后，我们面临一个有趣的工程挑战：如何在不显著增加内存消耗的前提下，保持对话上下文的连贯性？

可能的解决方向包括：