在前面的文章中，我们讨论了 AI Agent 的各个核心系统。今天，我想聊聊如何实现一个好用的多轮对话系统。说实话，这个话题我琢磨了很久，因为它直接影响到用户体验。
从一个槽点说起
还记得我最开始做对话系统时的一个典型场景：
用户：帮我查一下订单状态
助手：好的，请提供订单号。
用户：A12345
助手：抱歉，我需要您提供订单号才能查询。
用户：...我刚才不是说了吗？这种对话体验显然很糟糕。问题出在哪里？主要是 AI Agent 没有正确理解和维护对话上下文。后来我总结了几个关键点：
- 要理解用户意图
- 要记住上下文信息
- 要有状态管理机制
- 要能处理各种异常情况
对话系统的设计
经过多次迭代，我设计了一个相对完善的对话系统：
from typing import List, Dict, Any, Optional
from enum import Enum
from datetime import datetime
from pydantic import BaseModel
import asyncioclass DialogueState(Enum):IDLE = "idle"WAITING_INPUT = "waiting_input"PROCESSING = "processing"ERROR = "error"COMPLETED = "completed"class DialogueContext(BaseModel):session_id: struser_id: strcurrent_state: DialogueStatecurrent_intent: Optional[str]slots: Dict[str, Any]history: List[Dict[str, Any]]created_at: datetimeupdated_at: datetimeclass DialogueSystem:def __init__(self,llm,tool_registry,memory_system):self.llm = llmself.tool_registry = tool_registryself.memory_system = memory_systemself.sessions: Dict[str, DialogueContext] = {}async def process_message(self,session_id: str,user_id: str,message: str) -> str:# 1. 获取或创建会话上下文context = self._get_or_create_context(session_id,user_id)try:# 2. 更新状态context.current_state = DialogueState.PROCESSING# 3. 理解用户意图intent = await self._understand_intent(message,context)# 4. 更新上下文context.current_intent = intent.namecontext.slots.update(intent.slots)# 5. 执行对应的处理流程response = await self._handle_intent(intent,context)# 6. 记录对话历史self._update_history(context,message,response)return responseexcept Exception as e:context.current_state = DialogueState.ERRORreturn f"抱歉，处理您的请求时出现错误：{str(e)}"finally:# 保存上下文self._save_context(context)async def _understand_intent(self,message: str,context: DialogueContext) -> Intent:# 结合上下文理解用户意图response = await self.llm.understand_intent(message=message,history=context.history[-5:], # 最近5轮对话current_intent=context.current_intent,slots=context.slots)return Intent(name=response.intent,confidence=response.confidence,slots=response.slots)async def _handle_intent(self,intent: Intent,context: DialogueContext) -> str:# 检查是否有未填充的必要槽位missing_slots = self._get_missing_slots(intent)if missing_slots:# 返回槽位询问context.current_state = DialogueState.WAITING_INPUTreturn self._generate_slot_question(missing_slots[0])# 所有槽位都已填充，执行操作result = await self._execute_intent(intent,context)context.current_state = DialogueState.COMPLETEDreturn resultdef _get_or_create_context(self,session_id: str,user_id: str) -> DialogueContext:if session_id in self.sessions:return self.sessions[session_id]# 创建新会话context = DialogueContext(session_id=session_id,user_id=user_id,current_state=DialogueState.IDLE,current_intent=None,slots={},history=[],created_at=datetime.now(),updated_at=datetime.now())self.sessions[session_id] = contextreturn context使用示例：
# 初始化对话系统
dialogue = DialogueSystem(llm=ChatGPT(),tool_registry=tool_registry,memory_system=memory_system
)# 处理用户消息
async def chat():responses = []# 第一轮：查询订单response = await dialogue.process_message(session_id="123",user_id="user_1",message="帮我查一下订单状态")responses.append(response)# 输出：好的，请提供订单号。# 第二轮：提供订单号response = await dialogue.process_message(session_id="123",user_id="user_1",message="A12345")responses.append(response)# 输出：您的订单 A12345 正在配送中，预计明天送达。# 第三轮：追问细节response = await dialogue.process_message(session_id="123",user_id="user_1",message="具体什么时候到？")responses.append(response)# 输出：根据物流信息，预计明天上午10:00-12:00送达。return responses# 运行对话
responses = await chat()
for r in responses:print(r)关键实现细节
1. 意图理解
class IntentRecognizer:def __init__(self, llm):self.llm = llmasync def recognize(self,message: str,context: Dict[str, Any]) -> Intent:# 1. 准备提示词prompt = self._prepare_prompt(message,context)# 2. 调用 LLMresponse = await self.llm.generate(prompt)# 3. 解析结果intent = self._parse_response(response)# 4. 验证意图self._validate_intent(intent)return intentdef _prepare_prompt(self,message: str,context: Dict[str, Any]) -> str:return f"""请分析以下对话内容，识别用户意图：历史对话：{self._format_history(context.get('history', []))}当前状态：- 意图：{context.get('current_intent')}- 已知信息：{json.dumps(context.get('slots', {}), indent=2)}用户消息：{message}请返回：1. 意图名称2. 置信度3. 识别出的槽位信息"""2. 状态管理
class StateManager:def __init__(self):self.state_handlers = {DialogueState.IDLE: self._handle_idle,DialogueState.WAITING_INPUT: self._handle_waiting,DialogueState.PROCESSING: self._handle_processing,DialogueState.ERROR: self._handle_error,DialogueState.COMPLETED: self._handle_completed}async def handle_state(self,context: DialogueContext,message: str) -> str:# 获取当前状态的处理器handler = self.state_handlers.get(context.current_state)if not handler:raise ValueError(f"未知状态：{context.current_state}")# 执行状态处理return await handler(context, message)async def _handle_waiting(self,context: DialogueContext,message: str) -> str:# 检查是否填充了等待的槽位slot_name = context.waiting_for_slotif self._is_valid_slot_value(slot_name,message):# 更新槽位context.slots[slot_name] = message# 继续处理return await self._continue_processing(context)else:# 重新询问return f"抱歉，这似乎不是有效的{slot_name}，请重新输入。"3. 上下文管理
class ContextManager:def __init__(self, memory_system):self.memory = memory_systemself.max_history = 10async def update_context(self,context: DialogueContext,message: str,response: str):# 1. 更新对话历史context.history.append({"role": "user","content": message,"timestamp": datetime.now()})context.history.append({"role": "assistant","content": response,"timestamp": datetime.now()})# 2. 限制历史长度if len(context.history) > self.max_history * 2:# 保存旧对话到长期记忆old_messages = context.history[:-self.max_history * 2]await self._save_to_memory(context.session_id,old_messages)# 保留最近的对话context.history = context.history[-self.max_history * 2:]# 3. 更新时间戳context.updated_at = datetime.now()async def _save_to_memory(self,session_id: str,messages: List[Dict]):# 将对话保存到长期记忆await self.memory.remember(content=self._format_messages(messages),metadata={"type": "dialogue","session_id": session_id,"timestamp": datetime.now()})优化技巧
在实践中，我总结了一些提升用户体验的技巧：
1. 主动确认
class ConfirmationManager:def __init__(self, threshold: float = 0.8):self.threshold = thresholddef need_confirm(self,intent: Intent,context: DialogueContext) -> bool:# 检查是否需要确认if intent.confidence < self.threshold:return Trueif self._is_critical_operation(intent):return Truereturn Falsedef generate_confirmation(self,intent: Intent,context: DialogueContext) -> str:return f"""请确认您是否要{intent.description}？- 操作：{intent.name}- 参数：{json.dumps(intent.slots, indent=2)}回复"是"或"否"。"""2. 错误恢复
class ErrorRecovery:async def recover(self,error: Exception,context: DialogueContext) -> str:# 分析错误analysis = await self._analyze_error(error)if analysis.can_retry:# 自动重试return await self._retry_operation(context)elif analysis.need_clarification:# 请求用户澄清return self._generate_clarification_question(analysis)else:# 友好的错误提示return self._generate_error_message(analysis)3. 上下文压缩
class ContextCompressor:def compress_history(self,history: List[Dict],max_tokens: int) -> List[Dict]:# 1. 计算当前token数current_tokens = self._count_tokens(history)if current_tokens <= max_tokens:return history# 2. 提取关键信息key_messages = self._extract_key_messages(history)# 3. 压缩对话compressed = self._compress_messages(key_messages,max_tokens)return compresseddef _extract_key_messages(self,history: List[Dict]) -> List[Dict]:# 提取重要的对话轮次key_turns = []for i, msg in enumerate(history):if self._is_key_message(msg, history, i):key_turns.append(msg)return key_turns实践心得
在实现和优化对话系统的过程中，我总结了几点经验：
以用户体验为中心
- 理解用户真实意图
- 保持对话的连贯性
- 给出清晰的反馈
要有容错机制
- 优雅处理异常
- 支持意图澄清
- 允许用户更正
注意性能优化
- 合理管理上下文
- 及时清理无用信息
- 异步处理耗时操作
写在最后
一个好的对话系统应该像一个专业的客服，既要理解用户需求，又要高效地解决问题。它不仅要"能听懂"，还要"会说话"。
在下一篇文章中，我会讲解如何实现 AI Agent 的安全机制。如果你对对话系统的设计有什么想法，欢迎在评论区交流。
