1. 多轮对话的场景与基础多轮对话Multi-turn Dialogue是指用户与系统之间进行多次连续交互系统能够记住上下文并根据历史对话内容进行响应的能力。这是构建智能客服、聊天机器人、任务型助手等应用的核心。1.1 核心场景任务型对话用户分步骤完成一个复杂任务如订餐、订票系统需要记住用户已提供的信息如时间、地点、偏好。上下文问答用户基于前文进行追问如“上面提到的那个函数怎么用”系统需要理解指代关系。知识库聊天围绕一个主题进行深入探讨对话历史用于优化检索和生成质量。调试/故障排查技术客服逐步引导用户提供更多日志、截图等信息以定位问题。1.2 技术基础实现多轮对话通常需要解决以下三个核心问题会话管理如何为每个用户/对话创建独立的会话标识Session ID并存储和检索对应的对话历史。上下文维护如何组织、存储和传递历史消息通常为 [{role: user, content: ...}, {role: assistant, content: ...}] 格式。状态管理对于任务型对话如何维护对话状态Dialog State例如当前步骤、已填写的槽位Slots。2. 使用 FastAPI 实现多轮对话FastAPI 凭借其异步特性、高性能和简洁的依赖注入系统非常适合构建多轮对话 API。下面我们将分步骤实现一个完整的示例。2.1 项目结构与依赖首先创建项目并安装依赖mkdir fastapi-multiturn-chat cd fastapi-multiturn-chat python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install fastapi uvicorn python-dotenv redis创建requirements.txtfastapi0.104.1 uvicorn[standard]0.24.0 python-dotenv1.0.0 redis5.0.12.2 核心实现会话管理与存储我们使用 Redis 作为后端存储来维护会话和对话历史。创建app/main.pyfrom fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import List, Optional import redis import json import uuid from datetime import datetime, timedelta app FastAPI(title多轮对话 API) Redis 连接生产环境应使用连接池 redis_client redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) 数据模型 class Message(BaseModel): role: str # user 或 assistant content: str class ChatRequest(BaseModel): session_id: Optional[str] None # 为空则创建新会话 message: str max_history: Optional[int] 10 # 保留的最大历史消息数 class ChatResponse(BaseModel): session_id: str response: str history: List[Message] 依赖项获取或创建会话 def get_or_create_session(session_id: Optional[str] None): if not session_id: session_id str(uuid.uuid4()) # 检查会话是否存在不存在则初始化 if not redis_client.exists(fsession:{session_id}): redis_client.setex(fsession:{session_id}, timedelta(hours24), json.dumps([])) return session_id 核心对话端点 app.post(/chat, response_modelChatResponse) async def chat(request: ChatRequest): 1. 会话管理 session_id get_or_create_session(request.session_id) 2. 获取历史 history_json redis_client.get(fsession:{session_id}) history json.loads(history_json) if history_json else [] 3. 添加用户新消息 user_msg {role: user, content: request.message} history.append(user_msg) 4. 模拟 AI 回复此处可替换为真实 LLM 调用 简单回复回显用户消息并提及历史长度 assistant_response f我已收到您的消息{request.message}。当前会话历史共有 {len(history)} 条消息。 assistant_msg {role: assistant, content: assistant_response} history.append(assistant_msg) 5. 限制历史长度 if request.max_history and len(history) request.max_history * 2: # 乘以2因为每条对话含user和assistant history history[-request.max_history * 2:] 6. 保存更新后的历史 redis_client.setex(fsession:{session_id}, timedelta(hours24), json.dumps(history)) return ChatResponse( session_idsession_id, responseassistant_response, history[Message(**msg) for msg in history] ) 获取会话历史 app.get(/session/{session_id}/history) async def get_history(session_id: str): history_json redis_client.get(fsession:{session_id}) if not history_json: raise HTTPException(status_code404, detailSession not found) return json.loads(history_json) 删除会话 app.delete(/session/{session_id}) async def delete_session(session_id: str): deleted redis_client.delete(fsession:{session_id}) if not deleted: raise HTTPException(status_code404, detailSession not found) return {message: Session deleted} if name main: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)2.3 运行与测试确保 Redis 服务已启动然后运行uvicorn app.main:app --reload --port 8000使用 curl 或 Postman 测试多轮对话# 第一轮创建新会话 curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {message: 你好我想了解FastAPI} 响应会返回 session_id例如{session_id: abc-123, response: ..., history: [...]} 第二轮使用上轮的 session_id 继续对话 curl -X POST http://localhost:8000/chat -H Content-Type: application/json -d { session_id: abc-123, message: 多轮对话怎么实现 }3. 进阶集成真实 LLM 与状态管理3.1 集成 OpenAI GPT安装 openai 库并修改/chat端点import openai import os from dotenv import load_dotenv load_dotenv() openai.api_key os.getenv(OPENAI_API_KEY) 在 /chat 端点中替换模拟回复部分 async def generate_with_gpt(history_messages): # 格式化历史消息为 OpenAI 格式 messages [ {role: msg[role], content: msg[content]} for msg in history_messages ] response await openai.ChatCompletion.acreate( modelgpt-3.5-turbo, messagesmessages, max_tokens500, temperature0.7 ) return response.choices[0].message.content/code/pre 3.2 任务状态管理 对于订餐等任务型对话可扩展会话存储以包含状态 class DialogState(BaseModel): current_step: str greeting slots: dict {} # 例如 {food_type: , quantity: 0, address: } 在 Redis 中存储状态 def update_dialog_state(session_id: str, state: DialogState): redis_client.setex( fstate:{session_id}, timedelta(hours24), state.json() ) 4. 生产环境注意事项 存储选择高并发场景考虑 Redis Cluster 或数据库如 PostgreSQL。 会话过期根据业务设置合理的 TTL如 30 分钟无活动后清除。 上下文窗口LLM 有 token 限制需设计历史摘要或滑动窗口策略。 安全性对用户输入进行过滤防止注入攻击会话 ID 应不可预测。 可观测性记录对话日志监控会话增长和 API 延迟。 5. 总结 通过 FastAPI 实现多轮对话的核心在于 为每个对话分配唯一的 session_id。 使用 Redis 等存储维护 session_id → 对话历史 的映射。 在每次请求中获取历史、添加新消息、调用 LLM、保存更新后的历史。 根据业务需求可扩展加入对话状态管理、历史摘要、流式响应等高级功能。 本文提供的代码是一个可直接运行的基础框架你可以在此基础上集成真实的 LLM 服务如 OpenAI、文心一言、通义千问并添加业务逻辑快速构建出功能完整的多轮对话应用。 FastAPI 多轮对话实现指南：从场景基础到实战代码
1. 多轮对话的场景与基础多轮对话Multi-turn Dialogue是指用户与系统之间进行多次连续交互系统能够记住上下文并根据历史对话内容进行响应的能力。这是构建智能客服、聊天机器人、任务型助手等应用的核心。1.1 核心场景任务型对话用户分步骤完成一个复杂任务如订餐、订票系统需要记住用户已提供的信息如时间、地点、偏好。上下文问答用户基于前文进行追问如“上面提到的那个函数怎么用”系统需要理解指代关系。知识库聊天围绕一个主题进行深入探讨对话历史用于优化检索和生成质量。调试/故障排查技术客服逐步引导用户提供更多日志、截图等信息以定位问题。1.2 技术基础实现多轮对话通常需要解决以下三个核心问题会话管理如何为每个用户/对话创建独立的会话标识Session ID并存储和检索对应的对话历史。上下文维护如何组织、存储和传递历史消息通常为 [{role: user, content: ...}, {role: assistant, content: ...}] 格式。状态管理对于任务型对话如何维护对话状态Dialog State例如当前步骤、已填写的槽位Slots。2. 使用 FastAPI 实现多轮对话FastAPI 凭借其异步特性、高性能和简洁的依赖注入系统非常适合构建多轮对话 API。下面我们将分步骤实现一个完整的示例。2.1 项目结构与依赖首先创建项目并安装依赖mkdir fastapi-multiturn-chat cd fastapi-multiturn-chat python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install fastapi uvicorn python-dotenv redis创建requirements.txtfastapi0.104.1 uvicorn[standard]0.24.0 python-dotenv1.0.0 redis5.0.12.2 核心实现会话管理与存储我们使用 Redis 作为后端存储来维护会话和对话历史。创建app/main.pyfrom fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import List, Optional import redis import json import uuid from datetime import datetime, timedelta app FastAPI(title多轮对话 API) Redis 连接生产环境应使用连接池 redis_client redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) 数据模型 class Message(BaseModel): role: str # user 或 assistant content: str class ChatRequest(BaseModel): session_id: Optional[str] None # 为空则创建新会话 message: str max_history: Optional[int] 10 # 保留的最大历史消息数 class ChatResponse(BaseModel): session_id: str response: str history: List[Message] 依赖项获取或创建会话 def get_or_create_session(session_id: Optional[str] None): if not session_id: session_id str(uuid.uuid4()) # 检查会话是否存在不存在则初始化 if not redis_client.exists(fsession:{session_id}): redis_client.setex(fsession:{session_id}, timedelta(hours24), json.dumps([])) return session_id 核心对话端点 app.post(/chat, response_modelChatResponse) async def chat(request: ChatRequest): 1. 会话管理 session_id get_or_create_session(request.session_id) 2. 获取历史 history_json redis_client.get(fsession:{session_id}) history json.loads(history_json) if history_json else [] 3. 添加用户新消息 user_msg {role: user, content: request.message} history.append(user_msg) 4. 模拟 AI 回复此处可替换为真实 LLM 调用 简单回复回显用户消息并提及历史长度 assistant_response f我已收到您的消息{request.message}。当前会话历史共有 {len(history)} 条消息。 assistant_msg {role: assistant, content: assistant_response} history.append(assistant_msg) 5. 限制历史长度 if request.max_history and len(history) request.max_history * 2: # 乘以2因为每条对话含user和assistant history history[-request.max_history * 2:] 6. 保存更新后的历史 redis_client.setex(fsession:{session_id}, timedelta(hours24), json.dumps(history)) return ChatResponse( session_idsession_id, responseassistant_response, history[Message(**msg) for msg in history] ) 获取会话历史 app.get(/session/{session_id}/history) async def get_history(session_id: str): history_json redis_client.get(fsession:{session_id}) if not history_json: raise HTTPException(status_code404, detailSession not found) return json.loads(history_json) 删除会话 app.delete(/session/{session_id}) async def delete_session(session_id: str): deleted redis_client.delete(fsession:{session_id}) if not deleted: raise HTTPException(status_code404, detailSession not found) return {message: Session deleted} if name main: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)2.3 运行与测试确保 Redis 服务已启动然后运行uvicorn app.main:app --reload --port 8000使用 curl 或 Postman 测试多轮对话# 第一轮创建新会话 curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {message: 你好我想了解FastAPI} 响应会返回 session_id例如{session_id: abc-123, response: ..., history: [...]} 第二轮使用上轮的 session_id 继续对话 curl -X POST http://localhost:8000/chat -H Content-Type: application/json -d { session_id: abc-123, message: 多轮对话怎么实现 }3. 进阶集成真实 LLM 与状态管理3.1 集成 OpenAI GPT安装 openai 库并修改/chat端点import openai import os from dotenv import load_dotenv load_dotenv() openai.api_key os.getenv(OPENAI_API_KEY) 在 /chat 端点中替换模拟回复部分 async def generate_with_gpt(history_messages): # 格式化历史消息为 OpenAI 格式 messages [ {role: msg[role], content: msg[content]} for msg in history_messages ] response await openai.ChatCompletion.acreate( modelgpt-3.5-turbo, messagesmessages, max_tokens500, temperature0.7 ) return response.choices[0].message.content/code/pre 3.2 任务状态管理 对于订餐等任务型对话可扩展会话存储以包含状态 class DialogState(BaseModel): current_step: str greeting slots: dict {} # 例如 {food_type: , quantity: 0, address: } 在 Redis 中存储状态 def update_dialog_state(session_id: str, state: DialogState): redis_client.setex( fstate:{session_id}, timedelta(hours24), state.json() ) 4. 生产环境注意事项 存储选择高并发场景考虑 Redis Cluster 或数据库如 PostgreSQL。 会话过期根据业务设置合理的 TTL如 30 分钟无活动后清除。 上下文窗口LLM 有 token 限制需设计历史摘要或滑动窗口策略。 安全性对用户输入进行过滤防止注入攻击会话 ID 应不可预测。 可观测性记录对话日志监控会话增长和 API 延迟。 5. 总结 通过 FastAPI 实现多轮对话的核心在于 为每个对话分配唯一的 session_id。 使用 Redis 等存储维护 session_id → 对话历史 的映射。 在每次请求中获取历史、添加新消息、调用 LLM、保存更新后的历史。 根据业务需求可扩展加入对话状态管理、历史摘要、流式响应等高级功能。 本文提供的代码是一个可直接运行的基础框架你可以在此基础上集成真实的 LLM 服务如 OpenAI、文心一言、通义千问并添加业务逻辑快速构建出功能完整的多轮对话应用。