Bret与ChatGPT集成实战：解决大模型应用中的上下文管理难题

1次阅读

没有评论

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

在开发基于 ChatGPT 的复杂应用时，开发者普遍会遇到两个核心挑战：

多轮对话的上下文丢失：当用户会话因页面刷新或服务重启中断时，传统的无状态 HTTP 请求无法维持对话历史
Attention Window 限制：GPT 模型有固定的 token 处理上限（如 4096 tokens），超出部分会被强制截断

例如电商客服场景中，用户前五轮对话中提到的商品规格、优惠诉求等信息，可能在第六轮询问物流时突然丢失，导致体验断层。

优点：
开发简单，兼容现有 Web 技术栈
客户端存储减轻服务端压力
缺点：
存储容量有限（通常≤4KB）
敏感对话历史暴露在客户端不安全
无法处理跨设备会话同步

核心优势：
服务端持久化会话存储（支持 Redis/PostgreSQL 等后端）
自动的 token 计数与窗口滑动管理
内置的对话压缩和摘要生成能力

from bret import SessionManager
from redis import Redis

# 使用 Redis 作为持久化后端
session_store = Redis(host='redis-host', port=6379, db=0)
session_mgr = SessionManager(
    store_backend=session_store,
    default_ttl=3600  # 会话 1 小时无活动后过期
)

def get_chat_response(session_id: str, user_input: str):
    # 从存储恢复历史上下文
    history = session_mgr.get_session(session_id) or []

    # 构建符合 OpenAI 格式的消息列表
    messages = [{"role": "system", "content": "你是一个专业客服助手"},
        *history,
        {"role": "user", "content": user_input}
    ]

    # 自动处理 token 超限（滑动窗口算法）while count_tokens(messages) > 4000:  # 保留 96token 作为缓冲
        messages.pop(1)  # 移除最早的非系统消息

    # 调用 ChatGPT API
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages
    )

    # 持久化更新后的对话历史
    session_mgr.update_session(
        session_id,
        messages[1:] + [response.choices[0].message]
    )

    return response.choices[0].message.content

def count_tokens(messages: list) -> int:
    """基于消息结构估算总 token 数"""
    encoding = tiktoken.get_encoding("cl100k_base")
    return sum(len(encoding.encode(msg["content"])) 
        for msg in messages
    ) + len(messages) * 3  # 每个消息额外 3token 开销

# 中间件示例：处理过期会话
@app.middleware("http")
async def check_session(request: Request, call_next):
    session_id = request.cookies.get("session_id")
    if session_id and not session_mgr.session_exists(session_id):
        # 触发客户端重新创建会话
        response = JSONResponse({"error": "session_expired"})
        response.delete_cookie("session_id")
        return response
    return await call_next(request)