SGJki.
← blog AI Agent · 2026-04-09

探测窗口算法与状态机在AI面试Agent中的应用

探索扩展状态机、探测窗口与滑动窗口三种算法在AI面试Agent中的应用，解决LLM输出不确定性、格式错误、幻觉回答、截断检测等核心挑战

12 min read
#LLM#AI#AI-Interview

LLM输出限制 - 探测窗口算法与状态机在AI面试Agent中的应用

引言

在构建 AI 面试 Agent 时，一个核心挑战是如何确保 LLM 的输出质量。LLM 输出的不确定性（幻觉、格式错误、截断等）会直接影响面试体验。本文探讨两种核心方法：扩展状态机（Extended State Machine） 和 探测窗口算法（Detection Window），以及它们在 AI-Interview 项目中的实际应用。

核心问题：LLM 输出为何需要限制

LLM 输出失败可分为四类：

类型描述示例
格式类JSON 解析失败、schema 不匹配{"name": "error} 缺少引号
内容类幻觉回答、偏离问题、敏感词回答与问题无关
状态类空输出、截断输出、超时streaming 中断
语义类逻辑矛盾、循环回答、自我矛盾前后回答冲突

传统的做法是重试机制：输出失败就重新调用 LLM。但这种方法效率低下，尤其在流式输出场景下，用户已经看到了部分内容。

方法一：扩展状态机

核心思想

状态机将面试流程建模为离散的、有穷的、互斥的状态。每个状态转换都有明确的触发条件和动作。

stateDiagram-v2
    [*] --> WAITING
    WAITING --> EVALUATING: submit_answer
    EVALUATING --> CORRECT: deviation_score
    EVALUATING --> GUIDANCE: deviation_score
    EVALUATING --> CORRECTION: deviation_score

优点

  • 可预测性强：状态轨迹清晰，每个转换都可追踪
  • 可视化友好：便于调试和问题定位
  • 流程可控：异常恢复逻辑清晰

缺点

  • 状态爆炸：随着业务复杂度的增加，状态数量指数级增长
  • 不适合细粒度检测：只能在状态转换点检测，无法在流式输出中实时检测
  • 层次不清：当检测逻辑与流程逻辑混在一起时，代码难以维护

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 字段追踪面试阶段，每个阶段有明确的进入条件和退出条件。

方法二：探测窗口算法

核心思想

探测窗口是数据流 + 管道过滤的架构，将输出检测组织为多层独立的”窗口”，每个窗口专注特定类型的检测。

graph TD
    S["LLM Output Stream"]
    W1["Window 1: Format Detection (快速失败)<br/>JSON 语法 / Schema 结构 / 是否为空截断"]
    W2["Window 2: Safety Detection (安全检测)<br/>敏感词过滤 / 政治敏感内容 / 恶意代码检测"]
    W3["Window 3: Semantic Validation (语义合法性)<br/>LLM 驱动语义检测 / 幻觉判断 / 逻辑一致性"]
    W4["Window 4: Quality Scoring (质量评分)<br/>回答完整度 / 与问题相关性 / 偏离度评分"]
    S --> W1 --> W2 --> W3 --> W4

优点

  • 关注点分离：每层窗口独立职责，易于扩展
  • 增量检测：可以在流式输出中边生成边检测，快速失败
  • 适合多阶段验证：从格式到语义，分层过滤

缺点

  • 状态模糊：数据流没有明确的状态边界
  • 调试复杂：问题可能在多个窗口间传递，难以定位
  • 延迟累加：每个窗口都有处理延迟

架构对比：老版 vs 新版

老版 InterviewService 架构

单 Agent 流程，状态机控制整体流程，探测窗口验证输出：

graph TD
    subgraph IS["InterviewService"]
        F["Format Window"]
        S["Schema Window"]
        SAF["Safety Window"]
        SM["State Machine (Phase Control)"]
        LLM["LLM Service"]
    end
    F --> S --> SAF
    F --> SM
    S --> SM
    SAF --> SM
    SM --> LLM

新版 Orchestrator + ReviewAgent 架构

多 Agent 协作，Orchestrator 负责路由和流程控制（状态机），ReviewAgent 通过 LLM 驱动检测：

graph TD
    subgraph ORCH["Orchestrator (LangGraph)"]
        Q["Question Agent"]
        E["Evaluate Agent"]
        R["Review Agent"]
        LD["LLM-driven Detection<br/>(替代规则驱动的探测窗口)"]
        FB["Feedback Loop<br/>(无效输出→重试/降级)"]
    end
    Q --> E --> R
    E --> LD
    R --> LD
    LD --> FB
    Q --> FB

核心区别

维度老版架构新版架构
架构模式单 Agent + 规则引擎多 Agent 协作 (Orchestrator)
检测方式规则驱动 (Regex/Schema)LLM 驱动 (语义理解)
流程控制状态机硬编码Graph 路由 + 条件边
扩展方式增加规则增加 Agent 节点
反馈机制直接重试降级 + ReviewAgent 判断

核心结论：混合架构

两者结合是最佳实践：

场景推荐方案原因
面试流程阶段控制State Machine阶段明确、转换可控
LLM 输出质量检测Detection Window分层过滤、增量检测
异常恢复流程State Machine状态明确、动作确定
流式输出边生成边检测Detection Window快速失败、及时截断
graph TD
    subgraph HYB["Hybrid Architecture"]
        O["Orchestrator (State Machine)<br/>流程阶段控制 / 路由决策 / 异常恢复"]
        RG["ReviewAgent (Detection Window)<br/>Format/Schema 检测 / Safety 检测 / Semantic 验证"]
        FL["Feedback Loop<br/>无效输出 → 重试/降级<br/>降级策略：简化 prompt、减少要求"]
    end
    O --> RG --> FL

探测窗口算法设计要点

三阶段处理策略

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)

