LLM输出限制 - 探测窗口算法与状态机在AI面试Agent中的应用
引言
在构建 AI 面试 Agent 时,一个核心挑战是如何确保 LLM 的输出质量。LLM 输出的不确定性(幻觉、格式错误、截断等)会直接影响面试体验。本文探讨两种核心方法:扩展状态机(Extended State Machine) 和 探测窗口算法(Detection Window),以及它们在 AI-Interview 项目中的实际应用。
核心问题:LLM 输出为何需要限制
LLM 输出失败可分为四类:
| 类型 | 描述 | 示例 |
|---|---|---|
| 格式类 | JSON 解析失败、schema 不匹配 | {"name": "error} 缺少引号 |
| 内容类 | 幻觉回答、偏离问题、敏感词 | 回答与问题无关 |
| 状态类 | 空输出、截断输出、超时 | streaming 中断 |
| 语义类 | 逻辑矛盾、循环回答、自我矛盾 | 前后回答冲突 |
传统的做法是重试机制:输出失败就重新调用 LLM。但这种方法效率低下,尤其在流式输出场景下,用户已经看到了部分内容。
方法一:扩展状态机
核心思想
状态机将面试流程建模为离散的、有穷的、互斥的状态。每个状态转换都有明确的触发条件和动作。
┌─────────┐ submit_answer ┌─────────┐│ WAITING │ ──────────────────> │ EVALUATING │└─────────┘ └─────────┘ │ deviation_score ┌─────────────────┼─────────────────┐ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ CORRECT │ │ GUIDANCE │ │ CORRECTION│ └──────────┘ └──────────┘ └──────────┘优点
- 可预测性强:状态轨迹清晰,每个转换都可追踪
- 可视化友好:便于调试和问题定位
- 流程可控:异常恢复逻辑清晰
缺点
- 状态爆炸:随着业务复杂度的增加,状态数量指数级增长
- 不适合细粒度检测:只能在状态转换点检测,无法在流式输出中实时检测
- 层次不清:当检测逻辑与流程逻辑混在一起时,代码难以维护
InterviewState 示例
@dataclass(frozen=True)class InterviewState: session_id: str resume_id: str
# 当前面试进度 current_series: int = 1 current_question: Optional[Question] = None current_question_id: Optional[str] = None
# 追问链追踪 followup_depth: int = 0 max_followup_depth: int = 3 followup_chain: list[str] = field(default_factory=list)
# 回答记录 answers: dict[str, Answer] = field(default_factory=dict) feedbacks: dict[str, Feedback] = field(default_factory=dict)
# 状态 error_count: int = 0 phase: Literal["init", "warmup", "initial", "followup", "final_feedback"] = "init"状态机通过 phase 字段追踪面试阶段,每个阶段有明确的进入条件和退出条件。
方法二:探测窗口算法
核心思想
探测窗口是数据流 + 管道过滤的架构,将输出检测组织为多层独立的”窗口”,每个窗口专注特定类型的检测。
┌──────────────────────────────────────────────────────────────┐│ LLM Output Stream │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Window 1: Format Detection (快速失败) ││ - JSON 语法检测 ││ - Schema 结构检测 ││ - 是否为空/截断 │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Window 2: Safety Detection (安全检测) ││ - 敏感词过滤 ││ - 政治敏感内容 ││ - 恶意代码检测 │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Window 3: Semantic Validation (语义合法性) ││ - LLM 驱动的语义检测 ││ - 幻觉判断 ││ - 逻辑一致性 │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Window 4: Quality Scoring (质量评分) ││ - 回答完整度 ││ - 与问题的相关性 ││ - 偏离度评分 │└──────────────────────────────────────────────────────────────┘优点
- 关注点分离:每层窗口独立职责,易于扩展
- 增量检测:可以在流式输出中边生成边检测,快速失败
- 适合多阶段验证:从格式到语义,分层过滤
缺点
- 状态模糊:数据流没有明确的状态边界
- 调试复杂:问题可能在多个窗口间传递,难以定位
- 延迟累加:每个窗口都有处理延迟
架构对比:老版 vs 新版
老版 InterviewService 架构
单 Agent 流程,状态机控制整体流程,探测窗口验证输出:
┌─────────────────────────────────────────────────────────────┐│ InterviewService │├─────────────────────────────────────────────────────────────┤│ ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ││ │ Format │───>│ Schema │───>│ Safety │ ││ │ Window │ │ Window │ │ Window │ ││ └──────────────┘ └──────────────┘ └──────────────┘ ││ │ │ │ ││ └──────────────────┼──────────────────┘ ││ ▼ ││ ┌──────────────────┐ ││ │ State Machine │ ││ │ (Phase Control) │ ││ └──────────────────┘ ││ │ ││ ▼ ││ ┌──────────────────┐ ││ │ LLM Service │ ││ └──────────────────┘ │└─────────────────────────────────────────────────────────────┘新版 Orchestrator + ReviewAgent 架构
多 Agent 协作,Orchestrator 负责路由和流程控制(状态机),ReviewAgent 通过 LLM 驱动检测:
┌─────────────────────────────────────────────────────────────┐│ Orchestrator (LangGraph) │├─────────────────────────────────────────────────────────────┤│ ││ ┌────────────┐ ┌────────────┐ ┌────────────┐ ││ │ Question │───>│ Evaluate │───>│ Review │ ││ │ Agent │ │ Agent │ │ Agent │ ││ └────────────┘ └────────────┘ └────────────┘ ││ │ │ │ ││ │ ▼ ▼ ││ │ ┌────────────────────────────┐ ││ │ │ LLM-driven Detection │ ││ │ │ (替代规则驱动的探测窗口) │ ││ │ └────────────────────────────┘ ││ │ │ ││ └────────────────────────┼─────────────────────────┘│ ▼ ││ ┌──────────────────────┐ ││ │ Feedback Loop │ ││ │ (无效输出→重试/降级) │ ││ └──────────────────────┘ │└─────────────────────────────────────────────────────────────┘核心区别
| 维度 | 老版架构 | 新版架构 |
|---|---|---|
| 架构模式 | 单 Agent + 规则引擎 | 多 Agent 协作 (Orchestrator) |
| 检测方式 | 规则驱动 (Regex/Schema) | LLM 驱动 (语义理解) |
| 流程控制 | 状态机硬编码 | Graph 路由 + 条件边 |
| 扩展方式 | 增加规则 | 增加 Agent 节点 |
| 反馈机制 | 直接重试 | 降级 + ReviewAgent 判断 |
核心结论:混合架构
两者结合是最佳实践:
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 面试流程阶段控制 | State Machine | 阶段明确、转换可控 |
| LLM 输出质量检测 | Detection Window | 分层过滤、增量检测 |
| 异常恢复流程 | State Machine | 状态明确、动作确定 |
| 流式输出边生成边检测 | Detection Window | 快速失败、及时截断 |
┌─────────────────────────────────────────────────────────────┐│ Hybrid Architecture │├─────────────────────────────────────────────────────────────┤│ ││ ┌─────────────────────────────────────────────────────┐ ││ │ Orchestrator (State Machine) │ ││ │ - 流程阶段控制 │ ││ │ - 路由决策 │ ││ │ - 异常恢复 │ ││ └─────────────────────────────────────────────────────┘ ││ │ ││ ▼ ││ ┌─────────────────────────────────────────────────────┐ ││ │ ReviewAgent (Detection Window) │ ││ │ - Format/Schema 检测 │ ││ │ - Safety 检测 │ ││ │ - Semantic 验证 │ ││ └─────────────────────────────────────────────────────┘ ││ │ ││ ▼ ││ ┌─────────────────────────────────────────────────────┐ ││ │ Feedback Loop │ ││ │ - 无效输出 → 重试/降级 │ ││ │ - 降级策略:简化 prompt、减少要求 │ ││ └─────────────────────────────────────────────────────┘ ││ │└─────────────────────────────────────────────────────────────┘探测窗口算法设计要点
三阶段处理策略
async def process_llm_output(streaming_output): # Stage 1: 快速失败检测 (同步,阻塞式) format_result = await check_format_window(streaming_output) if format_result.is_invalid: return await fast_fail(format_result.error)
# Stage 2: 语义合法性检测 (异步,可等待) semantic_result = await check_semantic_window(streaming_output) if semantic_result.is_invalid: return await handle_semantic_failure(semantic_result)
# Stage 3: 降级与恢复 if semantic_result.needs_human_review: await escalate_to_human(semantic_result)
return StreamingResult(status="valid", content=streaming_output)可观测性设计
每个无效输出都应记录,用于后续分析和优化:
@dataclassclass OutputValidationRecord: timestamp: datetime output_type: str # "question", "feedback", "evaluation" validation_stage: str # "format", "safety", "semantic" is_valid: bool error_message: Optional[str] deviation_score: Optional[float] retry_count: int方法三:滑动窗口算法
核心思想
滑动窗口(Sliding Window)是一种动态数据处理范式,与静态的探测窗口不同,它在时间/字数维度上保持一个”窗口”,随数据流入不断滑动,适用于需要时序分析和趋势检测的场景。
┌─────────────────────────────────────────────────────────────────┐│ Detection Window (静态/管道式) ││ ││ Input ──► [Window A] ──► [Window B] ──► [Window C] ──► Output ││ ││ 特点:每个窗口接收完整输入,逐层过滤,无状态保留 │└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐│ Sliding Window (滑动/增量式) ││ ││ ┌──┬──┬──┬──┬──┬──┬──┐ ││ │D1│D2│D3│D4│D5│D6│D7│ ───► 时间/字数轴 ││ └──┴──┴──┴──┴──┴──┴──┘ ││ └──────┐ ││ Window Size = 4 (当前窗口) ││ ││ 每滑动一次:淘汰最旧 1个,加入最新 1个 = 增量更新 │└─────────────────────────────────────────────────────────────────┘滑动窗口的三种类型
| 类型 | 窗口大小 | 应用场景 |
|---|---|---|
| Tumbling Window | 固定大小,不重叠 | 批量统计、离线分析 |
| Sliding Window | 固定大小,重叠滑动 | 实时检测、流式报警 |
| Session Window | 动态大小,活动触发 | 用户会话、事件序列 |
在 LLM 输出检测中的实际应用
场景 1: 流式输出的字数滑动窗口
class SlidingWindowDetector: """检测 LLM 输出是否在合理字数范围内"""
def __init__(self, min_words=10, max_words=500, slide_step=5): self.min_words = min_words self.max_words = max_words self.slide_step = slide_step self.word_counts = [] # 滑动窗口记录
def process_token(self, new_token: str) -> DetectionResult: self.word_counts.append(len(new_token.split()))
# 窗口超过最大大小时,移除最旧的 if len(self.word_counts) > self.max_words: self.word_counts.pop(0)
# 检查当前窗口均值是否异常 if len(self.word_counts) >= self.min_words: avg = sum(self.word_counts) / len(self.word_counts) if avg < 2: # 平均每 token 词数过低,可能是截断 return DetectionResult.invalid("output_truncated")
return DetectionResult.valid()场景 2: 时间滑动窗口检测幻觉
class HallucinationSlidingWindow: """基于时间窗口的幻觉检测"""
def __init__(self, time_window_seconds=30, max_new_entities=5): self.time_window = time_window_seconds self.max_new_entities = max_new_entities self.entity_timeline = [] # (timestamp, entity_name)
def process_output(self, output: str, timestamp: datetime): entities = self.extract_entities(output)
# 添加时间戳到时间线 for entity in entities: self.entity_timeline.append((timestamp, entity))
# 移除超过窗口期的记录 cutoff = timestamp - timedelta(seconds=self.time_window) self.entity_timeline = [ (ts, e) for ts, e in self.entity_timeline if ts > cutoff ]
# 检查窗口内新实体数量 new_entities = set(e for ts, e in self.entity_timeline if ts == timestamp) if len(new_entities) > self.max_new_entities: return DetectionResult.invalid( f"possible_hallucination: {len(new_entities)} new entities in {self.time_window}s" )
return DetectionResult.valid()场景 3: 语义一致性的滑动窗口
class SemanticConsistencySlidingWindow: """检测回答序列的语义一致性"""
def __init__(self, window_size=3, consistency_threshold=0.6): self.window_size = window_size self.threshold = consistency_threshold self.answer_history = []
def check_consistency(self, new_answer: str) -> ConsistencyResult: embedding = self.get_embedding(new_answer) self.answer_history.append(embedding)
# 维持固定窗口大小 if len(self.answer_history) > self.window_size: self.answer_history.pop(0)
# 计算窗口内相邻回答的相似度 if len(self.answer_history) >= 2: similarities = [] for i in range(len(self.answer_history) - 1): sim = cosine_similarity( self.answer_history[i], self.answer_history[i+1] ) similarities.append(sim)
avg_similarity = sum(similarities) / len(similarities)
# 相似度骤降可能是矛盾信号 if avg_similarity < self.threshold: return ConsistencyResult.inconsistent( f"similarity_drop: {avg_similarity:.2f} < {self.threshold}" )
return ConsistencyResult.consistent()与探测窗口的核心区别
| 维度 | Detection Window | Sliding Window |
|---|---|---|
| 数据保留 | 仅保留当前处理的数据 | 保留窗口内历史数据 |
| 计算方式 | 单次计算 | 增量更新 |
| 适用场景 | 格式化检测、schema 验证 | 时序分析、趋势检测 |
| 状态管理 | 无状态 | 有状态(滑动历史) |
| 延迟 | 低(无需维护历史) | 中等(需维护窗口) |
综合架构:三种方法协同
流式 LLM 输出检测架构:
┌──────────────────────────────────────────────────────────────┐│ LLM Output Stream │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Sliding Window Layer 1: 格式/字数监控 ││ - 实时检测截断、空输出 ││ - 字数异常报警 │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Sliding Window Layer 2: 语义一致性监控 ││ - 回答序列矛盾检测 ││ - 主题漂移检测 │└──────────────────────────────────────────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────┐│ Detection Window Layer: 规则/Schema 验证 ││ - JSON 格式检查 ││ - 敏感词过滤 │└──────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────┐ │ State Machine │ │ (反馈环控制) │ └──────────────────┘实战经验总结
1. 状态机适用场景
在 AI-Interview 项目中,InterviewState.phase 字段使用状态机模式:
init→warmup:初始化后进入热身阶段warmup→initial:热身结束,开始正式提问initial→followup:基于偏差分数决定是否追问- 任何阶段 →
final_feedback:面试结束
2. 探测窗口适用场景
ReviewAgent 的反馈生成采用探测窗口思想:
# 不同偏差分数触发不同类型的反馈if deviation_score < 0.3: feedback_type = FeedbackType.CORRECTION # 直接纠错elif deviation_score < 0.6: feedback_type = FeedbackType.GUIDANCE # 引导性追问else: feedback_type = FeedbackType.COMMENT # 正面点评3. 混合架构实践
OrchestratorAdapter 展示了如何结合两者:
async def submit_answer(self, user_answer: str, question_id: str) -> QAResponse: # 1. 状态机:更新状态 self.state.answers[question_id] = answer
# 2. Graph 调用(内部包含 ReviewAgent 的检测逻辑) result = await self.graph.ainvoke(self.state)
# 3. 基于检测结果决定下一步 if self.state.next_action == "question_agent": # 生成下一个问题 ...总结
| 维度 | 状态机 | 探测窗口 | 滑动窗口 |
|---|---|---|---|
| 核心抽象 | 状态 + 转换 | 数据流 + 管道过滤 | 时间维度 + 增量更新 |
| 检测时机 | 状态转换点 | 持续流入/流出 | 滑动过程中实时 |
| 适用检测 | 流程合规性 | 输出质量(一次性) | 输出质量(时序/趋势) |
| 失败恢复 | 明确的状态转移 | 多级降级策略 | 窗口重置 |
| 调试体验 | 轨迹清晰 | 需要可观测性工具 | 需要窗口状态监控 |
| 典型应用 | 阶段切换 | Format/Safety 检测 | 截断检测、幻觉检测 |
最佳实践是三层混合架构:
- 状态机:控制面试流程阶段(init → warmup → initial → followup → final_feedback)
- 滑动窗口:实时检测流式输出的时序异常(截断、幻觉、一致性)
- 探测窗口:对完整输出进行规则/语义验证(JSON 格式、敏感词、Schema)
这种架构在 AI-Interview 项目中经过验证,能够有效处理 LLM 输出的不确定性,同时保持系统的可维护性和可扩展性。