26d - 错误处理模式

工作流执行出错
├── 错误是否为瞬时故障？（网络超时、429、5xx）
│   ├── 是 → 自动重试（指数退避）
│   │   ├── 重试成功 → 继续执行
│   │   └── 重试耗尽 → 进入死信队列 + 告警
│   └── 否 → 错误是否可降级？
│       ├── 是 → 执行降级路径（备用模型/缓存/默认值）
│       │   └── 降级成功 → 继续执行（标记降级）
│       └── 否 → 错误是否为数据问题？
│           ├── 是 → 记录到死信队列 + 跳过当前项
│           └── 否 → 终止执行 + 紧急告警

┌─────────────────────────────────────────────┐
│  第三层：Error Workflow（全局错误工作流）       │
│  ┌─────────────────────────────────────────┐ │
│  │  第二层：Error Output（节点错误输出分支）  │ │
│  │  ┌─────────────────────────────────────┐ │ │
│  │  │  第一层：Retry on Fail（节点自动重试）│ │ │
│  │  └─────────────────────────────────────┘ │ │
│  └─────────────────────────────────────────┘ │
└─────────────────────────────────────────────┘

{
  "retryOnFail": true,
  "maxTries": 3,
  "waitBetweenTries": 2000
}

// n8n Code 节点：幂等性检查示例
const itemId = $input.item.json.id;
const processedKey = `processed_${itemId}`;
 
// 检查是否已处理（使用 n8n 静态数据或外部存储）
const staticData = $getWorkflowStaticData('global');
if (staticData[processedKey]) {
  return []; // 已处理，跳过
}
 
// 标记为已处理
staticData[processedKey] = new Date().toISOString();
return $input.item;

{
  "error": {
    "message": "Request failed with status code 429",
    "name": "NodeApiError",
    "description": "Rate limit exceeded",
    "httpCode": "429"
  },
  "pairedItem": {
    "item": 0
  }
}

[HTTP Request: 主 API]
├── ✅ 成功输出 → [后续处理节点]
└── ❌ 错误输出 → [Switch: 错误分类]
    ├── 429 → [Wait 节点] → [HTTP Request: 重试]
    ├── 401 → [Slack: 认证过期告警]
    ├── 500 → [HTTP Request: 备用 API]
    └── 其他 → [Google Sheets: 记录错误] → [Slack: 通用告警]

// Switch 节点条件配置
// 条件 1：速率限制
{{ $json.error.httpCode === "429" }}
 
// 条件 2：认证失败
{{ $json.error.httpCode === "401" || $json.error.httpCode === "403" }}
 
// 条件 3：服务端错误
{{ parseInt($json.error.httpCode) >= 500 }}
 
// 条件 4：其他错误（Fallthrough）

{
  "execution": {
    "id": "231",
    "url": "https://your-n8n.com/execution/231",
    "retryOf": null,
    "error": {
      "message": "Request failed with status code 500",
      "name": "NodeApiError"
    },
    "lastNodeExecuted": "HTTP Request",
    "mode": "trigger"
  },
  "workflow": {
    "id": "15",
    "name": "AI 内容管线"
  }
}

