n8n + Claude 数据管道自愈：PostgreSQL 异常检测与自动修复

定时拉取 PostgreSQL 数据 → 统计/ML 异常检测 → Claude 分类与修复规划 → 自动执行修复 → 验证并通知

给谁

数据工程师、运维团队、需要自动化数据质量监控的组织

你会得到

每 15 分钟自动检测数据异常，AI 分类并自动修复（填充缺失值、截断异常值、修正格式、重试上游），失败时人工升级

你需要

n8n · postgresql · slack · pagerduty · datadog

步骤

1. 配置触发器与数据连接

   在 n8n 中导入模板后，先配置定时触发器（默认每 15 分钟）和 Set Pipeline Parameters 节点。填写 PostgreSQL 连接信息、源表名、Slack 频道 ID、PagerDuty 服务 ID 与 API 密钥、DataDog API 密钥。修改节点中的占位符：<__PLACEHOLDER_VALUE__source_table_name__>、<__PLACEHOLDER_VALUE__Slack channel ID for notifications__>、<__PLACEHOLDER_VALUE__PagerDuty API key__>、<__PLACEHOLDER_VALUE__DataDog API key__>。

2. 运行统计与 ML 异常检测

   工作流自动执行：Fetch Latest Data from PostgreSQL 拉取最近 15 分钟数据；Fetch Historical Baseline Data 加载 7 天基线；Compare with Historical Baseline 计算偏差；Calculate Trend Analysis Metrics 计算移动平均与趋势；Run ML-Based Anomaly Scoring 综合 5 项特征（基线偏差、趋势速度、历史模式、统计异常值、空值率）输出 0-1 异常分数。Run Statistical Quality Checks 执行空值率、IQR 异常值、 schema 一致性检查。

3. Claude 分类异常并规划修复

   若异常分数超过阈值（默认 0.75）或统计检查发现异常，Check If Anomaly Detected 分支进入 AI Anomaly Classifier Agent。该节点调用 Claude-3-5-sonnet，system prompt 设定为数据质量专家，user message 传入异常类型、受影响列、统计摘要，要求输出 JSON 格式的分类（schema_drift/missing_data/outliers/upstream_failure）、严重度（low/medium/high/critical）、置信度、解释。随后 AI Remediation Planner Agent 再次调用 Claude，基于分类结果生成单一修复动作（fill_nulls/cap_outliers/reformat_schema/retry_upstream）及参数、成功概率、推理。

   运行提示词 · 来源原文。AI Anomaly Classifier Agent 节点的 options.system 字段

   复制提示词

   You are a data quality expert specializing in anomaly detection and classification. Analyze data quality issues and provide accurate classifications with severity assessments.

   运行提示词 · 来源原文。AI Anomaly Classifier Agent 节点的 messages.values[0].content 字段

   复制提示词

   =Analyze this data quality anomaly and classify it:
   
   Anomaly Type: {{ $('Run Statistical Quality Checks').first().json.anomalyType }}
   Affected Columns: {{ $('Run Statistical Quality Checks').first().json.affectedColumns }}
   Statistics: {{ JSON.stringify($('Run Statistical Quality Checks').first().json.statistics) }}
   Total Records: {{ $('Run Statistical Quality Checks').first().json.totalRecords }}
   
   Classify this anomaly into one of these categories:
   - schema_drift: Unexpected changes in data structure or types
   - missing_data: High rates of null or missing values
   - outliers: Statistical anomalies in numeric data
   - upstream_failure: Issues with data source or pipeline
   
   Provide your response as JSON with:
   {
     "classification": "category_name",
     "severity": "low|medium|high|critical",
     "confidence": 0-100,
     "explanation": "detailed reasoning"
   }

   运行提示词 · 来源原文。AI Remediation Planner Agent 节点的 options.system 字段

   复制提示词

   You are a data engineering expert specializing in automated remediation strategies. Suggest practical, safe remediation actions for data quality issues based on anomaly classification and historical patterns.

   运行提示词 · 来源原文。AI Remediation Planner Agent 节点的 messages.values[0].content 字段

   复制提示词

   =Based on this classified anomaly, suggest a specific remediation action:
   
   Classification: {{ $json.classification }}
   Severity: {{ $json.severity }}
   Anomaly Details: {{ $json.explanation }}
   Affected Columns: {{ $('Run Statistical Quality Checks').first().json.affectedColumns }}
   Statistics: {{ JSON.stringify($('Run Statistical Quality Checks').first().json.statistics) }}
   
   Suggest ONE of these remediation actions:
   - fill_nulls: Fill missing values with appropriate defaults
   - cap_outliers: Cap extreme values to acceptable ranges
   - reformat_schema: Fix schema mismatches or type issues
   - retry_upstream: Retry fetching data from source
   
   Provide:
   1. Action (one of the above)
   2. Parameters needed for the action
   3. Expected success probability (0-100)
   4. Reasoning
   
   Return as JSON: {"action": "...", "parameters": {...}, "successProbability": ..., "reasoning": "..."}

4. 执行修复动作

   Query Remediation History from PostgreSQL 查询近 30 天同类异常的修复历史；Calculate Remediation Success Rate 计算各动作的历史成功率并附到修复计划。Route by Remediation Type 根据 action 字段路由到四个分支之一：Remediate Missing Values（数值列用中位数、类别列用众数填充）、Remediate Outliers（IQR 上下界截断）、Remediate Schema Format（按预期 schema 转换类型）、Retry Upstream Data Fetch（占位符，需自行实现实际重试逻辑）。严重度为 critical 时，Check If Critical Severity 并行触发 Post Critical Alert to PagerDuty 创建 PagerDuty 事件。

5. 验证修复结果并通知

   Merge Remediated Data Streams 合并各分支结果；Re-run Quality Checks on Remediated Data 重新运行空值、异常值、schema 检查。若验证通过（remediationSuccessful=true），Send Success Notification to Slack 发送成功通知；否则 Escalate to Human Review in Slack 发送含数据样本、统计信息的人工复核请求，并触发 Prepare Data Sample for Review 生成代表性样本。

6. 记录事件并更新监控

   Log Successful/Failed Remediation Incident 将结果写入 PostgreSQL 事件表；Aggregate Incident Logs 聚合；Generate Remediation Report 生成含成功率、按类型/严重度/动作 breakdown、top issues、建议的综合报告；Send Metrics to DataDog 推送异常计数、修复成功率、健康分数；Update Historical Baseline Metrics 更新基线表。

请根据以下 n8n 模板搭建数据管道自愈工作流。先确认你拥有：n8n 实例、PostgreSQL（源表+事件表+基线表）、Anthropic Claude API 密钥、Slack 频道权限、PagerDuty 服务 ID 与 API 密钥、DataDog API 密钥。导入模板后，按步骤替换所有占位符，测试单个节点执行，再启用定时触发器。