mcpField
C
cascade-workflow

cascade-workflow

github.com/rysweet开发76131 次安装21 周前更新
代码仓库

关于 cascade-workflow

cascade-workflow 是用于设计级联回退策略的技能,在最优方案失败或超时时仍提供可用结果。它适用于部分结果优于无结果的用户场景,并涵盖降级通知与指标跟踪。

主要功能

  1. 01三级回退方案设计
  2. 02可配置超时策略
  3. 03服务、质量与数据回退
  4. 04降级通知配置
  5. 05级联指标与优化

适用场景

  1. 01保持依赖 API 的操作响应
  2. 02在时限内返回部分结果
  3. 03为高可用场景规划回退

安装

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

$npx skills add https://github.com/rysweet/amplihack --skill cascade-workflow

安装后在 Claude Code 中输入/cascade-workflow确认已加载。

详细介绍

具备优雅降级能力的级联工作流技能

用途

通过级联回退策略实现优雅降级。当最优方案失败或超时,系统会自动回退到更简单、更可靠的替代方案,同时保持可接受的功能。

何时使用此技能

适用于:

  • 依赖外部服务的场景(API、数据库)
  • 时间敏感且可接受降级模式的操作
  • 部分结果优于没有结果的操作
  • 高可用性要求(系统必须始终响应)
  • 等待完美方案不如采用足够好的方案的场景

不适用于:

  • 要求精确正确的操作(无法接受降级)
  • 安全关键操作(身份验证、授权)
  • 金融交易(无法接受“近似”)
  • 必须向用户呈现故障的场景(诊断操作)
  • 没有有意义回退方案的简单操作

配置

核心参数

超时策略:

  • aggressive - 快速失败,快速降级(5s / 2s / 1s)
  • balanced - 合理尝试(30s / 10s / 5s)- 默认
  • patient - 回退前充分尝试(120s / 30s / 10s)
  • custom - 自定义超时时间

回退类型:

  • service - 外部 API → 缓存数据 → 静态默认值
  • quality - 全面 → 标准 → 最简分析
  • freshness - 实时 → 近期 → 历史数据
  • completeness - 完整数据集 → 样本 → 摘要
  • accuracy - 精确 → 近似 → 估算

降级通知:

  • silent - 仅记录日志,不通知用户
  • warning - 告知用户发生了降级
  • explicit - 详细说明降级内容及原因

级联级别要求

PRIMARY(最优):

  • 尽可能获得最佳结果
  • 可能依赖外部服务
  • 可能较慢或占用大量资源
  • 可能失败或超时

SECONDARY(可接受):

  • 质量有所降低,但仍可用
  • 比 PRIMARY 更可靠
  • 速度更快或依赖更少
  • 用户可以接受

TERTIARY(有保障):

  • 始终成功,绝不失败
  • 不依赖外部服务
  • 快速且可靠
  • 功能精简但可用
  • 关键:必须设计为永不失败

执行流程

步骤 1:定义级联级别
  • 使用架构师 Agent 确定级联级别
  • 定义 PRIMARY 方案(最优解决方案)
  • 定义 SECONDARY 方案(可接受的降级方案)
  • 定义 TERTIARY 方案(保证完成的方案)
  • 为每个级别设置超时时间
  • 记录每个级别的降级内容
  • 关键:确保 TERTIARY 始终成功

级联定义示例:

使用 AI 进行代码分析:

  • PRIMARY:GPT-4 全面分析(超时:30s)
  • SECONDARY:GPT-3.5 标准分析(超时:10s)
  • TERTIARY:使用正则表达式进行静态分析(超时:5s)

获取外部 API 数据:

  • PRIMARY:调用实时 API(超时:10s)
  • SECONDARY:缓存数据(超时:2s)
  • TERTIARY:默认值(超时:0s)

测试执行:

  • PRIMARY:完整测试套件(超时:120s)
  • SECONDARY:仅运行关键测试(超时:30s)
  • TERTIARY:冒烟测试(超时:10s)
步骤 2:尝试 PRIMARY 方案
  • 执行最优方案
  • 根据策略配置设置超时时间
  • 监控执行进度
  • 如果成功完成:结束(最佳结果)
  • 如果失败或超时:继续步骤 3
  • 记录尝试情况及失败原因
# Pseudocode for primary attempt
try:
    result = execute_primary_approach(timeout=PRIMARY_TIMEOUT)
    log_success(level="PRIMARY", result=result)
    return result  # DONE - best outcome achieved
except TimeoutError:
    log_failure(level="PRIMARY", reason="timeout")
    # Continue to Step 3
except ExternalServiceError as e:
    log_failure(level="PRIMARY", reason=f"service_error: {e}")
    # Continue to Step 3
步骤 3:尝试 SECONDARY 方案
  • 记录降级到 SECONDARY 级别
  • 执行可接受的回退方案
  • 设置更短的超时时间(通常为 PRIMARY 的 1/3)
  • 监控执行进度
  • 如果成功完成:结束(可接受的结果)
  • 如果失败或超时:继续步骤 4
  • 记录尝试情况及失败原因
# Pseudocode for secondary attempt
log_degradation(from_level="PRIMARY", to_level="SECONDARY")
try:
    result = execute_secondary_approach(timeout=SECONDARY_TIMEOUT)
    log_success(level="SECONDARY", result=result, degraded=True)
    return result  # DONE - acceptable outcome
except TimeoutError:
    log_failure(level="SECONDARY", reason="timeout")
    # Continue to Step 4
