一、项目概述

本教程基于 FastAPI 搭建接口服务，使用 Redis List 持久存储会话聊天记录，对接阿里云百炼通义千问大模型，实现：

1. 创建独立会话（多个对话窗口隔离）
2. 会话列表查询
3. 多轮连续对话，上下文记忆
4. 自动上下文压缩，防止超长文本超限
5. 查询单轮会话全部聊天记录

技术栈：FastAPI、Redis、OpenAI 兼容接口（阿里云百炼）、Python3.10+ 适合新手：跟着步骤部署，直接可运行。

二、核心概念讲解（新手必看）

1. 什么是会话（session）

用户每开启一个聊天窗口 = 一个会话，拥有唯一 session_id。 不同会话之间消息互相隔离；同一个 session 内，大模型可以读取历史聊天内容，实现多轮对话。

2. 为什么用 Redis List 存储对话

• List 有序，可以按照聊天先后顺序保存消息
• rpush 追加最新消息，天然适配聊天时序
• 轻量、读写速度快，适合临时会话缓存

3. 上下文压缩是什么问题？

大模型存在上下文长度限制，聊天轮数越多，历史消息越长。 如果一直携带全部历史，会： ① 消耗更多 token，扣费变高 ② 超出模型最大上下文长度，调用报错 解决方案：保留最近几轮对话，把更早历史交给大模型总结压缩。

4. 存储 Key 设计规范（非常重要）

plaintext

# 用户会话列表（存放该用户所有会话信息）
boss:llm:session:{user_id}

# 单个会话下全部聊天消息
boss:llm:messages:{user_id}:{session_id}

通过 user_id + session_id 双层隔离，支持多用户、多窗口同时聊天。

三、前置环境准备

1. 本地启动 Redis（默认端口 6379）
2. Python 依赖安装

bash

pip install fastapi uvicorn redis openai python-multipart

1. 阿里云百炼准备

• 申请 API_KEY
• 获取兼容模式 BaseURL
• 环境变量配置 DASHSCOPE_API_KEY=sk-xxx

四、完整代码实现

新建路由文件 llm_day02.py

python

运行

import json
import os
import uuid
from datetime import datetime

import redis
from fastapi import APIRouter
from openai import OpenAI

from app.schemas.llm_case1 import LlmCase2
from app.core.logging import logger

# 路由初始化
llm_day02_router = APIRouter(
    prefix="/llm-day02",
    tags=["LLM-DAY02-多轮对话"]
)

# Redis连接
redis_client = redis.Redis(
    host="localhost",
    port=6379,
    db=0,
    decode_responses=True,
    protocol=2
)

# 阿里云百炼OpenAI兼容客户端
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),#存放API KEY
    base_url="https://{存放你的业务空间ID}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

# ====================== 接口1：创建会话 ======================
@llm_day02_router.post("/create_session",summary="创建会话")
async def create_session(user_id: int):
    """用户新建聊天窗口，生成唯一session_id"""
    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
        }
    }

# ====================== 接口2：获取用户全部会话列表 ======================
@llm_day02_router.post("/get_session",summary="会话列表")
async def get_session_list(user_id: int):
    """查询该用户所有聊天窗口"""
    key = f"boss:llm:session:{user_id}"
    user_sessions = [json.loads(session) for session in redis_client.lrange(key, 0, -1)]
    return {
        "code":1,
        "message":"查询会话列表成功",
        "data":{
            "user_sessions": user_sessions
        }
    }

# ====================== 工具函数：调用大模型 ======================
def get_ai_response(messages:list):
    """传入消息数组，获取模型回复文本"""
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=messages
    )
    return completion.choices[0].message.content

# ====================== 工具函数：历史消息压缩 ======================
def compression_messages(messages:list):
    """
    消息压缩策略：
    如果消息数量较少，不压缩
    消息较多：保留最近2轮完整对话，更早历史进行摘要压缩
    """
    if len(messages) <= 2:
        logger.info("不需要压缩")
        return messages
    logger.info("需要压缩")
    # 最近两轮完整保留
    recently_messages = messages[-2:]
    # 更早历史提取出来进行压缩
    history_messages = messages[:-2]

    c_messages = [{
        "role":"user",
        "content":f"请将以下用户与大模型的上下文进行语义压缩,要保留核心的关键信息,上下文:{history_messages}"
    }]
    res = get_ai_response(c_messages)
    # 压缩摘要 + 最新对话，组成新上下文
    return [{"role":"user","content":res}]+recently_messages

