1. 多轮对话的场景与基础

多轮对话（Multi-turn Dialogue）是指用户与系统之间进行多次连续交互，系统能够记住上下文并根据历史对话内容进行响应的能力。这是构建智能客服、聊天机器人、任务型助手等应用的核心。

1.1 核心场景

• 任务型对话：用户分步骤完成一个复杂任务（如订餐、订票），系统需要记住用户已提供的信息（如时间、地点、偏好）。
• 上下文问答：用户基于前文进行追问（如“上面提到的那个函数怎么用？”），系统需要理解指代关系。
• 知识库聊天：围绕一个主题进行深入探讨，对话历史用于优化检索和生成质量。
• 调试/故障排查：技术客服逐步引导用户提供更多日志、截图等信息以定位问题。

1.2 技术基础

实现多轮对话通常需要解决以下三个核心问题：

1. 会话管理：如何为每个用户/对话创建独立的会话标识（Session ID），并存储和检索对应的对话历史。
2. 上下文维护：如何组织、存储和传递历史消息（通常为 [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}] 格式）。
3. 状态管理：对于任务型对话，如何维护对话状态（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.txt：

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-dotenv==1.0.0
redis==5.0.1

2.2 核心实现：会话管理与存储

我们使用 Redis 作为后端存储来维护会话和对话历史。创建 app/main.py：

from 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(host='localhost', port=6379, db=0, decode_responses=True)
数据模型
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(f"session:{session_id}"):
    redis_client.setex(f"session:{session_id}", timedelta(hours=24), json.dumps([]))
return session_id
核心对话端点
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
1. 会话管理
session_id = get_or_create_session(request.session_id)
2. 获取历史
history_json = redis_client.get(f"session:{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(f"session:{session_id}", timedelta(hours=24), json.dumps(history))
return ChatResponse(
session_id=session_id,
response=assistant_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(f"session:{session_id}")
if not history_json:
raise HTTPException(status_code=404, detail="Session not found")
return json.loads(history_json)
删除会话
@app.delete("/session/{session_id}")
async def delete_session(session_id: str):
deleted = redis_client.delete(f"session:{session_id}")
if not deleted:
raise HTTPException(status_code=404, detail="Session not found")
return {"message": "Session deleted"}
if name == "main":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

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(
    model="gpt-3.5-turbo",
    messages=messages,
    max_tokens=500,
    temperature=0.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(
f"state:{session_id}",
timedelta(hours=24),
state.json()
)
4. 生产环境注意事项
存储选择：高并发场景考虑 Redis Cluster 或数据库（如 PostgreSQL）。
会话过期：根据业务设置合理的 TTL（如 30 分钟无活动后清除）。
上下文窗口：LLM 有 token 限制，需设计历史摘要或滑动窗口策略。
安全性：对用户输入进行过滤，防止注入攻击；会话 ID 应不可预测。
可观测性：记录对话日志，监控会话增长和 API 延迟。
5. 总结
通过 FastAPI 实现多轮对话的核心在于：
为每个对话分配唯一的 session_id。
使用 Redis 等存储维护 session_id → 对话历史 的映射。
在每次请求中，获取历史、添加新消息、调用 LLM、保存更新后的历史。
根据业务需求，可扩展加入对话状态管理、历史摘要、流式响应等高级功能。
本文提供的代码是一个可直接运行的基础框架，你可以在此基础上集成真实的 LLM 服务（如 OpenAI、文心一言、通义千问），并添加业务逻辑，快速构建出功能完整的多轮对话应用。