多轮对话中的意图漂移检测：Agent 走向跑偏后的自动纠偏

多轮对话中的意图漂移检测Agent 走向跑偏后的自动纠偏一、用户还在问东南亚旅游攻略Agent 已经聊到了月球移民这是多轮对话 Agent 最常见也最令人无语的失败模式。用户前三轮在问泰国签证的流程第四轮随口说了句我也想体验下失重感Agent 立刻切到太空旅游的上下文里开始介绍 SpaceX 的发射计划。等用户第五轮问那曼谷呢Agent 已经没法把自己的思路拉回来了——上下文窗口被无效信息填满了大半。意图漂移的本质是Agent 没有在持续追踪用户的顶层任务目标。每一步只关注当前轮次的字面含义丢失了多轮对话积累的任务上下文。当用户的一个非严肃提及比如开玩笑、打岔、类比触发了 Agent 的语义关联时Agent 可能偏离主线并一路加速跑偏。解决思路的核心是在每一轮对话后做一个简单判断这一轮的用户消息是否导致了当前任务目标的偏离偏离的严重程度多大是否需要显式纠正二、底层机制与原理剖析意图漂移检测的核心架构是一个双轨并行系统主任务追踪轨道负责维护当前对话的顶层任务和子任务状态漂移检测轨道负责判断每一轮新消息与当前任务轨道的语义距离。核心组件分三个任务存储器TaskState维护当前对话的顶层任务描述。不是一个固定话题词而是一个结构化描述{topic: 泰国旅行规划, sub_topic: 签证办理, stage: 材料准备}。每一轮 LLM 生成后用另一个轻量级调用提取并更新这个状态。语义距离计算器将当前用户消息与任务存储器的状态描述分别编码为向量计算余弦距离。距离近 在聊同一件事距离远 可能跑偏了。纠偏注入器当检测到漂移时在不打断用户的前提下在下一轮的 System Prompt 中注入友好的纠偏上下文当前的核心任务是泰国签证办理。用户可能顺带提到了其他话题但请在回复中优先回到主线。三、生产级代码实现 多轮对话意图漂移检测器 核心设计双轨并行——主任务追踪 漂移检测 检测到漂移后通过 Prompt 注入而非强制打断来纠偏 from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple from enum import Enum import numpy as np import hashlib import time class DriftLevel(Enum): 漂移严重级别 NONE 0 # 无漂移 MILD 1 # 轻微偏移可能只是举例或打岔 MODERATE 2 # 中度偏移可能是子话题切换 SEVERE 3 # 严重偏移完全偏离主任务 dataclass class TaskState: 任务状态存储器 设计决策状态用结构化字段而非自由文本 因为结构化字段的语义距离计算更稳定、可比 topic: str # 主话题 sub_topic: str # 子话题 stage: str # 当前阶段如信息收集决策执行 confidence: float # 置信度 0-1 last_update: int # 时间戳秒 def to_description(self) - str: 生成用于语义计算的描述文本 return f当前任务{self.topic}子话题{self.sub_topic}阶段{self.stage} def fingerprint(self) - str: 生成任务状态的简短指纹用于日志追踪 raw f{self.topic}|{self.sub_topic}|{self.stage} return hashlib.md5(raw.encode()).hexdigest()[:8] dataclass class DriftDetectionResult: 漂移检测结果 drift_level: DriftLevel drift_score: float # 漂移分数 0-1值越大漂移越严重 current_task: TaskState suggested_action: str # 建议的纠偏动作 reason: str # 判定原因 class IntentDriftDetector: 意图漂移检测器 检测多轮对话中用户意图是否偏离了主线任务。 核心算法语义距离 规则增强。 def __init__( self, embed_fn, # 嵌入函数: str - np.ndarray drift_threshold_mild: float 0.3, drift_threshold_moderate: float 0.5, drift_threshold_severe: float 0.7, ): self.embed_fn embed_fn self.drift_threshold_mild drift_threshold_mild self.drift_threshold_moderate drift_threshold_moderate self.drift_threshold_severe drift_threshold_severe # 漂移历史记录用于判断是否是持续的漂移趋势 self._drift_history: List[float] [] self._max_history 5 def detect( self, user_message: str, task_state: TaskState, conversation_history: List[Dict], ) - DriftDetectionResult: 检测当前消息是否漂移 参数 - user_message: 当前轮用户消息 - task_state: 当前追踪到的主任务状态 - conversation_history: 最近 N 轮对话历史 格式: [{role: user/assistant, content: ...}] # Step 1: 计算语义距离 distance self._compute_semantic_distance(user_message, task_state) # Step 2: 计算上下文连贯性用户消息与前一轮助手回复的关联度 context_coherence self._compute_context_coherence( user_message, conversation_history ) # Step 3: 综合评分 # 设计决策语义距离权重 0.7, 上下文连贯性权重 0.3 # 语义距离是主信号上下文连贯性用于辅助判断防止误判 drift_score distance * 0.7 (1 - context_coherence) * 0.3 drift_score max(0.0, min(1.0, drift_score)) # Step 4: 记录历史 self._drift_history.append(drift_score) if len(self._drift_history) self._max_history: self._drift_history.pop(0) # Step 5: 连续漂移趋势增强 if len(self._drift_history) 3: recent_avg sum(self._drift_history[-3:]) / 3 if recent_avg self.drift_threshold_moderate: # 连续三轮高漂移加重判定 drift_score min(1.0, drift_score * 1.2) # Step 6: 判定漂移级别并生成建议 drift_level, action, reason self._classify(drift_score, distance, context_coherence) return DriftDetectionResult( drift_leveldrift_level, drift_scoreround(drift_score, 4), current_tasktask_state, suggested_actionaction, reasonreason, ) def _compute_semantic_distance(self, message: str, task: TaskState) - float: 计算用户消息与当前任务状态的语义距离 try: msg_embed self.embed_fn(message) task_embed self.embed_fn(task.to_description()) # 余弦距离 1 - 余弦相似度范围 [0, 1] # 处理零向量边缘情况 msg_norm np.linalg.norm(msg_embed) task_norm np.linalg.norm(task_embed) if msg_norm 1e-8 or task_norm 1e-8: return 0.5 # 降级返回模糊值 cosine_sim np.dot(msg_embed, task_embed) / (msg_norm * task_norm) # 裁剪到 [0, 1]余弦相似度为负时意味语义相反 distance max(0.0, min(1.0, 1.0 - cosine_sim)) return distance except Exception: # 嵌入计算异常保守判断为轻微漂移 return 0.35 def _compute_context_coherence( self, user_message: str, history: List[Dict], ) - float: 计算用户消息与上文对话的连贯性 如果用户消息与最近一条助手回复高度相关 说明用户大概率在继续上一个话题不是漂移。 if not history: return 0.5 # 无上下文中性值 # 取最近一轮的助手回复 last_assistant_msgs [ h[content] for h in history[-4:] # 最近4条 if h.get(role) assistant ] if not last_assistant_msgs: return 0.5 # 计算用户消息与最近助手回复的相似度 try: msg_embed self.embed_fn(user_message) ctx_embed self.embed_fn(last_assistant_msgs[-1]) cosine_sim np.dot(msg_embed, ctx_embed) / ( np.linalg.norm(msg_embed) * np.linalg.norm(ctx_embed) 1e-8 ) return max(0.0, min(1.0, cosine_sim)) except Exception: return 0.5 def _classify( self, drift_score: float, distance: float, coherence: float ) - Tuple[DriftLevel, str, str]: 根据分数判定漂移级别和纠偏建议 if drift_score self.drift_threshold_mild: return ( DriftLevel.NONE, 正常响应无需纠偏, f语义距离{distance:.2f}, 上下文连贯性{coherence:.2f}在正常范围内, ) elif drift_score self.drift_threshold_moderate: return ( DriftLevel.MILD, 在 Prompt 中注入轻量提醒引导回到主线, f语义距离{distance:.2f}可能是短暂打岔或举例, ) elif drift_score self.drift_threshold_severe: return ( DriftLevel.MODERATE, 在 Prompt 中明确提醒当前任务优先回到主线, f语义距离{distance:.2f}大概率是子话题切换, ) else: return ( DriftLevel.SEVERE, 主动向用户确认是否切换任务不隐含纠偏, f语义距离{distance:.2f}几乎确定已偏离主任务, ) def build_correction_prompt(self, result: DriftDetectionResult) - Optional[str]: 根据检测结果生成纠偏 Prompt 注入文本 不返回完整 Prompt只返回需要注入的纠偏段落。 设计决策纠偏文本应简洁、友好、不打断对话流畅度。 if result.drift_level DriftLevel.NONE: return None if result.drift_level DriftLevel.MILD: return ( f【系统提示】当前核心任务{result.current_task.to_description()}。 f用户可能短暂提及其他话题请在回复末尾自然引导回主线。 ) if result.drift_level DriftLevel.MODERATE: return ( f【系统提示】核心任务{result.current_task.to_description()}。 f检测到用户意图可能偏移。请在回复中优先围绕核心任务展开 f如有必要可礼貌询问用户是否仍需要之前的帮助。 ) # SEVERE return ( f【系统提示】原任务{result.current_task.to_description()}。 f检测到用户话题发生了显著变化。下一步回复中必须主动确认 f用户是希望切换任务还是回到原任务继续 ) # 辅助简易嵌入函数实际使用应替换为 OpenAI / Cohere / 本地模型 class SimpleEmbedder: 占位嵌入器——生产环境替换为真实的 Embedding API def __init__(self, dim: int 128): self.dim dim # 简易词袋向量器仅作演示 def __call__(self, text: str) - np.ndarray: # 简易实现基于 TF-IDF 词袋的思路 # 生产环境替换为openai.Embedding.create(inputtext, modeltext-embedding-3-small) words set(text.lower().split()) vec np.zeros(self.dim) for w in words: idx hash(w) % self.dim vec[idx] 1.0 norm np.linalg.norm(vec) if norm 0: vec vec / norm return vec四、边界分析与架构权衡意图漂移检测的缺点语义距离计算本身是一个近似方法不是精确判断。用户可能确实在进行类比推理——比如聊泰国旅游时提到失重感是在类比潜水体验这不是真的漂移。这种情况下的误判会将有效的对话强行拉回反而打断用户的自然表达。另一个问题是嵌入向量的计算延迟。每一轮对话都要额外调用 embedding API增加了 50-200ms 的延迟。如果对话吞吐量高embedding API 的调用量也是成本。适用边界最适合多轮对话任务明确且跨度长的场景——如客服问答、旅行规划、代码辅助。也适合有明确顶层任务目标的 Agent任务越结构化漂移检测越准确。禁用场景不适合开放式闲聊 Agent——闲聊本身就没有固定任务目标漂移检测没有意义。也不适合单轮问答场景多轮上下文都没积累起来。对延迟极度敏感的场景如实时语音对话embedding 的额外延迟可能不可接受。五、总结意图漂移不是 AI 的智商问题是任务追踪机制缺失导致的必然现象。用双轨并行架构主任务轨持续追踪任务状态漂移检测轨计算语义距离。发现偏移时通过 Prompt 注入而非强制打断来做柔性的纠偏。关键设计在于分级处理——轻微的偏移只需轻提醒严重的偏移才需主动确认。保持对话流畅度的同时守住任务主线。