项目 4: 多轮对话系统¶
⚠️ 时效性说明：本章涉及前沿模型/价格/榜单等信息，可能随版本快速变化；请以论文原文、官方发布页和 API 文档为准。
难度: ⭐⭐⭐⭐ 较难 时间: 15-20 小时 涉及知识: 对话管理、上下文理解、记忆机制、 LangChain
📖 项目概述¶
项目背景¶
单轮对话系统只能处理独立的问答，无法理解对话的上下文和历史信息。多轮对话系统能够维护对话历史，理解上下文，进行连贯的多轮交互。这在智能客服、虚拟助手、教育辅导等场景中非常重要。
项目目标¶
构建一个完整的多轮对话系统，能够： - 维护对话历史和上下文 - 理解用户意图和槽位 - 进行多轮对话管理 - 支持对话状态跟踪 - 提供个性化回复 - 支持多用户并发对话
技术栈¶
- 对话框架: LangChain, Rasa
- LLM: OpenAI GPT-4 / 通义千问
- 向量数据库: ChromaDB
- 后端框架: FastAPI
- 前端框架: Streamlit
- 状态管理: Redis
🏗️ 项目结构¶
Text Only
multi-turn-dialogue-system/
├── app/ # 应用主目录
│ ├── __init__.py
│ ├── config.py # 配置文件
│ ├── dialogue_manager.py # 对话管理器
│ ├── context_manager.py # 上下文管理器
│ ├── intent_classifier.py # 意图分类器
│ ├── slot_filler.py # 槽位填充器
│ ├── response_generator.py # 回复生成器
│ ├── memory/ # 记忆模块
│ │ ├── __init__.py
│ │ ├── short_term.py # 短期记忆
│ │ ├── long_term.py # 长期记忆
│ │ └── vector_store.py # 向量存储
│ └── api/ # API接口
│ ├── __init__.py
│ ├── chat.py # 聊天API
│ └── sessions.py # 会话管理API
├── frontend/ # 前端目录
│ ├── app.py # Streamlit应用
│ └── components/ # UI组件
├── data/ # 数据目录
│ ├── intents/ # 意图数据
│ ├── responses/ # 回复模板
│ └── knowledge_base/ # 知识库
├── models/ # 模型目录
│ ├── intent_classifier/ # 意图分类模型
│ └── embeddings/ # 嵌入模型
├── tests/ # 测试目录
│ ├── test_dialogue_manager.py
│ ├── test_context_manager.py
│ └── test_response_generator.py
├── utils/ # 工具函数
│ ├── __init__.py
│ ├── text_utils.py # 文本工具
│ └── time_utils.py # 时间工具
├── requirements.txt # Python依赖
└── README.md # 项目说明
🎯 核心功能¶
1. 对话管理¶
- 会话管理: 管理多个用户的会话
- 对话历史: 维护完整的对话历史
- 上下文理解: 理解当前对话的上下文
- 状态跟踪: 跟踪对话状态
2. 意图识别¶
- 意图分类: 识别用户意图
- 槽位提取: 提取关键信息
- 实体识别: 识别命名实体
- 意图消歧: 处理模糊意图
3. 回复生成¶
- 模板回复: 使用预定义模板
- 生成式回复: 使用 LLM 生成回复
- 个性化回复: 根据用户画像个性化
- 多模态回复: 支持文本、图片等
4. 记忆机制¶
- 短期记忆: 存储当前会话信息
- 长期记忆: 存储用户历史信息
- 向量检索: 基于向量检索历史对话
- 记忆更新: 动态更新记忆
5. 对话策略¶
- 主动提问: 主动询问缺失信息
- 澄清确认: 澄清模糊信息
- 话题转换: 处理话题转换
- 对话结束: 识别对话结束时机
💻 代码实现¶
1. 配置文件 (app/config.py)¶
Python
"""
多轮对话系统配置文件
"""
import os
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""应用配置"""
# API配置
API_HOST: str = "0.0.0.0"
API_PORT: int = 8000
API_PREFIX: str = "/api/v1"
# LLM配置
OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
OPENAI_MODEL: str = "gpt-4o"
OPENAI_TEMPERATURE: float = 0.7
OPENAI_MAX_TOKENS: int = 500
# 记忆配置
MAX_HISTORY_LENGTH: int = 10
MAX_SHORT_TERM_MEMORY: int = 100
MAX_LONG_TERM_MEMORY: int = 1000
# 向量数据库配置
CHROMA_PERSIST_DIR: str = "./data/vector_store"
CHROMA_COLLECTION_NAME: str = "dialogue_history"
# 对话配置
MAX_TURNS: int = 20
TIMEOUT_SECONDS: int = 300
# Redis配置
REDIS_HOST: str = "localhost"
REDIS_PORT: int = 6379
REDIS_DB: int = 0
# 意图分类配置
INTENT_MODEL_PATH: str = "./models/intent_classifier"
CONFIDENCE_THRESHOLD: float = 0.7
class Config:
env_file = ".env"
settings = Settings()
2. 对话管理器 (app/dialogue_manager.py)¶
Python
"""
对话管理器
"""
from datetime import datetime
import json
class DialogueManager:
"""对话管理器"""
def __init__(self, max_history_length: int = 10):
"""
初始化对话管理器
Args:
max_history_length: 最大历史长度
"""
self.max_history_length = max_history_length
self.dialogues: dict[str, list[dict]] = {}
def create_session(self, session_id: str) -> bool:
"""
创建新会话
Args:
session_id: 会话ID
Returns:
是否创建成功
"""
if session_id in self.dialogues:
return False
self.dialogues[session_id] = {
'history': [],
'context': {},
'state': 'active',
'created_at': datetime.now().isoformat(),
'last_activity': datetime.now().isoformat(),
}
return True
def add_message(
self,
session_id: str,
role: str,
content: str,
metadata: dict | None = None,
) -> bool:
"""
添加消息到对话历史
Args:
session_id: 会话ID
role: 角色 (user/assistant/system)
content: 消息内容
metadata: 元数据
Returns:
是否添加成功
"""
if session_id not in self.dialogues:
return False
dialogue = self.dialogues[session_id]
message = {
'role': role,
'content': content,
'timestamp': datetime.now().isoformat(),
'metadata': metadata or {},
}
dialogue['history'].append(message)
# 限制历史长度
if len(dialogue['history']) > self.max_history_length:
dialogue['history'] = dialogue['history'][-self.max_history_length:]
dialogue['last_activity'] = datetime.now().isoformat()
return True
def get_history(self, session_id: str) -> list[dict]:
"""
获取对话历史
Args:
session_id: 会话ID
Returns:
对话历史
"""
if session_id not in self.dialogues:
return []
return self.dialogues[session_id]['history']
def update_context(
self,
session_id: str,
context: dict,
) -> bool:
"""
更新对话上下文
Args:
session_id: 会话ID
context: 上下文信息
Returns:
是否更新成功
"""
if session_id not in self.dialogues:
return False
self.dialogues[session_id]['context'].update(context)
return True
def get_context(self, session_id: str) -> dict:
"""
获取对话上下文
Args:
session_id: 会话ID
Returns:
上下文信息
"""
if session_id not in self.dialogues:
return {}
return self.dialogues[session_id]['context']
def update_state(
self,
session_id: str,
state: str,
) -> bool:
"""
更新对话状态
Args:
session_id: 会话ID
state: 对话状态
Returns:
是否更新成功
"""
if session_id not in self.dialogues:
return False
self.dialogues[session_id]['state'] = state
return True
def get_state(self, session_id: str) -> str:
"""
获取对话状态
Args:
session_id: 会话ID
Returns:
对话状态
"""
if session_id not in self.dialogues:
return 'inactive'
return self.dialogues[session_id]['state']
def delete_session(self, session_id: str) -> bool:
"""
删除会话
Args:
session_id: 会话ID
Returns:
是否删除成功
"""
if session_id not in self.dialogues:
return False
del self.dialogues[session_id]
return True
def get_session_info(self, session_id: str) -> dict | None:
"""
获取会话信息
Args:
session_id: 会话ID
Returns:
会话信息
"""
if session_id not in self.dialogues:
return None
dialogue = self.dialogues[session_id]
return {
'session_id': session_id,
'state': dialogue['state'],
'message_count': len(dialogue['history']),
'context': dialogue['context'],
'created_at': dialogue['created_at'],
'last_activity': dialogue['last_activity'],
}
3. 上下文管理器 (app/context_manager.py)¶
Python
"""
上下文管理器
"""
from datetime import datetime
import re
class ContextManager:
"""上下文管理器"""
def __init__(self):
"""初始化上下文管理器"""
self.contexts: dict[str, dict] = {}
def create_context(self, session_id: str) -> dict:
"""
创建新上下文
Args:
session_id: 会话ID
Returns:
上下文字典
"""
context = {
'session_id': session_id,
'user_profile': {},
'current_topic': None,
'mentioned_entities': [],
'previous_intents': [],
'slot_values': {},
'created_at': datetime.now().isoformat(),
}
self.contexts[session_id] = context
return context
def update_user_profile(
self,
session_id: str,
profile_data: dict,
) -> bool:
"""
更新用户画像
Args:
session_id: 会话ID
profile_data: 用户画像数据
Returns:
是否更新成功
"""
if session_id not in self.contexts:
return False
self.contexts[session_id]['user_profile'].update(profile_data)
return True
def set_current_topic(
self,
session_id: str,
topic: str,
) -> bool:
"""
设置当前话题
Args:
session_id: 会话ID
topic: 话题
Returns:
是否设置成功
"""
if session_id not in self.contexts:
return False
self.contexts[session_id]['current_topic'] = topic
return True
def add_mentioned_entity(
self,
session_id: str,
entity: str,
entity_type: str,
) -> bool:
"""
添加提及的实体
Args:
session_id: 会话ID
entity: 实体
entity_type: 实体类型
Returns:
是否添加成功
"""
if session_id not in self.contexts:
return False
entity_info = {
'entity': entity,
'type': entity_type,
'timestamp': datetime.now().isoformat(),
}
self.contexts[session_id]['mentioned_entities'].append(entity_info)
return True
def update_slot_value(
self,
session_id: str,
slot_name: str,
slot_value: str,
) -> bool:
"""
更新槽位值
Args:
session_id: 会话ID
slot_name: 槽位名称
slot_value: 槽位值
Returns:
是否更新成功
"""
if session_id not in self.contexts:
return False
self.contexts[session_id]['slot_values'][slot_name] = slot_value
return True
def get_slot_value(
self,
session_id: str,
slot_name: str,
) -> str | None:
"""
获取槽位值
Args:
session_id: 会话ID
slot_name: 槽位名称
Returns:
槽位值
"""
if session_id not in self.contexts:
return None
return self.contexts[session_id]['slot_values'].get(slot_name)
def get_context_summary(
self,
session_id: str,
) -> dict | None:
"""
获取上下文摘要
Args:
session_id: 会话ID
Returns:
上下文摘要
"""
if session_id not in self.contexts:
return None
context = self.contexts[session_id]
return {
'session_id': session_id,
'current_topic': context['current_topic'],
'recent_entities': context['mentioned_entities'][-5:],
'filled_slots': context['slot_values'],
'user_profile': context['user_profile'],
}
def resolve_reference(
self,
session_id: str,
reference: str,
) -> str | None:
"""
解析指代
Args:
session_id: 会话ID
reference: 指代词
Returns:
解析后的实体
"""
if session_id not in self.contexts:
return None
context = self.contexts[session_id]
# 解析代词
if reference in ['他', '她', '它']:
# 返回最近提及的人名/物名
for entity_info in reversed(context['mentioned_entities']):
if entity_info['type'] in ['PERSON', 'OBJECT']:
return entity_info['entity']
elif reference in ['这个', '那个']:
# 返回最近提及的实体
if context['mentioned_entities']:
return context['mentioned_entities'][-1]['entity'] # [-1]负索引取最后一个元素
return None
4. 回复生成器 (app/response_generator.py)¶
Python
"""
回复生成器
"""
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
class ResponseGenerator:
"""回复生成器"""
def __init__(
self,
model_name: str = "gpt-4o",
temperature: float = 0.7,
max_tokens: int = 500,
):
"""
初始化回复生成器
Args:
model_name: 模型名称
temperature: 温度参数
max_tokens: 最大token数
"""
self.llm = ChatOpenAI(
model_name=model_name,
temperature=temperature,
max_tokens=max_tokens,
)
# 使用LCEL方式构建对话链
self.prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的AI助手,请根据上下文和对话历史进行自然的多轮对话。"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}"),
])
self.chain = self.prompt | self.llm
self.history: list = [] # 对话历史(Message对象列表)
def generate_response(
self,
user_input: str,
context: dict | None = None,
history: list[dict] | None = None,
) -> str:
"""
生成回复
Args:
user_input: 用户输入
context: 对话上下文
history: 对话历史
Returns:
生成的回复
"""
# 构建提示
prompt = self._build_prompt(user_input, context, history)
# 生成回复(LCEL invoke方式)
response = self.chain.invoke({
"history": self.history,
"input": prompt,
})
# 更新对话历史
self.history.append(HumanMessage(content=prompt))
self.history.append(AIMessage(content=response.content))
return response.content
def _build_prompt(
self,
user_input: str,
context: dict | None = None,
history: list[dict] | None = None,
) -> str:
"""
构建提示
Args:
user_input: 用户输入
context: 对话上下文
history: 对话历史
Returns:
提示字符串
"""
prompt_parts = []
# 添加上下文信息
if context:
if context.get('current_topic'):
prompt_parts.append(f"当前话题: {context['current_topic']}")
if context.get('filled_slots'):
slots_info = ", ".join(
f"{k}={v}"
for k, v in context['filled_slots'].items()
)
prompt_parts.append(f"已知信息: {slots_info}")
# 添加对话历史
if history:
prompt_parts.append("\n对话历史:")
for msg in history[-3:]: # 只使用最近3轮
role = "用户" if msg['role'] == 'user' else "助手"
prompt_parts.append(f"{role}: {msg['content']}")
# 添加当前输入
prompt_parts.append(f"\n用户: {user_input}")
prompt_parts.append("助手:")
return "\n".join(prompt_parts)
def generate_template_response(
self,
template: str,
variables: dict[str, str],
) -> str:
"""
生成模板回复
Args:
template: 回复模板
variables: 变量字典
Returns:
生成的回复
"""
return template.format(**variables)
def generate_clarification(
self,
missing_slots: list[str],
) -> str:
"""
生成澄清问题
Args:
missing_slots: 缺失的槽位列表
Returns:
澄清问题
"""
if len(missing_slots) == 1:
return f"请问{missing_slots[0]}是什么？"
else:
slots_str = "、".join(missing_slots)
return f"请问{slots_str}分别是什么？"
5. Streamlit 前端 (frontend/app.py)¶
Python
"""
多轮对话系统Web界面
"""
import streamlit as st
from app.dialogue_manager import DialogueManager
from app.context_manager import ContextManager
from app.response_generator import ResponseGenerator
from app.config import settings
import uuid
# 页面配置
st.set_page_config(
page_title="多轮对话系统",
page_icon="💬",
layout="wide"
)
# 初始化会话状态
if 'session_id' not in st.session_state:
st.session_state.session_id = str(uuid.uuid4())
if 'dialogue_manager' not in st.session_state:
st.session_state.dialogue_manager = DialogueManager(
max_history_length=settings.MAX_HISTORY_LENGTH
)
st.session_state.dialogue_manager.create_session(st.session_state.session_id)
if 'context_manager' not in st.session_state:
st.session_state.context_manager = ContextManager()
st.session_state.context_manager.create_context(st.session_state.session_id)
if 'response_generator' not in st.session_state:
st.session_state.response_generator = ResponseGenerator(
model_name=settings.OPENAI_MODEL,
temperature=settings.OPENAI_TEMPERATURE,
max_tokens=settings.OPENAI_MAX_TOKENS,
)
# 标题
st.title("💬 多轮对话系统")
st.markdown("---")
# 获取管理器
dialogue_manager = st.session_state.dialogue_manager
context_manager = st.session_state.context_manager
response_generator = st.session_state.response_generator
session_id = st.session_state.session_id
# 侧边栏
st.sidebar.header("对话信息")
# 显示会话ID
st.sidebar.text(f"会话ID: {session_id}")
# 显示对话状态
state = dialogue_manager.get_state(session_id)
st.sidebar.metric("对话状态", state)
# 显示消息数量
history = dialogue_manager.get_history(session_id)
st.sidebar.metric("消息数量", len(history))
# 显示上下文信息
context = context_manager.get_context_summary(session_id)
if context:
st.sidebar.subheader("上下文信息")
if context.get('current_topic'):
st.sidebar.text(f"话题: {context['current_topic']}")
if context.get('filled_slots'):
st.sidebar.text("已知信息:")
for slot, value in context['filled_slots'].items():
st.sidebar.text(f" {slot}: {value}")
# 主界面
col1, col2 = st.columns([3, 1])
with col1:
st.subheader("对话历史")
# 显示对话历史
chat_container = st.container()
for msg in history:
if msg['role'] == 'user':
with chat_container.chat_message("user"):
st.write(msg['content'])
else:
with chat_container.chat_message("assistant"):
st.write(msg['content'])
# 用户输入
user_input = st.chat_input("请输入您的消息...")
if user_input:
# 添加用户消息
dialogue_manager.add_message(
session_id,
'user',
user_input,
)
# 获取上下文和历史
context = context_manager.get_context_summary(session_id)
history = dialogue_manager.get_history(session_id)
# 生成回复
with st.spinner("正在思考..."):
response = response_generator.generate_response(
user_input,
context,
history,
)
# 添加助手消息
dialogue_manager.add_message(
session_id,
'assistant',
response,
)
# 显示回复
with chat_container.chat_message("assistant"):
st.write(response)
# 重新运行页面以更新显示
st.rerun()
with col2:
st.subheader("操作")
# 清空对话
if st.button("清空对话"):
dialogue_manager.delete_session(session_id)
dialogue_manager.create_session(session_id)
context_manager.create_context(session_id)
st.rerun()
# 显示会话详情
if st.button("查看会话详情"):
session_info = dialogue_manager.get_session_info(session_id)
st.json(session_info)
🧪 测试方法¶
1. 单元测试¶
Python
"""
单元测试示例
"""
import pytest
from app.dialogue_manager import DialogueManager
def test_dialogue_manager():
"""测试对话管理器"""
manager = DialogueManager()
session_id = "test_session"
# 创建会话
assert manager.create_session(session_id) == True # assert断言：条件False时抛出AssertionError
assert manager.create_session(session_id) == False
# 添加消息
assert manager.add_message(session_id, 'user', 'Hello') == True
# 获取历史
history = manager.get_history(session_id)
assert len(history) == 1
assert history[0]['content'] == 'Hello'
# 删除会话
assert manager.delete_session(session_id) == True
2. 集成测试¶
Python
"""
集成测试示例
"""
def test_multi_turn_dialogue():
"""测试多轮对话"""
manager = DialogueManager()
context_manager = ContextManager()
response_generator = ResponseGenerator()
session_id = "test_session"
manager.create_session(session_id)
context_manager.create_context(session_id)
# 第一轮对话
manager.add_message(session_id, 'user', '我叫小明')
context_manager.update_user_profile(session_id, {'name': '小明'})
history = manager.get_history(session_id)
context = context_manager.get_context_summary(session_id)
response = response_generator.generate_response('我叫小明', context, history)
manager.add_message(session_id, 'assistant', response)
# 第二轮对话
manager.add_message(session_id, 'user', '我今年20岁')
context_manager.update_user_profile(session_id, {'age': '20'})
history = manager.get_history(session_id)
context = context_manager.get_context_summary(session_id)
response = response_generator.generate_response('我今年20岁', context, history)
# 验证回复包含用户信息
assert '小明' in response or '20' in response
📊 扩展建议¶
1. 功能扩展¶
- 多模态支持: 支持图片、语音输入
- 情感分析: 识别用户情感并调整回复
- 知识图谱: 集成知识图谱增强对话
- 个性化推荐: 基于用户画像推荐内容
- 多语言支持: 支持多语言对话
2. 性能优化¶
- 缓存机制: 缓存常见回复
- 异步处理: 使用异步处理提升性能
- 模型压缩: 使用小模型提升响应速度
- 分布式部署: 支持分布式部署
3. 智能增强¶
- 主动学习: 从对话中学习
- 对话策略优化: 优化对话策略
- 意图迁移: 处理意图迁移
- 话题管理: 智能管理话题转换
📚 学习收获¶
完成本项目后，你将掌握：
- ✅ 多轮对话系统架构设计
- ✅ 对话状态跟踪技术
- ✅ 上下文理解和管理
- ✅ LangChain 框架应用
- ✅ 记忆机制设计
- ✅ Streamlit Web 应用开发
- ✅ 完整的对话系统开发流程
🔗 参考资源¶
项目完成时间: 15-20 小时 难度等级: ⭐⭐⭐⭐ 较难 推荐指数: ⭐⭐⭐⭐⭐