FastAPI + Redis 实现大模型多轮对话（会话管理 + 上下文压缩）

关键词：FastAPI、Redis、通义千问、多轮对话、上下文压缩、会话管理

这是大模型接入实战的 Day2。Day1 跑通了单轮的非流式 / SSE 流式调用（见上一篇），Day2 的目标是把「多轮对话」做扎实：每个用户能开多个会话、会话有历史消息、发给模型时还要做上下文长度管理（滑动窗口 + 历史压缩）。

---

一、背景与整体设计

单轮调用把整段 messages 塞给模型即可。但真实聊天产品需要：

1. 多会话：一个用户能开多个对话，互不干扰
2. 消息持久化：刷新页面、重启服务，历史还在
3. 上下文管理：模型有 token 上限，聊久了必须裁剪历史

选型：

• 存储：Redis（项目本就有 Redis，list 结构天然适合「追加消息」）
• 模型：通义千问 qwen-plus，走百炼兼容 OpenAI 协议
• 上下文策略：保留最近 N 条 + 把更早的历史用模型「语义压缩」成一条

Key 设计

会话列表：  boss:llm:session:{user_id}           → list，每个元素 {"session_id","title","create_time"}
消息列表：  boss:llm:case2:messages:{user_id}:{session_id} → list，每个元素 {"role","content","create_time"}

把「会话维度」和「消息维度」拆成两个 key，互不干扰。

---

二、校验模型 app/schemas/llm_case1.py

今天在原来的 LLMCase1 基础上加了 LLMCase2：

from pydantic import BaseModel, Field


class LLMCase1(BaseModel):
    question: str = Field(..., title="问题", description="问题")


class LLMCase2(BaseModel):
    user_id: int = Field(..., title="用户ID", description="用户ID")
    session_id: str = Field(..., title="会话ID", description="会话ID")
    message: str = Field(..., title="用户消息", description="用户消息")

发送消息接口用 LLMCase2：靠 user_id + session_id 定位到具体会话的消息列表。

---

三、核心接口 app/apis/llm/case2_api.py

3.1 初始化 Redis 与客户端

import json
import uuid
from datetime import datetime

import redis
from openai import OpenAI

redis_client = redis.Redis(
    host='127.0.0.1', port=6379, db=1,
    decode_responses=True, protocol=2
)

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

llm_day02_router = APIRouter(prefix="/llm-day02", tags=["LLM-DAY02"])

decode_responses=True 让 lrange 直接返回字符串（否则是 bytes，还得手动解码）。

3.2 创建会话

