构建高效ChatGPT提词库：从技术选型到生产环境实践

1次阅读

没有评论

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

在使用 ChatGPT API 进行对话生成时，开发者常遇到以下问题：

上下文丢失 ：ChatGPT API 本身是无状态的，每次请求都需要携带完整的对话历史
重复生成 ：相同或相似的提示词可能被反复提交，导致 API 调用成本增加
响应延迟 ：频繁的 API 调用和大量上下文数据的传输会增加系统延迟
成本控制 ：按 token 计费的模式下，重复传输相同上下文会造成不必要的开销

针对提词库的存储需求，我们评估了三种主流方案：

纯内存存储
优点：零延迟，实现简单
缺点：数据易丢失，扩展性差
适用场景：开发测试环境
SQL 数据库
优点：数据持久化，支持复杂查询
缺点：性能瓶颈明显，维护成本高
适用场景：需要长期保存和分析的历史数据
Redis 缓存
优点：高性能，支持丰富的数据结构
缺点：需要额外维护缓存服务
适用场景：生产环境高频访问数据

综合评估后，Redis 在性能和功能上最为平衡，特别适合提词库场景。

以下是一个符合 PEP8 规范的实现示例：

import json
import logging
from typing import Optional, Dict
import redis
from pydantic import BaseModel

# 配置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PromptCache:
    """基于 Redis 的提词库实现"""

    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
        try:
            self.redis = redis.StrictRedis(
                host=host,
                port=port,
                db=db,
                decode_responses=True
            )
            self.redis.ping()  # 测试连接
        except redis.ConnectionError as e:
            logger.error(f"Redis 连接失败: {e}")
            raise

    def store_prompt(self, user_id: str, prompt: str, response: str, ttl: int = 3600) -> bool:
        """存储提示词和响应"""
        try:
            data = {'prompt': prompt, 'response': response}
            return self.redis.setex(name=f"prompt:{user_id}:{hash(prompt)}",
                time=ttl,
                value=json.dumps(data)
            )
        except Exception as e:
            logger.error(f"存储提示词失败: {e}")
            return False

    def get_response(self, user_id: str, prompt: str) -> Optional[str]:
        """获取缓存的响应"""
        try:
            data = self.redis.get(f"prompt:{user_id}:{hash(prompt)}")
            return json.loads(data).get('response') if data else None
        except Exception as e:
            logger.error(f"获取响应失败: {e}")
            return None

批量处理
使用 Redis 的 pipeline 减少网络往返

示例：

pipe = self.redis.pipeline()
for prompt in prompts:
    pipe.get(f"prompt:{user_id}:{hash(prompt)}")
responses = pipe.execute()

异步 IO
采用 aioredis 等异步客户端
配合 asyncio 实现非阻塞调用
缓存策略
设置合理的 TTL（建议 1 -24 小时）
实现 LRU 淘汰机制
添加本地内存作为二级缓存

并发竞争
使用 Redis 的 WATCH/MULTI 实现乐观锁

示例：

with self.redis.pipeline() as pipe:
    while True:
        try:
            pipe.watch(key)
            # 业务逻辑
            pipe.multi()
            pipe.execute()
            break
        except redis.WatchError:
            continue