自主Agent自愈：认知闭环与三事故解剖

graph TB
    subgraph "第五层：熔断降级"
        F1[全局熔断器]
        F2[降级策略引擎]
        F3[人类介入网关]
    end

    subgraph "第四层：检查点回滚"
        E1[状态快照]
        E2[变更日志]
        E3[回滚执行器]
    end

    subgraph "第三层：反思修正"
        D1[输出自评估]
        D2[错误归因]
        D3[策略修正]
    end

    subgraph "第二层：上下文健康"
        C1[压缩质量监控]
        C2[关键信息锚定]
        C3[窗口利用率]
    end

    subgraph "第一层：循环健康"
        A1[心跳检测]
        A2[停滞检测]
        A3[重复行为检测]
    end

    A1 --> C1
    C1 --> D1
    D1 --> E1
    E1 --> F1

    style F1 fill:#ff6b6b,color:#fff
    style E1 fill:#ffa502,color:#fff
    style D1 fill:#2ed573,color:#fff
    style C1 fill:#1e90ff,color:#fff
    style A1 fill:#a29bfe,color:#fff

# 循环健康检测器核心逻辑
class LoopHealthMonitor:
    def __init__(self, max_stall_rounds=4, max_repeat_actions=3):
        self.action_history = []
        self.stall_counter = 0
        self.max_stall = max_stall_rounds
        self.max_repeat = max_repeat_actions

    def check(self, current_action) -> HealthReport:
        # 重复行为检测
        recent = self.action_history[-self.max_repeat:]
        if len(recent) >= self.max_repeat:
            if all(a.signature == current_action.signature for a in recent):
                return HealthReport(
                    status="REPEAT_LOOP",
                    severity="CRITICAL",
                    message=f"Agent连续{self.max_repeat}次执行相同动作"
                )

        # 停滞检测
        if not current_action.mutated_state:
            self.stall_counter += 1
        else:
            self.stall_counter = 0

        if self.stall_counter >= self.max_stall:
            return HealthReport(
                status="STALLED",
                severity="HIGH",
                message=f"Agent已{self.stall_counter}轮未产生实质性进展"
            )

        self.action_history.append(current_action)
        return HealthReport(status="HEALTHY", severity="NONE")

# 上下文锚定机制
class ContextAnchor:
    def __init__(self, critical_entities: list[str]):
        self.anchors = {e: {"last_seen": 0, "importance": 1.0} for e in critical_entities}

    def pre_compaction_inject(self, messages: list) -> list:
        """在压缩前强制注入锚定信息"""
        anchor_block = "## 任务关键约束（不可省略）\n"
        for entity, meta in self.anchors.items():
            if meta["last_seen"] > 5:  # 5轮未出现
                anchor_block += f"- {entity}\n"

        # 注入到系统提示之后
        messages.insert(1, {"role": "system", "content": anchor_block})
        return messages

# Reflexion自愈循环
class ReflexionAgent:
    def __init__(self, llm, max_reflections=3):
        self.llm = llm
        self.reflection_memory = []
        self.max_reflections = max_reflections

    async def execute_with_self_healing(self, task):
        for attempt in range(self.max_reflections):
            # 构建包含历史反思的提示
            context = self._build_context(task)
            result = await self.llm.generate(context)

            # 自评估
            evaluation = await self._self_evaluate(result, task)
            if evaluation.passed:
                return result

            # 结构化反思并存储
            reflection = await self._reflect(result, evaluation, task)
            self.reflection_memory.append({
                "attempt": attempt,
                "action_taken": result.action,
                "failure_reason": evaluation.failure_reason,
                "reflection": reflection,
                "proposed_fix": reflection.suggested_strategy
            })

        # 超过最大反思次数，触发上层降级
        raise MaxReflectionExceeded(
            f"任务在{self.max_reflections}次反思后仍未通过，"
            f"最后一次失败原因：{evaluation.failure_reason}"
        )

# 检查点与回滚管理器
class CheckpointManager:
    def __init__(self):
        self.snapshots = []
        self.mutation_log = []

    def create_checkpoint(self, agent_state, description=""):
        checkpoint = {
            "id": len(self.snapshots),
            "timestamp": datetime.now(),
            "agent_state": deepcopy(agent_state),
            "description": description,
            "mutations_since_last": list(self.mutation_log)
        }
        self.snapshots.append(checkpoint)
        self.mutation_log = []
        return checkpoint["id"]

    def register_mutation(self, mutation: Mutation):
        """注册一个可变操作"""
        if mutation.irreversible:
            # 不可逆操作需要人类确认
            if not human_confirm(f"即将执行不可逆操作：{mutation.describe()}"):
                raise IrreversibleActionBlocked()
        self.mutation_log.append(mutation)

    def rollback_to(self, checkpoint_id):
        """回滚到指定检查点"""
        target = self.snapshots[checkpoint_id]
        # 反向执行所有变更
        for mutation in reversed(target["mutations_since_last"]):
            if mutation.irreversible:
                raise RollbackError(
                    f"检查点{checkpoint_id}后存在不可逆操作，无法回滚"
                )
            mutation.reverse()
        return target["agent_state"]

stateDiagram-v2
    [*] --> Closed: Agent启动
    Closed --> Open: 连续失败N次
    Open --> HalfOpen: 冷却期后
    HalfOpen --> Closed: 探测成功
    HalfOpen --> Open: 探测失败

    state Closed {
        [*] --> 正常执行
        正常执行 --> 记录成功
    }

    state Open {
        [*] --> 拒绝新任务
        拒绝新任务 --> 返回降级响应
    }

    state HalfOpen {
        [*] --> 放行一个探测任务
        放行一个探测任务 --> 评估结果
    }

# 修复后的压缩逻辑
async def compact_with_guard(self, messages, max_retries=3):
    last_failure_hash = None
    for attempt in range(max_retries):
        result = await self.compact(messages)
        if result.success:
            return result
        failure_hash = hash(result.error_message)
        if failure_hash == last_failure_hash:
            # 相同错误重复出现，重试无效
            raise CompactionStalled(
                f"压缩连续{attempt+1}次以相同原因失败：{result.error_message}"
            )
        last_failure_hash = failure_hash
    raise MaxCompactionRetriesExceeded()

# 逆向假设检验
class HypothesisChallenger:
    async def challenge(self, current_diagnosis, failed_attempts):
        prompt = f"""
        我的诊断"{current_diagnosis}"已经尝试修复{failed_attempts}次但问题未解决。
        这强烈暗示诊断可能是错的。

        请列出至少3个与当前诊断完全不同的替代假设，
        并为每个假设设计一个低成本的验证实验。
        """
        alternatives = await self.llm.generate(prompt)
        return alternatives

Level 0: 无自愈 — Agent挂了就挂了
Level 1: 基础重试 — 有重试逻辑但无上限
Level 2: 有界恢复 — 重试有上限，失败后降级
Level 3: 检查点回滚 — 可回退到已知良好状态
Level 4: 认知自愈 — 能识别自身错误并修正策略
Level 5: 预测性自愈 — 在故障发生前预判并预防