一、学习核心内容今天我完成了一个完整的AI 问答小项目核心是实现基于阿里云百炼千问大模型的多轮对话能力并通过 Redis 管理对话上下文同时封装了完整的 Web 接口。核心功能点多轮对话管理通过 Redis 存储用户对话历史实现上下文记忆流式输出接口将大模型回复以 SSE 流式方式推送给前端对话生命周期管理创建新对话、获取历史对话、查看对话详情上下文语义压缩解决长对话 Token 超限问题优化调用成本用户身份关联通过 Token 获取用户 ID实现多用户隔离二、核心模块拆解1. 多轮对话基础实现通过 Redis 的 List 结构存储每一轮的用户提问和模型回复构建完整对话上下文。python运行# 封装模型回复 def get_model_reply(questions: list): completion client.chat.completions.create( modelqwen-plus, messagesquestions ) return completion.choices[0].message.content # 示例两轮对话流程 redis_key user_id:temp # 第一轮 first_q {role: user, content: 蜘蛛侠} r.rpush(redis_key, json.dumps(first_q, ensure_asciiFalse)) history_talk [json.loads(i) for i in r.lrange(redis_key, 0, -1)] first_a {role: assistant, content: get_model_reply(history_talk)} r.rpush(redis_key, json.dumps(first_a, ensure_asciiFalse)) # 第二轮 second_q {role: user, content: 钢铁侠} r.rpush(redis_key, json.dumps(second_q, ensure_asciiFalse)) history_talk [json.loads(i) for i in r.lrange(redis_key, 0, -1)] second_a {role: assistant, content: get_model_reply(history_talk)} r.rpush(redis_key, json.dumps(second_a, ensure_asciiFalse))2. 流式输出封装将大模型的流式回复封装为 SSE 接口实现实时打字效果。python运行def handle_chunks(messages: list, key: str): client OpenAI( api_keyos.getenv(DASHSCAPE_API_KEY), base_urlhttps://ws-tqrxpklvvj1pxdkp.cn-beijing.maas.aliyuncs.com/compatible-mode/v1 ) completion client.chat.completions.create( modelqwen-plus, messagesmessages, streamTrue ) res [] for chunk in completion: if chunk.choices: if chunk.choices[0].delta.content is not None: res.append(chunk.choices[0].delta.content) yield fdata: {chunk.choices[0].delta.content}\n\n # 流式结束后将完整回复存入Redis answer {role: assistant, content: .join(res)} r.rpush(key, json.dumps(answer, ensure_asciiFalse)) yield data: done!3. 对话管理接口创建新对话python运行llm_router.post(path/create_new_talk/, summary创建新的对话) async def create_new_talk(useridDepends(getinfo_form_token)): snowflake SnowflakeSingleton(worker_id1) talkid str(snowflake.get_id()) temp_dict { talkid: talkid, title: 新对话, createtime: time.strftime(%Y-%m-%d %H:%M:%S, time.localtime()) } user_key fp4boss:llm:userid:{userid} talk_key ftalkid:{talkid} r.hset(user_key, talk_key, json.dumps(temp_dict, ensure_asciiFalse)) # 初始化系统提示 user_talk_key fp4boss:llmtalk:{userid}:{talkid} system_info {role: system, content: 你是一个智能助手} r.rpush(user_talk_key, json.dumps(system_info, ensure_asciiFalse)) return {code: 200, msg: 创建成功}获取用户所有对话python运行llm_router.get(path/get_talks/, summary获取用户所有历史对话) async def get_talks(useridDepends(getinfo_form_token)): user_key fp4boss:llm:userid:{userid} temp_data r.hgetall(user_key) result [json.loads(j) for i, j in temp_data.items()] result.sort(keylambda x: x[createtime]) return {code: 200, msg: 获取成功, data: result}查看对话详情python运行llm_router.get(path/get_talk_detail/, summary查看用户历史对话记录) async def get_talk_detail(talkid: int, useridDepends(getinfo_form_token)): user_talk_key fp4boss:llmtalk:{userid}:{talkid} temp r.lrange(user_talk_key, 1, -1) # 跳过系统提示 data [json.loads(i) for i in temp] return {code: 200, data: data, msg: ok}发送消息核心接口python运行llm_router.post(path/sendmsg/, summary用户发送消息) async def send_msg(talkid: int, question: Request, useridDepends(getinfo_form_token)): question await question.json() question question.get(question) or user_info {role: user, content: question} user_talk_key fp4boss:llmtalk:{userid}:{talkid} r.rpush(user_talk_key, json.dumps(user_info, ensure_asciiFalse)) # 更新对话标题 user_key fp4boss:llm:userid:{userid} talk_key ftalkid:{talkid} temp json.loads(r.hget(user_key, talk_key)) temp[title] question[:10] ... if len(question) 10 else question r.hset(user_key, talk_key, json.dumps(temp, ensure_asciiFalse)) # 获取完整上下文并压缩 redis_msgs r.lrange(user_talk_key, 0, -1) messages [json.loads(i) for i in redis_msgs] # messages compression_messages(messages) # 可选上下文压缩 return StreamingResponse( contenthandle_chunks(messages, user_talk_key), media_typetext/event-stream )4. 上下文语义压缩解决长对话 Token 超限问题通过大模型对历史上下文进行语义压缩。python运行def compression_messages(messages: list): if len(messages) 5: return messages # 保留最近2轮对话压缩更早的内容 compression messages[:-2] keep messages[-2:] temp_dict [{ role: user, content: f请将以下用户与大模型的上下文进行语义压缩要保留核心的关键信息上下文:{compression} }] res get_ai_response(temp_dict) return [{role: user, content: res}] keep三、项目架构与流程1. 数据存储设计用户 - 对话映射Redis Hash (p4boss:llm:userid:{userid}) → 存储用户所有对话的元信息标题、创建时间对话内容Redis List (p4boss:llmtalk:{userid}:{talkid}) → 按顺序存储每一轮对话消息system/user/assistant2. 完整交互流程用户登录获取 Token前端调用/create_new_talk创建新对话得到talkid用户输入问题调用/sendmsg接口传入talkid和问题内容后端将问题存入 Redis更新对话标题后端读取完整对话上下文调用千问大模型流式接口后端通过 SSE 将模型回复逐字推送给前端流式结束后将完整回复存入 Redis更新对话上下文前端可调用/get_talks和/get_talk_detail查看历史对话四、今日学习总结✅ 核心收获掌握了多轮对话实现理解了如何通过 Redis 存储和管理对话上下文实现大模型的记忆能力熟练了流式输出掌握了 OpenAI 兼容接口的流式调用方式以及 FastAPI 中 SSE 的实现完成了工程化封装将 AI 能力封装为完整的 Web 服务具备了生产环境使用的基础了解了上下文优化学习了通过语义压缩解决长对话 Token 超限的方案 技术要点Redis 选型List 结构天然适合按顺序存储对话消息Hash 结构适合存储用户 - 对话映射流式传输SSE 是实现 Web 端实时交互的轻量方案比 WebSocket 更简单高效上下文管理通过保留最近 N 轮对话 语义压缩平衡了对话记忆和 Token 成本代码分层接口层、服务层、数据层分离代码结构清晰易维护