mcpField
S
self-healing-agents

self-healing-agents

github.com/itallstartedwithaideaAgent 工作流4493 次安装28 周前更新
代码仓库

关于 self-healing-agents

self-healing-agents 是用于设计故障检测与恢复机制的 Agent 技能。它介绍按错误类型采取恢复策略,包括调整方案重试、切换备用模型、修复输出,以及在无法完全恢复时平稳降级。

主要功能

  1. 01故障分类
  2. 02退避重试
  3. 03调整策略后重试
  4. 04选择备用模型
  5. 05输出校验与修复
  6. 06平稳降级

适用场景

  1. 01构建适应不稳定 API 的自主 Agent
  2. 02处理速率限制与网络超时
  3. 03修复不符合结构要求的输出
  4. 04避免长时工作流直接失败

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/itallstartedwithaidea/agent-skills --skill self-healing-agents

安装后在 Claude Code 中输入/self-healing-agents确认已加载。

详细介绍

自愈型 Agent

Agent Skills™ 的一部分,由 googleadsagent.ai™ 提供

描述

自愈型 Agent 是能够检测自身故障模式并在无需人工干预的情况下自我纠正的自主系统。在生产环境中,Agent 故障并非异常情况,而是意料之中的事。网络调用会超时,API 会返回意外的架构,大模型会自信地编造内容,工具输出也会违反预期。原型与生产级 Agent 的区别,在于能否从各类故障中妥善恢复。

这项技能封装了为 Buddy™ Agent 开发的自愈模式。在 googleadsagent.ai™,即使上游 API 发生变化、触发速率限制,或大模型输出包含结构错误,自主式 Google Ads 分析也必须可靠完成。该系统采用检测—诊断—修复循环,模拟生物免疫反应:识别病原体、判断威胁类型,并部署适当的应对措施。

自愈不只是重试逻辑。它还包括错误分类、策略调整(换一种方法重试,而不是原样重复)、备用模型选择、带自动修复的输出验证,以及无法完全恢复时的优雅降级。采用这些模式构建的 Agent 在生产环境中的任务完成率可达 99% 以上。

适用场景

  • 构建必须在无人监督下自主运行的 Agent
  • 工具调用或 API 集成不可靠,或受到速率限制
  • 大模型输出必须符合严格的架构,但有时并不符合
  • 长时间运行的工作流无法承受执行中途失败
  • 需要确保由 Agent 驱动的功能满足 SLA 承诺
  • Agent 必须处理此前未遇到过的新型错误

工作原理

graph TD
    A[Agent Action] --> B[Output Validation]
    B -->|Valid| C[Continue Execution]
    B -->|Invalid| D[Error Classifier]
    D --> E{Error Type}
    E -->|Transient| F[Retry with Backoff]
    E -->|Structural| G[Mutate Strategy]
    E -->|Model Error| H[Fallback Model]
    E -->|Unrecoverable| I[Graceful Degradation]
    F --> J{Retry Budget Remaining?}
    J -->|Yes| A
    J -->|No| G
    G --> K[Modified Prompt/Approach]
    K --> A
    H --> L[Alternative Model Execution]
    L --> B
    I --> M[Partial Result + Error Report]

每当输出验证检测到异常时,自愈循环便会启动。错误分类器将故障分为四种类型:瞬时错误(网络超时、速率限制)会通过指数退避重试;结构错误(架构违规、字段缺失)会触发策略调整,让 Agent 修改处理方法;模型错误(幻觉、拒绝)会调用备用模型;无法恢复的错误则会触发优雅降级,返回最佳的部分结果,并附上清晰的错误报告。

实现

错误分类引擎:

enum ErrorType {
  Transient = "transient",
  Structural = "structural",
  ModelError = "model_error",
  Unrecoverable = "unrecoverable",
}

interface ClassifiedError {
  type: ErrorType;
  message: string;
  retryable: boolean;
  suggestedStrategy: string;
}

function classifyError(error: unknown, context: ExecutionContext): ClassifiedError {
  if (error instanceof NetworkError || error instanceof RateLimitError) {
    return {
      type: ErrorType.Transient,
      message: String(error),
      retryable: true,
      suggestedStrategy: "exponential_backoff",
    };
  }
  if (error instanceof SchemaValidationError) {
    return {
      type: ErrorType.Structural,
      message: `Schema violation: ${error.path} — ${error.message}`,
      retryable: true,
      suggestedStrategy: "mutate_prompt",
    };
  }
  if (error instanceof ModelRefusalError || isHallucination(error, context)) {
    return {
      type: ErrorType.ModelError,
      message: String(error),
      retryable: true,
      suggestedStrategy: "fallback_model",
    };
  }
  return {
    type: ErrorType.Unrecoverable,
    message: String(error),
    retryable: false,
    suggestedStrategy: "graceful_degradation",
  };
}

