ChatGPT极客指南：如何构建高可用AI对话系统的工程实践

1次阅读

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

在将 ChatGPT API 投入生产环境时，开发者通常会面临三个主要挑战：

API 调用成本高：按 Token 计费的模式使得长对话场景成本快速攀升
上下文管理困难：随着对话轮次增加，如何有效维护对话历史成为技术难点
流式响应延迟：直接调用 API 可能导致终端用户等待时间过长

通过将多个用户请求打包成单个 API 调用，可以显著降低网络开销和 Token 消耗。我们使用 aiohttp 实现异步批处理：

import aiohttp
from collections import deque

class BatchProcessor:
    def __init__(self, max_batch_size=10, max_wait_time=0.1):
        self.batch_queue = deque()
        self.max_batch_size = max_batch_size
        self.max_wait_time = max_wait_time  # 秒

    async def process_request(self, prompt):
        """将单个请求加入批处理队列"""
        self.batch_queue.append(prompt)

        if len(self.batch_queue) >= self.max_batch_size:
            return await self._flush_batch()

        # 设置超时等待更多请求
        await asyncio.sleep(self.max_wait_time)
        return await self._flush_batch()

    async def _flush_batch(self):
        """执行批量请求"""
        if not self.batch_queue:
            return []

        async with aiohttp.ClientSession() as session:
            payload = {
                "messages": [{"role": "user", "content": prompt} 
                    for prompt in self.batch_queue
                ],
                "max_tokens": 150
            }

            async with session.post(
                "https://api.openai.com/v1/chat/completions",
                json=payload,
                headers={"Authorization": f"Bearer {API_KEY}"}
            ) as resp:
                results = await resp.json()

        self.batch_queue.clear()
        return [choice['message']['content'] for choice in results['choices']]

我们实现基于 Token 计数的上下文窗口优化策略：

维护对话历史的 Token 计数
当接近模型上限 (如 4096 Tokens) 时触发压缩
使用摘要技术保留关键信息

from transformers import GPT2Tokenizer
tokenizer = GPT2Tokenizer.from_pretrained("gpt2")

class ContextManager:
    MAX_TOKENS = 3500  # 保留安全边界

    def __init__(self):
        self.history = []
        self.token_count = 0

    def add_message(self, role, content):
        """添加新消息并更新 Token 计数"""
        new_tokens = len(tokenizer.encode(content))

        # 触发上下文压缩
        while self.token_count + new_tokens > self.MAX_TOKENS:
            self._compress_context()

        self.history.append({"role": role, "content": content})
        self.token_count += new_tokens

    def _compress_context(self):
        """执行上下文压缩"""
        if len(self.history) < 2:
            return

        # 找到最不重要的消息（简单的启发式算法）min_importance = float('inf')
        to_remove = 0

        for i, msg in enumerate(self.history[1:-1]):
            importance = self._calculate_importance(msg)
            if importance < min_importance:
                min_importance = importance
                to_remove = i + 1

        # 移除并更新 Token 计数
        removed = self.history.pop(to_remove)
        self.token_count -= len(tokenizer.encode(removed["content"]))

    def _calculate_importance(self, message):
        """简单的重要性评估（可替换为更复杂的算法）"""
        content = message["content"]
        return len(content.split())  # 用词数作为代理指标

对话缓存需要解决两个核心问题：

如何识别语义相似的查询
如何处理对话上下文相关性

我们采用以下策略：

import hashlib
from typing import Dict, List

class DialogueCache:
    def __init__(self, max_size=1000):
        self.cache: Dict[str, str] = {}
        self.max_size = max_size

    def get_cache_key(self, messages: List[dict]) -> str:
        """生成基于对话上下文的缓存键"""
        # 对最近的 3 轮对话进行哈希
        recent = messages[-3:] if len(messages) > 3 else messages
        key_str = "|".join(f"{msg['role']}:{msg['content']}" for msg in recent
        )
        return hashlib.sha256(key_str.encode()).hexdigest()

    def get(self, messages: List[dict]) -> str:
        """获取缓存响应"""
        key = self.get_cache_key(messages)
        return self.cache.get(key)

    def set(self, messages: List[dict], response: str):
        """设置缓存项"""
        if len(self.cache) >= self.max_size:
            self.cache.popitem()  # 简单 LRU 策略

        key = self.get_cache_key(messages)
        self.cache[key] = response

我们对关键优化策略进行了基准测试：