@llm_day02_router.post("/create_session", summary="创建会话")
async def create_session(user_id: str):
    key = f"boss:llm:session:{user_id}"
    current_session = {
        "session_id": str(uuid.uuid4()),
        "title": "新会话",
        "create_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }
    redis_client.lpush(key, json.dumps(current_session, ensure_ascii=False))
    return {
        "code": 1, "message": "会话创建成功",
        "data": {"current_session": current_session},
    }

• 用 uuid4() 生成会话 ID
• lpush 把新会话压进该用户的会话列表
• ensure_ascii=False 很关键：否则中文标题会被转成 \uXXXX，前端展示乱码

3.3 查询会话列表

@llm_day02_router.get("/get_session_list", summary="查询会话列表")
async def get_session_list(user_id: int):
    key = f"boss:llm:session:{user_id}"
    user_sessions = [json.loads(s) for s in redis_client.lrange(key, 0, -1)]
    return {
        "code": 1, "message": "会话列表查询成功",
        "data": {"user_sessions": user_sessions},
    }

lrange(key, 0, -1) 取全部，json.loads 还原成 dict 列表。

3.4 发送消息（核心）

@llm_day02_router.post("/send_message", summary="发送消息")
async def send_message(case_request: LLMCase2):
    key = f"boss:llm:case2:messages:{case_request.user_id}:{case_request.session_id}"

    # 1. 取历史消息
    user_session_messages = redis_client.lrange(key, 0, -1)

    # 2. 第一轮对话：注入 system 人设
    if len(user_session_messages) == 0:
        system_message = {"role": "system", "content": "你是一个智能助手"}
        redis_client.rpush(key, json.dumps(system_message, ensure_ascii=False))

        # 用首条用户消息前 10 字更新会话标题
        session_key = f"boss:llm:session:{case_request.user_id}"
        user_sessions = redis_client.lrange(session_key, 0, -1)
        for index, session in enumerate(user_sessions):
            session_dict = json.loads(session)
            if session_dict["session_id"] == case_request.session_id:
                session_dict["title"] = case_request.message[:10] + "..."
                redis_client.lset(session_key, index, json.dumps(session_dict, ensure_ascii=False))

    # 3. 追加当前用户消息
    current_user_message = {
        "role": "user", "content": case_request.message,
        "create_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }
    redis_client.rpush(key, json.dumps(current_user_message, ensure_ascii=False))

    # 4. 取完整消息 + 上下文压缩
    messages = [json.loads(m) for m in redis_client.lrange(key, 0, -1)]
    messages = compression_messages(messages)

    # 5. 调模型、追加回复
    ai_reply = get_ai_response(messages)
    ai_reply_message = {
        "role": "assistant", "content": ai_reply,
        "create_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }
    redis_client.rpush(key, json.dumps(ai_reply_message, ensure_ascii=False))
    return {"code": 1, "message": "发送消息成功"}

get_ai_response 就是一层薄封装：

def get_ai_response(messages: list):
    completions = client.chat.completions.create(model="qwen-plus", messages=messages)
    return completions.choices[0].message.content

3.5 查询会话详情

@llm_day02_router.get("/get_user_session_messages", summary="查询会话详情")
async def get_user_session_messages(user_id: int, session_id: str):
    key = f"boss:llm:case2:messages:{user_id}:{session_id}"
    # 跳过第 0 条的 system 人设，只回业务消息
    user_session_messages = redis_client.lrange(key, 1, -1)
    user_session_messages = [json.loads(m) for m in user_session_messages]
    return {
        "code": 1, "message": "查询会话详情成功",
        "data": {"user_session_messages": user_session_messages},
    }

注意 lrange(key, 1, -1) 从索引 1 开始，跳过 system 人设，只把 user/assistant 的消息返给前端。

---

四、上下文压缩 compression_messages

聊久了 messages 会超 token 上限，这里用「滑动窗口 + 历史压缩」：

from app.core.logging import logger

def compression_messages(messages: list):
    logger.info(f"当前消息的长度:{len(messages)}")
    if len(messages) <= 5:
        logger.info("不需要压缩")
        return messages
    logger.info("需要压缩")

    # 保留最近 5 条
    recently_messages = messages[-5:]

    # 5 条之前的做语义压缩
    history_messages = messages[:-5]
    c_messages = [{
        "role": "user",
        "content": f"请将以下用户与大模型的上下文进行语义压缩,"
                   f"要保留核心的关键信息,上下文:{history_messages}"
    }]
    res = get_ai_response(c_messages)
    return [{"role": "user", "content": res}] + recently_messages

思路：

• 消息 ≤ 5 条：原样返回
• 消息 > 5 条：最近 5 条原样保留；更早的历史拼成一条压缩 prompt 让模型「总结成一段」，结果作为一条 user 消息顶在前面
• 效果：模型既能看到「压缩后的早期背景」，又不丢「最近几轮的精确对话」

---

五、路由注册 main.py

from app.apis.llm.case2_api import llm_day02_router

app.include_router(llm_day02_router)

接口汇总：

方法	路径	说明
POST	/llm-day02/create_session	创建会话
GET	/llm-day02/get_session_list	会话列表
POST	/llm-day02/send_message	发送消息（含压缩）
GET	/llm-day02/get_user_session_messages	会话详情

练手草稿：llm/case3.py（多轮 messages 演练）、llm/case4.py（Redis 存对话演练）目前是全注释学习版，验证逻辑后保留备查。

---

六、踩坑清单（重点）

1. 中文乱码：Redis 存 JSON 必须 json.dumps(obj, ensure_ascii=False)，否则中文变 \uXXXX。
2. system 人设只注入一次：在 len(messages)==0 时 rpush system，之后每轮都带上，避免重复注入。
3. 会话标题更新用 lset 按索引：先 lrange 找到目标 session 的 index，改完 lset 写回；list 没有「按字段更新」，只能整条替换。
4. 详情接口跳过 system：lrange(key, 1, -1) 从 1 开始，别把 system 人设吐给前端。
5. 压缩请求别污染主对话：压缩用的 prompt 是单独拼的 c_messages，结果作为一条 user 历史顶在前面，不进真正对话轮次。
6. 类型一致性：create_session 的 user_id 是 str，而列表/详情接口是 int，Redis key 拼接时混用会查不到（建议在入口统一转 str）。
7. decode_responses=True：直接拿字符串，记得 json.loads 还原；若用 False（默认）拿到的是 bytes，要 bytes.decode()。
8. 调试 print 残留：代码里还有 print(...)，生产应统一换 logger。
9. Redis 库隔离：demo 用 db=1，生产按业务分库，避免和缓存 / 其他模块混用。
10. protocol=2：兼容老版本 redis-py，升级客户端后可去掉。

---

七、小结

Day2 把单轮调用升级成了可用的多轮对话：

• 会话层：Redis list 存每个用户的会话列表，支持多会话
• 消息层：每个会话一条消息 list，追加式存储，刷新 / 重启不丢
• 上下文层：滑动窗口保留最近 5 条 + 历史语义压缩，解决 token 超限

后续可继续做：

• 流式输出接多轮（SSE + 会话存储）
• 会话标题让模型自动生成（首轮后调一次摘要）
• 消息持久化从 Redis 迁到 MySQL（长期存档），Redis 只做热数据

注：本文代码片段均来自当天真实提交的后端文件（app/apis/llm/case2_api.py、app/schemas/llm_case1.py、main.py），仅做脱敏（API Key 走环境变量）。