[Error Trigger]
  → [Code: 错误分级]
  → [Switch: 严重度路由]
      ├── 🔴 Critical → [PagerDuty: 创建事件] + [Slack: #alerts-critical]
      ├── 🟡 Warning  → [Slack: #alerts-warning]
      └── 🟢 Info     → [Google Sheets: 错误日志]

// Code 节点：错误严重度分级
const error = $input.item.json.execution.error;
const workflow = $input.item.json.workflow;
const errorMessage = error.message || '';
 
let severity = 'info';
let emoji = '🟢';
 
// Critical：认证失败、数据库连接失败、支付相关
if (
  errorMessage.includes('401') ||
  errorMessage.includes('403') ||
  errorMessage.includes('ECONNREFUSED') ||
  workflow.name.toLowerCase().includes('payment') ||
  workflow.name.toLowerCase().includes('billing')
) {
  severity = 'critical';
  emoji = '🔴';
}
// Warning：速率限制、超时、5xx
else if (
  errorMessage.includes('429') ||
  errorMessage.includes('timeout') ||
  errorMessage.includes('ETIMEDOUT') ||
  errorMessage.includes('500') ||
  errorMessage.includes('502') ||
  errorMessage.includes('503')
) {
  severity = 'warning';
  emoji = '🟡';
}
 
return {
  severity,
  emoji,
  workflow_name: workflow.name,
  workflow_id: workflow.id,
  error_message: errorMessage,
  execution_id: $input.item.json.execution.id,
  execution_url: $input.item.json.execution.url,
  timestamp: new Date().toISOString()
};

${emoji} *工作流错误告警*

*工作流*：${workflow_name}（ID: ${workflow_id}）
*严重度*：${severity}
*错误信息*：${error_message}
*执行 ID*：${execution_id}
*时间*：${timestamp}

<${execution_url}|查看执行详情>

[模块 A: 扣款] → [模块 B: 发货] → [模块 C: 通知]
                       │
                       └── ❌ Error Handler → [Rollback]
                           （模块 A 的扣款也会被回滚）

[获取数据] → [AI 摘要生成] → [发送 Slack 通知]
                                    │
                                    └── ❌ Error Handler → [Ignore]
                                        （通知失败不影响整体流程）

[AI 摘要生成]
├── ✅ 成功 → [后续处理]（使用 AI 摘要）
└── ❌ Error Handler
    → [Text: 截取原文前200字]
    → [Resume]（使用截取文本作为替代输出）
    → [后续处理]（使用替代摘要继续）

[HTTP Request: 调用外部 API]
│
└── ❌ Error Handler
    → [Break]
        ├── 自动重试：3 次，间隔 15 分钟
        └── 重试耗尽 → 留在 Incomplete Executions 队列
            → 人工在面板中查看并处理

[目标模块]
│
└── ❌ Error Route
    → [Slack: 发送错误通知]
    → [Google Sheets: 记录错误日志]
    → [Resume / Break / Ignore / Rollback / Commit]

等待时间 = min(base_delay × 2^(attempt - 1) + random_jitter, max_delay)

[触发器] → [设置重试参数] → [Loop]
    → [HTTP Request: API 调用]
    ├── ✅ 成功 → [跳出循环] → [后续处理]
    └── ❌ 错误输出 → [Code: 计算退避时间]
        → [IF: 是否超过最大重试次数？]
        ├── 是 → [跳出循环] → [错误处理/死信队列]
        └── 否 → [Wait: 动态等待] → [回到循环开始]

{
  "attempt": 0,
  "maxAttempts": 5,
  "baseDelay": 1000,
  "maxDelay": 60000
}

// Code 节点：计算指数退避等待时间
const attempt = $input.item.json.attempt + 1;
const baseDelay = $input.item.json.baseDelay || 1000;
const maxDelay = $input.item.json.maxDelay || 60000;
 
// 指数退避 + 随机抖动
const exponentialDelay = baseDelay * Math.pow(2, attempt - 1);
const jitter = Math.random() * exponentialDelay * 0.5;
const waitTime = Math.min(exponentialDelay + jitter, maxDelay);
 
return {
  attempt,
  maxAttempts: $input.item.json.maxAttempts,
  baseDelay,
  maxDelay,
  waitTime: Math.round(waitTime),
  waitTimeSeconds: Math.round(waitTime / 1000),
  error: $input.item.json.error
};

条件：{{ $json.attempt >= $json.maxAttempts }}
├── true  → 退出循环，进入错误处理
└── false → 继续循环，回到 API 调用

[主工作流]
  → [Execute Sub-Workflow: 带退避的 API 调用]
      ├── 输入：url, method, body, attempt, maxAttempts
      ├── 内部逻辑：
      │   [HTTP Request]
      │   ├── ✅ 成功 → 返回结果
      │   └── ❌ 失败 → [计算退避] → [Wait] → [递归调用自身]
      └── 输出：API 响应或最终错误

[模块 A: API 调用]
│
└── ❌ Error Handler
    → [Set Variable: retry_count + 1]
    → [Router]
        ├── 条件：retry_count <= max_retries
        │   → [Sleep: 动态计算时间]
        │   → [模块 A 的克隆: 重试 API 调用]
        │   ├── ✅ 成功 → [Resume]（使用重试结果继续）
        │   └── ❌ 失败 → [回到 Router]
        └── 条件：retry_count > max_retries
            → [Slack: 告警通知]
            → [Google Sheets: 记录到死信队列]
            → [Ignore / Break]

变量名：wait_seconds
值：{{min(pow(2, retry_count) * 1 + floor(random * pow(2, retry_count) * 0.5), 60)}}

你是一个 n8n 工作流错误处理专家。请为以下场景设计指数退避重试逻辑：

**场景描述**：[描述你的 API 调用场景，例如"调用 OpenAI API 生成文章摘要"]
**预期错误类型**：[例如"429 速率限制、500 服务器错误、网络超时"]
**最大重试次数**：[例如 5]
**基础延迟**：[例如 1 秒]
**最大延迟**：[例如 60 秒]

请提供：
1. 完整的 n8n 节点拓扑图（ASCII 格式）
2. Code 节点中的退避时间计算代码
3. 各节点的关键配置参数
4. 重试耗尽后的处理建议

[生产工作流]
  → [节点执行失败]
  → [重试耗尽]
  → [Error Output / Error Workflow]
  → [DLQ Writer: 写入死信队列]
      ├── 存储到 Postgres/Supabase 表
      ├── 或存储到 Google Sheets
      └── 或存储到 Redis List

[DLQ 处理工作流]（独立工作流）
  → [Schedule Trigger: 定时扫描]
  → [读取 DLQ 中的待处理项]
  → [逐项重试]
  ├── ✅ 成功 → 标记为已处理
  └── ❌ 失败 → 更新重试计数
      ├── 未超限 → 保留在队列中
      └── 已超限 → 标记为永久失败 + 告警

CREATE TABLE dead_letter_queue (
  id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
  workflow_id VARCHAR(50) NOT NULL,
  workflow_name VARCHAR(255) NOT NULL,
  execution_id VARCHAR(50),
  node_name VARCHAR(255),
  error_message TEXT,
  error_code VARCHAR(20),
  input_data JSONB,
  retry_count INTEGER DEFAULT 0,
  max_retries INTEGER DEFAULT 5,
  status VARCHAR(20) DEFAULT 'pending',  -- pending, retrying, resolved, dead
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  next_retry_at TIMESTAMP,
  resolved_at TIMESTAMP
);
 
-- 索引优化
CREATE INDEX idx_dlq_status ON dead_letter_queue(status);
CREATE INDEX idx_dlq_next_retry ON dead_letter_queue(next_retry_at)
  WHERE status = 'pending';
CREATE INDEX idx_dlq_workflow ON dead_letter_queue(workflow_id);

INSERT INTO dead_letter_queue
  (workflow_id, workflow_name, execution_id, node_name,
   error_message, error_code, input_data, next_retry_at)
VALUES
  ($1, $2, $3, $4, $5, $6, $7::jsonb,
   NOW() + INTERVAL '15 minutes')

[Schedule Trigger: 每 15 分钟]
  → [Postgres: 查询待处理项]
      SELECT * FROM dead_letter_queue
      WHERE status = 'pending'
        AND next_retry_at <= NOW()
      ORDER BY created_at ASC
      LIMIT 10
  → [Loop Over Items]
      → [Code: 构建重试请求]
      → [HTTP Request: 重试原始操作]
      ├── ✅ 成功
      │   → [Postgres: UPDATE status='resolved', resolved_at=NOW()]
      └── ❌ 失败
          → [Code: 更新重试计数和下次重试时间]
          → [IF: retry_count >= max_retries?]
          ├── 是 → [Postgres: UPDATE status='dead']
          │       → [Slack: 永久失败告警]
          └── 否 → [Postgres: UPDATE retry_count, next_retry_at]

// Code 节点：计算下次重试时间（指数退避）
const retryCount = $input.item.json.retry_count + 1;
const baseMinutes = 15;
const maxMinutes = 1440; // 24 小时
 
const delayMinutes = Math.min(
  baseMinutes * Math.pow(2, retryCount - 1),
  maxMinutes
);
 
const nextRetryAt = new Date(
  Date.now() + delayMinutes * 60 * 1000
).toISOString();
 
return {
  ...$input.item.json,
  retry_count: retryCount,
  next_retry_at: nextRetryAt,
  delay_minutes: delayMinutes
};

[目标模块]
│
└── ❌ Error Handler
    → [HTTP Request: POST 到 Supabase/Airtable]
        Body: {
          "scenario_id": "{{scenario.id}}",
          "scenario_name": "{{scenario.name}}",
          "error": "{{error.message}}",
          "input_data": {{toJSON(input)}},
          "timestamp": "{{now}}"
        }
    → [Slack: 发送 DLQ 入队通知]
    → [Ignore]（让场景继续处理其他项目）

-- DLQ 概览仪表板
SELECT
  status,
  COUNT(*) as count,
  AVG(retry_count) as avg_retries,
  MIN(created_at) as oldest_item,
  MAX(created_at) as newest_item
FROM dead_letter_queue
WHERE created_at >= NOW() - INTERVAL '24 hours'
GROUP BY status;
 
-- 按工作流分组的失败统计
SELECT
  workflow_name,
  COUNT(*) as total_failures,
  COUNT(*) FILTER (WHERE status = 'resolved') as resolved,
  COUNT(*) FILTER (WHERE status = 'dead') as permanent_failures,
  ROUND(
    COUNT(*) FILTER (WHERE status = 'resolved')::numeric /
    NULLIF(COUNT(*), 0) * 100, 1
  ) as resolution_rate_pct
FROM dead_letter_queue
WHERE created_at >= NOW() - INTERVAL '7 days'
GROUP BY workflow_name
ORDER BY total_failures DESC;

[输入数据]
  → [OpenAI: GPT-4.1]（主模型）
  ├── ✅ 成功 → [输出验证] → [后续处理]
  └── ❌ Error Output
      → [Anthropic: Claude Sonnet]（备用模型 1）
      ├── ✅ 成功 → [输出验证] → [后续处理]
      └── ❌ Error Output
          → [Google AI: Gemini 2.5 Flash]（备用模型 2）
          ├── ✅ 成功 → [输出验证] → [后续处理]
          └── ❌ Error Output
              → [缓存/默认值降级] → [后续处理（标记降级）]

// Code 节点：最终降级逻辑
const originalInput = $input.item.json.originalText || '';
 
// 使用简单的文本截取作为降级摘要
const fallbackSummary = originalInput.substring(0, 200) + '...';
 
return {
  summary: fallbackSummary,
  model_used: 'fallback_truncation',
  is_degraded: true,
  degradation_reason: 'All LLM providers unavailable',
  timestamp: new Date().toISOString()
};

// 在输出中添加降级标记
const isDegraded = $input.item.json.is_degraded || false;
const modelUsed = $input.item.json.model_used || 'gpt-4.1';
 
return {
  ...$input.item.json,
  metadata: {
    model_used: modelUsed,
    is_degraded: isDegraded,
    processed_at: new Date().toISOString()
  }
};

[输入数据]
  → [Code: 生成缓存键]
  → [Redis/Supabase: 查询缓存]
  → [IF: 缓存命中？]
  ├── 是 → [使用缓存数据]（跳过 API 调用）
  └── 否 → [API 调用]
      ├── ✅ 成功 → [Redis/Supabase: 写入缓存] → [后续处理]
      └── ❌ 失败 → [使用默认值] → [后续处理（标记降级）]

[核心数据处理]
  → [AI 情感分析]（可选增强）
  ├── ✅ 成功 → [合并结果]
  └── ❌ Error Output → [Set: sentiment = "unknown"] → [合并结果]
  → [AI 关键词提取]（可选增强）
  ├── ✅ 成功 → [合并结果]
  └── ❌ Error Output → [Set: keywords = []] → [合并结果]
  → [写入数据库]（核心操作，不降级）

[OpenAI: 生成摘要]
├── ✅ 成功 → [后续处理]
└── ❌ Error Handler
    → [Router]
    ├── 路径 1：尝试备用模型
    │   → [HTTP Request: Anthropic API]
    │   ├── ✅ 成功 → [Resume]（使用备用结果）
    │   └── ❌ Error Handler → [路径 2]
    └── 路径 2：使用默认值
        → [Text: 截取原文前200字]
        → [Resume]（使用默认值）

[HTTP Request: 主 API]
│
└── ❌ Error Handler
    → [Set Variable: error_type = {{error.code}}]
    → [Router]
    ├── 条件：error_type = 429（速率限制）
    │   → [Sleep: 30秒] → [HTTP Request: 重试] → [Resume]
    ├── 条件：error_type >= 500（服务端错误）
    │   → [HTTP Request: 备用 API] → [Resume]
    └── 默认路径
        → [Slack: 告警] → [Ignore]

你是一个 [n8n/Make.com] 工作流架构师。请为以下 AI 工作流设计完整的降级路径：

**工作流描述**：[描述你的工作流，例如"从 RSS 抓取文章 → AI 生成摘要 → 翻译 → 发布到 Slack"]
**关键节点**：[列出不能失败的核心节点]
**可降级节点**：[列出可以降级的非关键节点]
**可用的备用服务**：[例如"备用 LLM: Claude, Gemini; 备用通知: Email"]

请提供：
1. 每个可降级节点的降级策略（模型降级/缓存降级/默认值降级）
2. 降级后的数据标记方案（如何标识降级数据）
3. 降级触发条件和恢复条件
4. 完整的节点拓扑图

[Error Trigger]
  → [Code: 错误分级 + 去重检查]
  → [Switch: 按严重度路由]
  ├── P0 紧急
  │   → [PagerDuty: 创建事件]
  │   → [Slack: #alerts-critical（@channel）]
  │   → [Postgres: 记录错误日志]
  ├── P1 严重
  │   → [Slack: #alerts-critical]
  │   → [Gmail: 发送告警邮件]
  │   → [Postgres: 记录错误日志]
  ├── P2 警告
  │   → [Slack: #alerts-warning]
  │   → [Postgres: 记录错误日志]
  └── P3 信息
      → [Postgres: 记录错误日志]

// Code 节点：告警去重
const error = $input.item.json.execution.error;
const workflow = $input.item.json.workflow;
 
// 生成去重键：工作流ID + 错误消息的哈希
const dedupeKey = `${workflow.id}_${error.message}`.substring(0, 100);
 
// 检查最近 30 分钟内是否已发送过相同告警
const staticData = $getWorkflowStaticData('global');
const now = Date.now();
const cooldownMs = 30 * 60 * 1000; // 30 分钟冷却期
 
if (staticData[dedupeKey] && (now - staticData[dedupeKey]) < cooldownMs) {
  // 在冷却期内，跳过告警但仍记录日志
  return {
    ...formatError($input.item.json),
    alert_suppressed: true,
    suppression_reason: 'Duplicate within cooldown period'
  };
}
 
// 更新最后告警时间
staticData[dedupeKey] = now;
 
// 清理过期的去重记录（避免内存泄漏）
for (const key of Object.keys(staticData)) {
  if (staticData[key] < now - cooldownMs * 2) {
    delete staticData[key];
  }
}
 
return {
  ...formatError($input.item.json),
  alert_suppressed: false
};
 
function formatError(data) {
  const error = data.execution.error;
  const workflow = data.workflow;
  let severity = 'P3';
  let emoji = '🟢';
 
  // P0：支付、认证、数据库连接
  if (
    workflow.name.match(/payment|billing|order/i) ||
    (error.message && error.message.match(/ECONNREFUSED|connection.*refused/i))
  ) {
    severity = 'P0';
    emoji = '🔴';
  }
  // P1：认证过期、持续性 5xx
  else if (
    error.message && error.message.match(/401|403|unauthorized/i)
  ) {
    severity = 'P1';
    emoji = '🟠';
  }
  // P2：速率限制、超时
  else if (
    error.message && error.message.match(/429|timeout|ETIMEDOUT|rate.?limit/i)
  ) {
    severity = 'P2';
    emoji = '🟡';
  }
 
  return {
    severity,
    emoji,
    workflow_name: workflow.name,
    workflow_id: workflow.id,
    error_message: error.message || 'Unknown error',
    execution_id: data.execution.id,
    execution_url: data.execution.url,
    timestamp: new Date().toISOString()
  };
}

[Schedule Trigger: 每天 09:00]
  → [Postgres: 查询过去 24 小时错误统计]
  → [Code: 生成报告]
  → [Slack: 发送每日错误摘要]

SELECT
  workflow_name,
  COUNT(*) as error_count,
  COUNT(DISTINCT error_message) as unique_errors,
  MIN(created_at) as first_occurrence,
  MAX(created_at) as last_occurrence,
  ARRAY_AGG(DISTINCT error_code) as error_codes
FROM dead_letter_queue
WHERE created_at >= NOW() - INTERVAL '24 hours'
GROUP BY workflow_name
ORDER BY error_count DESC
LIMIT 20;

📊 *每日工作流错误报告*
📅 ${date}

*概览*
• 总错误数：${totalErrors}
• 受影响工作流：${affectedWorkflows}
• 已自动恢复：${autoResolved}（${autoResolvedPct}%）
• 需要人工处理：${manualRequired}

*Top 5 错误工作流*
${top5List}

*DLQ 状态*
• 队列深度：${queueDepth}
• 永久失败：${deadCount}
• 待重试：${pendingCount}

<${dashboardUrl}|查看完整仪表板>

[目标模块]
│
└── ❌ Error Handler
    → [Slack: 发送告警]
        Channel: #automation-alerts
        Message: "⚠️ 场景 '{{scenario.name}}' 执行失败
                  模块: {{failedModule.name}}
                  错误: {{error.message}}
                  时间: {{formatDate(now; 'YYYY-MM-DD HH:mm:ss')}}"
    → [Google Sheets: 记录错误日志]
        Spreadsheet: "Automation Error Log"
        Row: [{{now}}, {{scenario.name}}, {{error.message}}, {{error.type}}]
    → [Break / Resume / Ignore]

{
  "blocks": [
    {
      "type": "header",
      "text": {
        "type": "plain_text",
        "text": "${emoji} 工作流错误告警 [${severity}]"
      }
    },
    {
      "type": "section",
      "fields": [
        {
          "type": "mrkdwn",
          "text": "*工作流:*\n${workflow_name}"
        },
        {
          "type": "mrkdwn",
          "text": "*严重度:*\n${severity}"
        },
        {
          "type": "mrkdwn",
          "text": "*错误信息:*\n${error_message}"
        },
        {
          "type": "mrkdwn",
          "text": "*时间:*\n${timestamp}"
        }
      ]
    },
    {
      "type": "actions",
      "elements": [
        {
          "type": "button",
          "text": {
            "type": "plain_text",
            "text": "查看执行详情"
          },
          "url": "${execution_url}"
        }
      ]
    }
  ]
}

[Switch: P0 或 P1]
  → [Gmail: 发送告警邮件]
      To: [ops-team@company.com]
      Subject: "[${severity}] 工作流告警: ${workflow_name}"
      Body: "
        工作流 '${workflow_name}' 执行失败。

        错误信息：${error_message}
        执行 ID：${execution_id}
        时间：${timestamp}

        请立即查看：${execution_url}
      "

[Switch: P0]
  → [HTTP Request: PagerDuty Events API v2]
      Method: POST
      URL: https://events.pagerduty.com/v2/enqueue
      Body: {
        "routing_key": "[你的 PagerDuty Integration Key]",
        "event_action": "trigger",
        "payload": {
          "summary": "工作流 '${workflow_name}' 严重故障",
          "severity": "critical",
          "source": "n8n-automation",
          "component": "${workflow_name}",
          "custom_details": {
            "error_message": "${error_message}",
            "execution_url": "${execution_url}"
          }
        }
      }

你是一个自动化运维专家。请为以下 [n8n/Make.com] 工作流设计完整的告警体系：

**工作流清单**：
[列出你的生产工作流，例如：
- AI 内容管线（每日运行）
- 线索评分（Webhook 触发）
- 数据同步（每小时运行）]

**团队规模**：[例如"3 人团队"]
**值班制度**：[例如"工作日 9-18 点有人值班"]
**现有通知渠道**：[例如"Slack、邮件、企业微信"]

请提供：
1. 告警分级标准（P0-P3）及每级的触发条件
2. 每级告警的通知渠道和响应时间要求
3. 告警去重和抑制策略
4. 每日/每周错误汇总报告的内容和格式
5. 值班升级流程

// Code 节点：LLM 输出验证
const llmOutput = $input.item.json.text || $input.item.json.output || '';
 
const validationResult = {
  raw_output: llmOutput,
  is_valid: true,
  errors: [],
  parsed_data: null
};
 
// 1. 检查是否为空
if (!llmOutput || llmOutput.trim().length === 0) {
  validationResult.is_valid = false;
  validationResult.errors.push('LLM 输出为空');
  return validationResult;
}
 
// 2. 尝试 JSON 解析（如果期望 JSON 输出）
try {
  // 尝试从 markdown 代码块中提取 JSON
  const jsonMatch = llmOutput.match(/```(?:json)?\s*([\s\S]*?)```/);
  const jsonStr = jsonMatch ? jsonMatch[1].trim() : llmOutput.trim();
  validationResult.parsed_data = JSON.parse(jsonStr);
} catch (e) {
  validationResult.is_valid = false;
  validationResult.errors.push(`JSON 解析失败: ${e.message}`);
}
 
// 3. 检查必需字段（根据你的 schema 调整）
if (validationResult.parsed_data) {
  const required = ['title', 'summary', 'category'];
  for (const field of required) {
    if (!validationResult.parsed_data[field]) {
      validationResult.is_valid = false;
      validationResult.errors.push(`缺少必需字段: ${field}`);
    }
  }
}
 
// 4. 检查拒绝回答
const refusalPatterns = [
  /I (?:cannot|can't|am unable to)/i,
  /I'm sorry,? (?:but )?I/i,
  /as an AI/i,
  /I don't have (?:access|the ability)/i
];
for (const pattern of refusalPatterns) {
  if (pattern.test(llmOutput)) {
    validationResult.is_valid = false;
    validationResult.errors.push('LLM 拒绝回答');
    break;
  }
}
 
// 5. 长度检查
if (llmOutput.length < 50) {
  validationResult.errors.push('输出过短（< 50 字符）');
}
if (llmOutput.length > 10000) {
  validationResult.errors.push('输出过长（> 10000 字符）');
}
 
return validationResult;

[LLM 调用]
  → [Code: 输出验证]
  → [IF: is_valid?]
  ├── ✅ 有效 → [后续处理]
  └── ❌ 无效 → [Code: 构建修正 prompt]
      → [LLM 调用: 重试（含修正指令）]
      → [Code: 二次验证]
      → [IF: 二次验证通过？]
      ├── ✅ 通过 → [后续处理]
      └── ❌ 失败 → [降级处理 / DLQ]

// 在 AI Agent 节点的 System Message 中添加超时约束
const systemMessage = `
你是一个高效的 AI 助手。请遵守以下约束：
1. 最多使用 5 次工具调用来完成任务
2. 如果 3 次工具调用后仍无法完成，请返回当前最佳结果并说明原因
3. 不要重复调用相同的工具（除非参数不同）
4. 如果遇到错误，最多重试 2 次，然后报告错误
`;

[AI Agent 节点]
  → [Code: 检查执行统计]
      // 检查 Agent 的工具调用次数和执行时间
      const agentOutput = $input.item.json;
      const toolCalls = agentOutput.steps?.length || 0;

      if (toolCalls > 10) {
        // Agent 可能陷入循环
        return {
          warning: 'Agent 工具调用次数过多',
          tool_calls: toolCalls,
          needs_review: true
        };
      }

[批量数据输入]
  → [Split In Batches: 每批 5 个]
  → [Wait: 1 秒（批次间延迟）]
  → [HTTP Request: API 调用]
  ├── ✅ 成功 → [收集结果]
  └── ❌ Error Output
      → [IF: 是 429 错误？]
      ├── 是 → [Code: 解析 Retry-After 头]
      │       → [Wait: 动态等待]
      │       → [HTTP Request: 重试]
      └── 否 → [其他错误处理]

// Code 节点：解析速率限制响应
const error = $input.item.json.error;
const headers = error.headers || {};
 
// 尝试从 Retry-After 头获取等待时间
let waitSeconds = 60; // 默认等待 60 秒
 
if (headers['retry-after']) {
  const retryAfter = headers['retry-after'];
  if (/^\d+$/.test(retryAfter)) {
    waitSeconds = parseInt(retryAfter);
  } else {
    // Retry-After 可能是日期格式
    const retryDate = new Date(retryAfter);
    waitSeconds = Math.max(
      Math.ceil((retryDate.getTime() - Date.now()) / 1000),
      1
    );
  }
}
 
// 添加 10% 的安全余量
waitSeconds = Math.ceil(waitSeconds * 1.1);
 
return {
  wait_seconds: waitSeconds,
  original_error: error.message
};

[Schedule Trigger: 每天 08:00]
  → [RSS Feed Read: 多源抓取]
  ├── ✅ 成功 → [Merge + Filter: 去重]
  └── ❌ Error Output → [Slack: RSS 源不可用告警] → [Stop]

  → [Loop Over Items]
      → [Code: 幂等性检查（跳过已处理文章）]
      → [IF: 已处理？]
      ├── 是 → [跳过]
      └── 否 →
          [OpenAI: 生成摘要]（Retry: 3次, 5000ms）
          ├── ✅ 成功 → [Code: 输出验证]
          │   ├── ✅ 有效 → [继续]
          │   └── ❌ 无效 → [Anthropic: 备用模型重试]
          │       ├── ✅ 成功 → [继续]
          │       └── ❌ 失败 → [Code: 文本截取降级] → [继续（标记降级）]
          └── ❌ Error Output
              → [Switch: 错误分类]
              ├── 429 → [Code: 指数退避] → [Wait] → [重试]
              ├── 401 → [Slack: 认证过期告警] → [DLQ 写入]
              └── 其他 → [Anthropic: 备用模型]
                  ├── ✅ 成功 → [继续]
                  └── ❌ 失败 → [DLQ 写入]

          → [Slack: 发送摘要]（On Error: Output Error Data）
          ├── ✅ 成功 → [继续]
          └── ❌ Error Output → [Log: Slack 发送失败] → [继续（不阻塞）]

          → [Notion: 创建页面]（On Error: Output Error Data）
          ├── ✅ 成功 → [继续]
          └── ❌ Error Output → [DLQ 写入]

          → [Code: 标记为已处理]

  → [Code: 生成执行摘要]
  → [Slack: 发送每日处理报告]

[Error Workflow: 全局告警]
  → [Error Trigger]
  → [Code: 错误分级 + 去重]
  → [Switch: 路由]
  ├── P0/P1 → [Slack: #alerts-critical] + [邮件]
  └── P2/P3 → [Slack: #alerts-warning] + [日志]

[DLQ 处理工作流: 每 30 分钟]
  → [Schedule Trigger]
  → [Postgres: 查询待重试项]
  → [Loop: 逐项重试]
  → [更新状态]