步骤 4:尝试第三级方案
  • 记录降级至第三级
  • 执行保证完成的方案
  • 设置最短超时时间(通常为 1s)
  • 必须成功——不允许失败
  • 返回精简但可用的结果
  • 记录成功(已降级但仍可用)
  • 完成(保证完成)
# Pseudocode for tertiary attempt
log_degradation(from_level="SECONDARY", to_level="TERTIARY")
try:
    result = execute_tertiary_approach(timeout=TERTIARY_TIMEOUT)
    log_success(level="TERTIARY", result=result, heavily_degraded=True)
    return result  # DONE - minimal but functional
except Exception as e:
    # THIS SHOULD NEVER HAPPEN
    log_critical_failure("TERTIARY approach failed - this is a bug!")
    raise SystemError("Cascade safety violation: tertiary failed")
步骤 5:报告降级情况
  • 根据配置确定通知级别
  • **静默:**仅记录日志,不向用户发送消息
  • **警告:**向用户发送简短通知
  • **明确:**详细说明降级情况
  • 记录成功的级别
  • 说明降级造成的影响
  • 记录采取的级联路径以供分析

降级报告模板:

静默模式:

[LOG] CASCADE: PRIMARY timeout (30s) → SECONDARY success (6s)
Result: standard_analysis (degraded from comprehensive)

警告模式:

⚠️  Using cached data (less than 1 hour old)
Current real-time data unavailable.

明确模式:

ℹ️  Analysis Quality Notice

We attempted to provide comprehensive code analysis using GPT-4,
but encountered slow response times (>30s timeout).

Fallback Applied:
- Used: GPT-3.5 standard analysis (completed in 6s)
- Quality: Standard (vs. Comprehensive)
- Impact: Advanced semantic insights not included

What You're Getting:
✓ Basic pattern detection
✓ Standard recommendations
✓ Code quality assessment

What's Missing:
✗ Complex architectural insights
✗ Deep semantic analysis
✗ Advanced refactoring suggestions
步骤 6:记录级联指标
  • 记录采取的级联路径
  • 记录达到的级别(第一级/第二级/第三级)
  • 记录每个尝试级别的耗时
  • 跟踪降级频率
  • 识别失败模式
  • 必要时更新级联策略

跟踪指标:

  • 各级别的成功率
  • 平均响应时间
  • 降级频率
  • 用户影响评估
步骤 7:持续优化
  • 使用分析器 Agent检查级联指标
  • 识别优化机会
  • 根据成功率调整超时时间
  • 如果经常使用第二级方案,则改进该方案
  • 如果第三级方案不够完善,则进行更新
  • 使用 amplihack.memory.discoveries 中的 store_discovery() 将经验存入记忆

优化标准:

  • **如果 PRIMARY 成功率低于 50%:**超时时间过于激进 → 增加超时时间
  • **如果 SECONDARY 的使用率超过 40%:**第二级方案实际上是“常规”方案 → 交换第一级和第二级方案
  • **如果 TERTIARY 的使用率超过 10%:**第二级方案不够可靠 → 改进第二级方案

权衡

**优点:**系统总能完成,不会彻底失败
**代价:**用户可能收到降级后的响应
**适用场景:**注重响应速度的面向用户功能

示例

示例 1:天气 API 集成

配置:

  • 策略:均衡(30s / 10s / 5s)
  • 类型:服务回退
  • 通知:警告

实现:

async def get_weather(location: str) -> WeatherData:
    """Get weather data with cascade fallback"""

    # PRIMARY: Live weather API
    try:
        return await fetch_weather_api(location, timeout=30)
    except (TimeoutError, APIError):
        log.warning("PRIMARY weather API failed, trying cache")

    # SECONDARY: Cached weather data
    try:
        cached = await get_cached_weather(location, max_age=3600)
        if cached:
            notify_user("Using weather data from cache (< 1 hour old)")
            return cached
    except CacheError:
        log.warning("SECONDARY cache failed, using defaults")

    # TERTIARY: Default weather data
    return get_default_weather(location)  # Never fails

**结果:**系统始终会返回天气数据,质量会平稳降级

示例 2:使用 AI 进行代码审查

配置:

  • 策略:耐心(120s / 30s / 10s)
  • 类型:质量回退
  • 通知:明确

级联路径:

  1. PRIMARY:GPT-4 全面审查——120s 后超时
  2. SECONDARY:GPT-3.5 标准审查——18s 内成功
  3. 未尝试 TERTIARY
示例 3:搜索结果排序

配置:

  • 策略:激进(5s / 2s / 1s)
  • 类型:准确性回退
  • 通知:静默

实现:

def search_and_rank(query: str) -> List[Result]:
    """Search with ML ranking, fallback to simple ranking"""

    results = fetch_results(query)

    # PRIMARY: ML-based ranking (sophisticated)
    try:
        return ml_rank(results, timeout=5)
    except TimeoutError:
        pass  # Silent fallback

    # SECONDARY: Heuristic ranking (good enough)
    try:
        return heuristic_rank(results, timeout=2)
    except TimeoutError:
        pass

    # TERTIARY: Simple text match ranking (basic)
    return simple_rank(results)  # Always fast

理念契合

此工作流贯彻以下理念:

  • **弹性:**系统总能完成,不会彻底失败
  • **用户体验:**提供降级服务优于显示错误消息
  • **透明度:**让用户了解自己获得的服务(明确模式下)
  • **渐进式增强:**默认提供最优体验,必要时再降级
  • **质量可衡量:**明确界定各级别的降级内容
  • **持续改进:**根据指标优化超时时间
  • **保证完成:**第三级方案绝不能失败

核心原则

提供降级服务也胜过完全不提供服务

评论

支持 Markdown