可观测性设计

每个无效输出都应记录，用于后续分析和优化：

@dataclass
class 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）是一种动态数据处理范式，与静态的探测窗口不同，它在时间/字数维度上保持一个”窗口”，随数据流入不断滑动，适用于需要时序分析和趋势检测的场景。

graph LR
    subgraph DW["Detection Window (静态/管道式)"]
        In["Input"] --> A["Window A"] --> B["Window B"] --> C["Window C"] --> Out["Output"]
    end
    Note1["特点：每个窗口接收完整输入，逐层过滤，无状态保留"]
    subgraph SW["Sliding Window (滑动/增量式)"]
        D1["D1"] --> D2["D2"] --> D3["D3"] --> D4["D4"] --> D5["D5"] --> D6["D6"] --> D7["D7"]
    end
    Note2["Window Size = 4 (当前窗口)<br/>每滑动一次：淘汰最旧 1个，加入最新 1个 = 增量更新<br/>→ 时间/字数轴"]

滑动窗口的三种类型

类型窗口大小应用场景
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 WindowSliding Window
数据保留仅保留当前处理的数据保留窗口内历史数据
计算方式单次计算增量更新
适用场景格式化检测、schema 验证时序分析、趋势检测
状态管理无状态有状态（滑动历史）
延迟低（无需维护历史）中等（需维护窗口）

综合架构：三种方法协同

graph TD
    S["流式 LLM 输出检测架构<br/>LLM Output Stream"]
    L1["Sliding Window Layer 1: 格式/字数监控<br/>实时检测截断、空输出 / 字数异常报警"]
    L2["Sliding Window Layer 2: 语义一致性监控<br/>回答序列矛盾检测 / 主题漂移检测"]
    DW["Detection Window Layer: 规则/Schema 验证<br/>JSON 格式检查 / 敏感词过滤"]
    SM["State Machine (反馈环控制)"]
    S --> L1 --> L2 --> DW --> SM

实战经验总结

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 输出的不确定性，同时保持系统的可维护性和可扩展性。

Sources

No external sources for this entry.

Related

2026-06-15 · blog

Hermes Agent 核心功能与 Skill 自进化机制

剖析 MemTensor 的 Hermes Agent：一个 Python agent 外壳经 JSON-RPC 接入 MemOS Reflect2Evolve V7 记忆引擎，靠 crystallize→verify→lifecycle 闭环实现 skill 自进化，用 η 的…AI Agent · 10 min read 2026-04-09 · blog

AI Application Engineer interview concerned Question

基于简历分析，为 AI应用工程师岗位设计面试指南：Question concerned of interview-designer Skill, AI Agent · 8 min read 2026-04-09 · blog

prompt-injection

深入探讨 Prompt Injection 的四层防御体系，涵盖威胁模型、输入层防御、结构化约束、工具权限控制与架构防御，帮助开发者构建更安全的 LLM 应用。AI Agent · 4 min read