# ====================== 接口3：发送消息，实现多轮对话【核心】 ======================
@llm_day02_router.post("/send_message",summary="发送消息")
async def send_message(case_request: LlmCase2):
    """
    多轮对话核心接口
    逻辑流程：
    1. 根据user_id+session_id读取当前会话历史
    2. 首次对话写入系统提示词
    3. 保存用户最新提问
    4. 读取全部历史消息，执行上下文压缩
    5. 请求大模型得到回答
    6. 保存AI回复至Redis
    """
    key = f"boss:llm:messages:{case_request.user_id}:{case_request.session_id}"
    user_session_messages = redis_client.lrange(key,0,-1)

    # 第一轮对话：初始化system角色设定 + 更新会话标题
    if len(user_session_messages) == 0:
        system_message = {
            "role":"system",
            "content":"你是一个智能助手"
        }
        redis_client.rpush(key, json.dumps(system_message, ensure_ascii=False))

        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)
            # 匹配当前会话，设置标题为第一条消息前10个字
            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))

    # 保存用户本次提问
    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))

    # 读取完整对话记录
    messages = [json.loads(message) for message in redis_client.lrange(key,0,-1)]
    # 上下文压缩处理
    messages = compression_messages(messages)
    # 请求大模型
    ai_reply = get_ai_response(messages)
    # 存储AI回复
    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":"发送消息成功"
    }

# ====================== 接口4：查询会话全部聊天记录 ======================
@llm_day02_router.post("/get_user_session_message",summary="查询会话详情")
async def get_user_session_message(user_id:int,session_id:str):
    """读取指定会话所有聊天内容（跳过system系统提示词）"""
    key = f"boss:llm:messages:{user_id}:{session_id}"
    raw_msg_list = redis_client.lrange(key,1,-1)
    # json字符串转为对象，方便前端渲染
    user_session_message = [json.loads(item) for item in raw_msg_list]
    return {
        "code":1,
        "message":"查询会话详情成功",
        "data":{
            "user_session_message":user_session_message
        }
    }

五、请求模型定义 app/schemas/llm_case1.py

python

运行

from pydantic import BaseModel

class LlmCase2(BaseModel):
    user_id: int
    session_id: str
    message: str

六、完整执行步骤（新手一步步操作）

步骤 1：启动 Redis

本地打开 redis 服务，确保 127.0.0.1:6379 正常连通。

步骤 2：配置环境变量

Windows cmd：

plaintext

set DASHSCOPE_API_KEY=sk-你的阿里云百炼key

步骤 3：项目注册路由

在 main.py 中导入并挂载路由

python

运行

from app.api.llm_day02 import llm_day02_router
app.include_router(llm_day02_router)

步骤 4：启动 FastAPI 服务

bash

uvicorn main:app --reload

访问文档地址：http://127.0.0.1:8000/docs

七、接口调用顺序（演示完整对话流程）

1. 创建会话 POST /llm-day02/create_session 参数：user_id = 1001 返回得到 session_id，保存下来后续所有接口都要用。

2. 发送第一条消息 POST /llm-day02/send_message

json

{
  "user_id":1001,
  "session_id":"刚刚获取的session_id",
  "message":"你好，介绍一下自己"
}

1. 继续发送第二条消息（多轮测试）

json

{
  "user_id":1001,
  "session_id":"同一个session_id",
  "message":"刚刚你说了什么？"
}

此时模型能够记住上一轮对话内容，实现上下文记忆。

1. 查询会话聊天记录：/llm-day02/get_user_session_message
2. 获取所有会话列表：/llm-day02/get_session

八、底层运行流程拆解（send_message 核心逻辑）

1. 根据 user_id + session_id，从 Redis 读取历史消息
2. 判断是否首次对话：首次对话存入 system 提示词（设定助手身份）
3. 将用户新消息追加存入 Redis 列表
4. 读取全部历史消息，执行上下文压缩函数
5. 将处理后的消息数组传给大模型
6. 获取 AI 回答，存入 Redis 持久保存
7. 返回成功，前端刷新对话记录