自愈执行循环:

class SelfHealingAgent:
    def __init__(self, primary_model, fallback_models, max_retries=3):
        self.primary_model = primary_model
        self.fallback_models = fallback_models
        self.max_retries = max_retries

    async def execute(self, task, validator):
        attempts = []
        current_model = self.primary_model
        current_prompt = task.prompt

        for attempt in range(self.max_retries + len(self.fallback_models)):
            try:
                result = await current_model.generate(current_prompt)
                validation = validator.validate(result)
                if validation.is_valid:
                    return HealingResult(result=result, attempts=attempts)

                error = classify_error(validation.error, task.context)
                attempts.append({"attempt": attempt, "error": error, "model": current_model.name})

                if error.strategy == "mutate_prompt":
                    current_prompt = self.mutate_prompt(current_prompt, error)
                elif error.strategy == "fallback_model":
                    current_model = self.next_fallback(current_model)

            except Exception as e:
                error = classify_error(e, task.context)
                attempts.append({"attempt": attempt, "error": error})
                if not error.retryable:
                    break
                await asyncio.sleep(2 ** attempt)

        return HealingResult(
            result=self.graceful_degradation(task, attempts),
            attempts=attempts,
            degraded=True
        )

    def mutate_prompt(self, prompt, error):
        mutations = {
            "schema_violation": f"{prompt}\n\nPrevious attempt had error: {error.message}. Ensure strict schema compliance.",
            "missing_field": f"{prompt}\n\nYou MUST include all required fields. Missing: {error.message}",
        }
        return mutations.get(error.subtype, f"{prompt}\n\nPrevious error: {error.message}. Adjust approach.")

    def next_fallback(self, current):
        idx = ([self.primary_model] + self.fallback_models).index(current)
        if idx < len(self.fallback_models):
            return self.fallback_models[idx]
        return current

带自动修复的输出验证:

def validate_and_repair(output: str, schema: dict) -> tuple[dict, bool]:
    try:
        parsed = json.loads(output)
    except json.JSONDecodeError:
        extracted = extract_json_from_text(output)
        if extracted:
            parsed = extracted
        else:
            raise StructuralError("No valid JSON found in output")

    repaired = False
    for field, rules in schema.get("required_fields", {}).items():
        if field not in parsed:
            if "default" in rules:
                parsed[field] = rules["default"]
                repaired = True
            else:
                raise StructuralError(f"Missing required field: {field}")

    return parsed, repaired

最佳实践

  1. 先分类,再重试 — 盲目重试会浪费令牌和时间;始终先判断错误类型,再选择最佳恢复策略。
  2. 重试时调整策略,不要重复原样操作 — 如果同一提示词已失败,再次原样发送很少能解决问题;应补充错误上下文或重新组织请求。
  3. 设定严格的重试预算 — 限制重试总次数(通常为 3–5 次)和令牌总开销,以防自愈循环失控。
  4. 记录每次自愈尝试 — 每次重试、策略调整和备用方案调用都应记录完整上下文,以便事后分析。
  5. 设计优雅降级输出 — 带有清晰错误注释的部分结果,远比空白失败有用。
  6. 明确测试自愈路径 — 在测试中注入已知故障模式,验证每条自愈路径都能正确启动。
  7. 监控自愈率 — 自愈率超过 10% 表明上游存在问题,应从源头修复,而不是靠自愈机制掩盖。

平台兼容性

功能Claude CodeCursorCodexGemini CLI
带策略变更的重试✅ 完整✅ 完整✅ 完整✅ 完整
后备模型选择✅ 通过 API✅ 通过扩展✅ 通过 API✅ 通过 API
输出验证✅ 完整✅ 完整✅ 完整✅ 完整
错误分类✅ 完整✅ 完整✅ 完整✅ 完整
优雅降级✅ 完整✅ 完整✅ 完整✅ 完整

相关技能

  • 验证循环 - 多阶段验证流水线,可检测自愈 Agent 能够恢复的故障
  • Agent 本能系统 - 操作前安全关卡,可在错误需要自愈之前将其阻止
  • 对抗性韧性 - 防御层,可处理对抗性因素引发的故障,以及瞬态错误和结构性错误

关键词

自愈、错误恢复、重试策略、后备模型、输出验证、优雅降级、策略变更、错误分类、韧性、Agent 技能


© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License

评论

支持 Markdown