测试

先看本地脚本怎么玩——核心思路就一句：每一轮把用户消息和模型回复都 rpush 进同一个 Redis list，下一轮把整个 list 读出来当 messages 发给模型，模型就"记得"前面聊过啥了。

client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="自己",
)

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


key="boss:llm:case4:messages:3"
#第一轮对话
message1={"role":"user","content":"国内大模型有哪些"}
redis_client.rpush(key,json.dumps(message1,ensure_ascii=False))
redis_messaeges=redis_client.lrange(key,0,-1)
messages=[json.loads(message) for message in redis_messaeges]

ai_reply=get_ai_response(messages)
print(f"第一轮大模型的回复:{ai_reply}")
ai_reply_message={"role":"assistant","content":ai_reply}
redis_client.rpush(key,json.dumps(ai_reply_message,ensure_ascii=False))


#第二轮对话
message2={"role":"user","content":"每一个大模型的擅长什么?"}
redis_client.rpush(key,json.dumps(message2,ensure_ascii=False))
redis_messages=redis_client.lrange(key,0,-1)
print(redis_messages)
messages=[json.loads(message) for message in redis_messaeges]

ai_reply=get_ai_response(messages)
print(f"第二轮大模型的回复:{ai_reply}")
ai_reply_message={"role":"assistant","content":ai_reply}
redis_client.rpush(key,json.dumps(ai_reply_message,ensure_ascii=False))

get_ai_response:它就是个搬运工，把拼好的 messages 丢给模型，把 choices[0].message.content 抠出来返回。所有"调模型"的动作都走它，后面接口也复用。

剩下的：

• rpush 是往 list 尾部追加，用户一句、模型一句，顺序天然就对。

• 每轮都 lrange(key, 0, -1) 全量读出，再 json.loads 还原——这就是"带记忆"的关键：发给模型的永远是完整历史。

• 第二轮的提问"每一个大模型擅长什么"能接住第一轮，靠的就是第一轮存的 assistant 回复。

• 原稿里有两个坑：① redis_messaeges 拼写错误，第二轮的读取和还原用了两个不同变量名，会直接 NameError；② 第一轮的 redis_messaeges（变量名）和第二轮的不一致。上面已统一改成 redis_messages。

跑完用 redis-cli lrange boss:llm:case4:messages:3 0 -1 能看到一整段对话原文。

接口

脚本能跑了，但项目里会话得按用户、按会话维度管理。下面四个接口就是干这个的：create_session 建会话、get_session_list 拉列表、send_message 发消息（带上下文压缩）、get_user_session_messages 查详情。

创建会话接口

入参只要 user_id，给这个用户生成一条新会话记录存进 Redis（key 形如 boss:llm:session:{user_id}）。会话标题先默认"新会话"，等用户发第一条消息时我们再拿消息前缀去改它（见 2.3）。返回里带上刚建的会话，前端拿到 session_id 才能继续发消息。

@llm_day02_router.post("/create_session", summary="创建会话")
async def create_session(user_id:int):
    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
        }
    }

获取会话列表接口

按 user_id 把"这个用户建过的所有会话"一次性返回。因为建的时候用的是 lpush，所以这里读出来最新会话排在前面，前端直接渲染就行，不用再手动排序。

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

发送消息接口

这个接口是整套里最重的，干了好几件事：首次发消息补 system 提示词、更新会话标题、把历史读出来、做上下文压缩、调模型、存回复。先把两个被它复用的函数讲清楚。

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

def compression_messages(messages:list):
    logger.info(f"当前消息的长度是:{len(messages)}")
    if len(messages)<=2:
        return messages
    logger.info(f"开始压缩消息")
    recently_messages=messages[-2:]
    logger.info(f"保留最近的2条:{recently_messages}")
    history_messages=messages[:-2]
    logger.info(f"将2条之前的消息进行压缩:{history_messages}")
    c_messages=[{"role":"user","content":f"请将以下用户与大模型的上下文进行语义压缩,要保留核心的关键信息,上下文:{history_messages}"}]
    logger.info(f"大模型压缩后的结果:{c_messages}")
    res=get_ai_response(c_messages)
    return [{"role": "user", "content": res}] + recently_messages



@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}"
    user_session_messages=redis_client.lrange(key, 0, -1)
    if len(user_session_messages)==0:
        system_message={
            "role": "system",
            "content": "你是一个智能助手"
        }
        redis_client.rpush(key, json.dumps(system_message, ensure_ascii=False))

        # key=f'boss:llm:session:{case_request.user_id}'
        # user_sessions=redis_client.lrange(key, 0, -1)
        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(key, index, json.dumps(session_dict, ensure_ascii=False))
                redis_client.lset(session_key, index, json.dumps(session_dict, ensure_ascii=False))

    current_user_messgae={
        "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_messgae, ensure_ascii=False))
    messages=[json.loads(message) for message in redis_client.lrange(key, 0, -1)]
    messages=compression_messages(messages)
    logger.info(f"大模型的输入:{messages}")
    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':'消息发送成功',
    }

compression_messages：防上下文过长。逻辑是「最近 2 条原样留，更早的丢给模型压成一段摘要」，最后拼成 [压缩摘要, 最近2条]。

主接口：

1. 用 user_id + session_id 拼出专属 key，保证不同人、不同会话互不串台。

1. 空会话首条消息时自动补 system 人设，并把会话标题改成消息前缀——这就是 2.1 里"新会话"被改掉的地方。

1. lset 是按下标改 Redis list 里的某条，用来原地更新会话标题，比删了重建省事。

1. 读全部历史 → compression_messages 压缩 → 调模型 → 把 assistant 回复连同时间戳一起存回去。

查询会话详情接口

@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}"
    user_session_messages = redis_client.lrange(key, 1, -1)
    user_sesssion_messages=[json.loads(message) for message in user_session_messages]
    return {
        'code': 1,
        'message': '查询会话详情成功',
        'data': {
            "session_messages": user_sesssion_messages
        }
    }

按 user_id + session_id 把这轮会话的所有消息拉回来。注意 lrange(key, 1, -1) 是从第 1 条（下标 0 是 system）开始取，等于把人设挡在返回之外，前端拿到的就是纯净的"用户↔助手"对话流。user_id 和 session_id 作为 query 参数传，前端点开某个会话时调用。