_{🧠 图解记忆：重复状态、预算耗尽、上下文膨胀和证据缺失都应转成可观测异常信号；点击图片可查看原图。}

**异常检测架构：**

展开 Python 代码示例（71 行）

class AgentAnomalyDetector:
    def __init__(self):
        self.consecutive_identical = 0
        self.max_identical_steps = 3
        self.max_total_steps = 20
        self.history_hashes = []  # 存储历史状态 hash
    
    def detect_loop(self, step_result: str) -> bool:
        """检测重复步骤"""
        current_hash = hash(step_result)
        
        if current_hash in self.history_hashes:
            self.consecutive_identical += 1
            if self.consecutive_identical >= self.max_identical_steps:
                return True
        else:
            self.consecutive_identical = 0
        
        self.history_hashes.append(current_hash)
        return False
    
    def detect_context_bloat(self, messages: list) -> bool:
        """检测上下文膨胀"""
        total_tokens = sum(count_tokens(m) for m in messages)
        # 超过上下文窗口 80% 则告警
        if total_tokens > CONTEXT_LIMIT * 0.8:
            return True
        return False
    
    def detect_hallucination_risk(self, response: str, context: list) -> float:
        """用 Entailment 模型检测幻觉风险"""
        # 检测 response 中的事实陈述是否被 context 支持
        facts = extract_factual_statements(response)
        supported = 0
        
        for fact in facts:
            # 用 NLI 模型判断 entailment
            if nli_model.verify(fact, context) == "entailment":
                supported += 1
        
        return 1.0 - (supported / len(facts)) if facts else 0.0

# 生产集成示例
@track_cost(model="qwen3.5-plus", ...)
async def agent_run(query: str):
    detector = AgentAnomalyDetector()
    messages = []
    step_count = 0
    
    while step_count < MAX_STEPS:
        step_result = await agent.step(query, messages)
        
        # 循环检测
        if detector.detect_loop(step_result):
            raise LoopDetectedError("Detected repeated steps")
        
        # 上下文膨胀检测
        if detector.detect_context_bloat(messages):
            messages = compress_with_llmlingua(messages)
        
        # 幻觉风险检测
        hallucination_score = detector.detect_hallucination_risk(
            step_result, messages
        )
        if hallucination_score > 0.5:
            logger.warning(f"High hallucination risk: {hallucination_score}")
        
        messages.append(step_result)
        step_count += 1
    
    return final_response(messages)

📚 参考：Langfuse：异常检测（循环/幻觉监控）

面试话术：

"Agent 异常检测可覆盖重复状态/动作、预算耗尽、上下文增长、工具错误和证据不支持。阈值不能照搬固定数字，应使用历史分布、业务 SLO 和误报成本校准；状态 hash 也要区分正常重试与死循环。告警应能关联 trace、版本和 runbook。"

---