AI Agent的多轮对话：提升用户体验的关键技巧

在前面的文章中，我们讨论了 AI Agent 的各个核心系统。今天，我想聊聊如何实现一个好用的多轮对话系统。说实话，这个话题我琢磨了很久，因为它直接影响到用户体验。

从一个槽点说起

还记得我最开始做对话系统时的一个典型场景：

用户：帮我查一下订单状态
助手：好的，请提供订单号。
用户：A12345
助手：抱歉，我需要您提供订单号才能查询。
用户：...我刚才不是说了吗？

这种对话体验显然很糟糕。问题出在哪里？主要是 AI Agent 没有正确理解和维护对话上下文。后来我总结了几个关键点：

1. 要理解用户意图
2. 要记住上下文信息
3. 要有状态管理机制
4. 要能处理各种异常情况

对话系统的设计

经过多次迭代，我设计了一个相对完善的对话系统：

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

实践心得

在实现和优化对话系统的过程中，我总结了几点经验：

1. 以用户体验为中心

   • 理解用户真实意图
   • 保持对话的连贯性
   • 给出清晰的反馈

2. 要有容错机制

   • 优雅处理异常
   • 支持意图澄清
   • 允许用户更正

3. 注意性能优化

   • 合理管理上下文
   • 及时清理无用信息
   • 异步处理耗时操作

写在最后

一个好的对话系统应该像一个专业的客服，既要理解用户需求，又要高效地解决问题。它不仅要"能听懂"，还要"会说话"。

在下一篇文章中，我会讲解如何实现 AI Agent 的安全机制。如果你对对话系统的设计有什么想法，欢迎在评论区交流。