企业官网建设流程全解析

一、学习核心内容

今天我完成了一个完整的AI 问答小项目，核心是实现基于阿里云百炼千问大模型的多轮对话能力，并通过 Redis 管理对话上下文，同时封装了完整的 Web 接口。

核心功能点

1. 多轮对话管理：通过 Redis 存储用户对话历史，实现上下文记忆
2. 流式输出接口：将大模型回复以 SSE 流式方式推送给前端
3. 对话生命周期管理：创建新对话、获取历史对话、查看对话详情
4. 上下文语义压缩：解决长对话 Token 超限问题，优化调用成本
5. 用户身份关联：通过 Token 获取用户 ID，实现多用户隔离

二、核心模块拆解

1. 多轮对话基础实现

通过 Redis 的 List 结构存储每一轮的用户提问和模型回复，构建完整对话上下文。

python

运行

# 封装模型回复 def get_model_reply(questions: list): completion = client.chat.completions.create( model='qwen-plus', messages=questions ) 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_ascii=False)) 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_ascii=False)) # 第二轮 second_q = {"role": "user", "content": "钢铁侠"} r.rpush(redis_key, json.dumps(second_q, ensure_ascii=False)) 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_ascii=False))

2. 流式输出封装

将大模型的流式回复封装为 SSE 接口，实现实时打字效果。

python

运行

def handle_chunks(messages: list, key: str): client = OpenAI( api_key=os.getenv('DASHSCAPE_API_KEY'), base_url='https://ws-tqrxpklvvj1pxdkp.cn-beijing.maas.aliyuncs.com/compatible-mode/v1' ) completion = client.chat.completions.create( model='qwen-plus', messages=messages, stream=True ) 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 f'data: {chunk.choices[0].delta.content}\n\n' # 流式结束后将完整回复存入Redis answer = {'role': 'assistant', 'content': ''.join(res)} r.rpush(key, json.dumps(answer, ensure_ascii=False)) yield 'data: done!'

3. 对话管理接口

创建新对话

python

运行

@llm_router.post(path='/create_new_talk/', summary='创建新的对话') async def create_new_talk(userid=Depends(getinfo_form_token)): snowflake = SnowflakeSingleton(worker_id=1) talkid = str(snowflake.get_id()) temp_dict = { 'talkid': talkid, 'title': '新对话', 'createtime': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) } user_key = f'p4boss:llm:userid:{userid}' talk_key = f'talkid:{talkid}' r.hset(user_key, talk_key, json.dumps(temp_dict, ensure_ascii=False)) # 初始化系统提示 user_talk_key = f'p4boss:llmtalk:{userid}:{talkid}' system_info = {'role': 'system', 'content': '你是一个智能助手！'} r.rpush(user_talk_key, json.dumps(system_info, ensure_ascii=False)) return {'code': 200, 'msg': '创建成功'}

获取用户所有对话

python

运行

@llm_router.get(path='/get_talks/', summary='获取用户所有历史对话') async def get_talks(userid=Depends(getinfo_form_token)): user_key = f'p4boss:llm:userid:{userid}' temp_data = r.hgetall(user_key) result = [json.loads(j) for i, j in temp_data.items()] result.sort(key=lambda 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, userid=Depends(getinfo_form_token)): user_talk_key = f'p4boss: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, userid=Depends(getinfo_form_token)): question = await question.json() question = question.get('question') or '' user_info = {'role': 'user', 'content': question} user_talk_key = f'p4boss:llmtalk:{userid}:{talkid}' r.rpush(user_talk_key, json.dumps(user_info, ensure_ascii=False)) # 更新对话标题 user_key = f'p4boss:llm:userid:{userid}' talk_key = f'talkid:{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_ascii=False)) # 获取完整上下文并压缩 redis_msgs = r.lrange(user_talk_key, 0, -1) messages = [json.loads(i) for i in redis_msgs] # messages = compression_messages(messages) # 可选：上下文压缩 return StreamingResponse( content=handle_chunks(messages, user_talk_key), media_type="text/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/assistant）

2. 完整交互流程

1. 用户登录，获取 Token
2. 前端调用/create_new_talk创建新对话，得到talkid
3. 用户输入问题，调用/sendmsg接口，传入talkid和问题内容
4. 后端将问题存入 Redis，更新对话标题
5. 后端读取完整对话上下文，调用千问大模型流式接口
6. 后端通过 SSE 将模型回复逐字推送给前端
7. 流式结束后，将完整回复存入 Redis，更新对话上下文
8. 前端可调用/get_talks和/get_talk_detail查看历史对话

四、今日学习总结

✅ 核心收获

1. 掌握了多轮对话实现：理解了如何通过 Redis 存储和管理对话上下文，实现大模型的记忆能力
2. 熟练了流式输出：掌握了 OpenAI 兼容接口的流式调用方式，以及 FastAPI 中 SSE 的实现
3. 完成了工程化封装：将 AI 能力封装为完整的 Web 服务，具备了生产环境使用的基础
4. 了解了上下文优化：学习了通过语义压缩解决长对话 Token 超限的方案

📌 技术要点

• Redis 选型：List 结构天然适合按顺序存储对话消息，Hash 结构适合存储用户 - 对话映射
• 流式传输：SSE 是实现 Web 端实时交互的轻量方案，比 WebSocket 更简单高效
• 上下文管理：通过保留最近 N 轮对话 + 语义压缩，平衡了对话记忆和 Token 成本
• 代码分层：接口层、服务层、数据层分离，代码结构清晰易维护