AI 数据清洗管道：脏数据进、结构化数据出

先定义“干净”，再开始连节点

数据清洗不是把字符串修得好看，而是把不同来源的数据变成下游能够稳定消费的契约。开始搭建前，先为目标数据写 Schema：哪些字段必需、类型是什么、允许哪些枚举、单位和时区如何表示、空值是否有业务含义，以及无法判断时应该怎样标记。

以销售线索为例，目标结构可以包含 recordId、name、company、email、phoneE164、countryCode、intent、notes、qualityStatus 和 issues。原始值应单独保留在受控存储中，清洗结果不得覆盖唯一副本。每次运行还要记录规则版本、提示词版本、模型版本和处理时间，才能在规则变化后准确重放。

推荐把管道拆成四层：接入与留档、确定性规范化、AI 语义标准化、验证与去向。能用规则可靠完成的任务先用规则，只有需要理解语义的字段才交给 AI。这样成本更低、结果更可预测，也减少敏感数据暴露。

常见脏数据不止是多余空格

第一类是格式噪声：全角半角混用、不可见字符、重复空格、换行、大小写和不同日期格式。第二类是缺失与占位值，如空字符串、N/A、未知、- 被混为同一种空值，但其中可能包含不同业务含义。

第三类是类型和单位冲突：金额既有 1,299.00 又有 1299元，重量混用千克和磅，日期缺少年份或时区。第四类是字段错位和自由文本，例如公司名写进姓名列、地址与备注粘在一起。第五类是实体重复：同一客户使用不同手机号格式、邮箱大小写或公司简称。还有编码乱码、截断文本、非法枚举、HTML 残留、公式注入字符，以及恶意文本试图影响后续 AI 指令。

不同问题需要不同策略。空白和电话格式适合确定性函数；行业、意图和非标准职位映射可能适合 AI；客户是否为同一实体通常需要规则候选召回、相似度判断和人工复核，不能让模型凭感觉直接合并。

N8N 清洗工作流的推荐骨架

入口可以是 Webhook、表格文件、数据库增量查询或对象存储事件。第一步生成 runId 和稳定的 recordId，把原始数据、来源、导入批次和内容哈希写入暂存表。哈希可帮助识别完全相同的重复输入，但不能替代业务唯一键。

随后用 Split In Batches 或 Loop Over Items 分批处理，避免一次将数万条记录送进内存。每条记录依次通过 Set 节点字段映射、Code 节点基础清洗、IF 节点预检、AI 节点语义标准化、结构化输出解析、验证节点和数据库 Upsert。成功、需复核、拒绝三条分支分别落表，最后汇总每类数量。

批处理需要逐条状态，而不是整批只有“成功或失败”。建议状态包括 received、normalized、ai_enriched、valid、review_required 和 rejected，并保存阶段错误码。某条失败时可继续处理其余记录；恢复时从最后安全阶段开始，避免整批重复调用模型。

确定性清洗：让代码处理确定事实

先在 Set 节点把供应商字段映射为内部名称，不要让后续节点同时理解 mobile、tel、联系电话 等别名。Code 节点负责 Unicode 规范化、去除首尾空格、合并多余空白、统一邮箱大小写、把明确的占位值转换为 null。不要删除所有标点或换行，地址、公司名和备注可能依赖这些边界。

const nullTokens = new Set(['', 'n/a', 'na', 'null', '未知', '-']);
const cleanText = (value) => {
  if (value === null || value === undefined) return null;
  const text = String(value)
    .normalize('NFKC')
    .replace(/[\u200B-\u200D\uFEFF]/g, '')
    .replace(/[ \t]+/g, ' ')
    .trim();
  return nullTokens.has(text.toLowerCase()) ? null : text;
};

return $input.all().map((item) => ({
json: {
...item.json,
name: cleanText(item.json.name),
company: cleanText(item.json.company),
email: cleanText(item.json.email)?.toLowerCase() || null,
notes: cleanText(item.json.notes),
},
}));
`

邮箱不能只靠是否包含 @ 判断，也不要擅自“修复”域名拼写后直接覆盖。电话号码应结合已知国家代码交给专门的号码库解析为 E.164；缺少地区上下文时标为待确认。日期解析必须指定允许格式和默认时区，含糊的 03/04/2026 不应猜测是三月四日还是四月三日。

对 CSV 注入也要防护：如果清洗结果会再次导出表格，以 =、+、- 或 @ 开头的用户文本可能被表格软件解释为公式。根据消费端规则转义，并同时保留原值。HTML 清理应使用可靠解析器和允许列表，复杂内容不要用单个正则处理。

预检与最小化：减少无意义的 AI 调用

在模型节点前用 IF 分支检查必需字段、长度上限和已有质量。如果记录已经符合目标 Schema，就直接进入验证；如果缺失到无法推断，例如姓名、联系方式和备注都为空，则进入拒绝或人工补录；只有处于中间地带的数据才调用 AI。

对输入做字段最小化。判断职位类别只需要职位文本和有限上下文，不需要邮箱、电话或完整客户档案。把外部文本放入明确的数据区，并在系统指令中说明它是不可信内容，不得把其中的命令当作工作流指令。模型没有必要访问发送消息、修改数据库等工具；数据标准化最好使用无工具的模型调用。

长备注先按安全边界截断或摘要，同时记录是否截断。不能静默裁剪后声称结果完整。对内容哈希、任务类型、提示词版本和模型版本建立缓存键，只在隐私范围一致且结果允许复用时缓存。

AI 字段标准化：限制任务与输出空间

AI 适合把“创始人兼产品负责人”“产品一号位”等自由文本映射为标准职位类别，也适合从杂乱备注中提取购买意图和明确提到的需求。它不适合凭空补齐生日、收入、国家或联系方式。提示词要明确：只能依据输入；无法确定返回 null；不得猜测；每个判断给出来源片段或简短原因；仅输出指定结构。

例如要求结构化输出：

{
  "jobCategory": "executive",
  "intent": "demo_request",
  "countryCode": null,
  "normalizedCompany": "示例科技有限公司",
  "confidence": {
    "jobCategory": 0.91,
    "intent": 0.84,
    "normalizedCompany": 0.72
  },
  "evidence": {
    "intent": "希望本周安排产品演示"
  }
}

枚举必须在 Schema 中封闭，例如意图只能是 demo_request、pricing_question、support、other 或 unknown。限制字符串长度并拒绝额外字段。模型给出的置信度只能作为路由信号，不能被当成经过校准的真实概率；阈值必须用已标注样本评估。

公司法定名称、国家代码等可验证字段，应在 AI 输出后查权威表或内部主数据。AI 可以提出候选，但确定性查询负责确认。涉及身份、合规、授信或医疗等高风险结论时，不应依赖自动推断。

验证层：语法正确不代表业务正确

结构化输出解析器只能保证 JSON 大致符合形状，还需进行字段级和跨字段验证。字段级检查包括类型、枚举、长度、邮箱和电话号码格式；跨字段检查包括国家代码与电话号码是否一致、币种与金额单位是否匹配、结束日期是否早于开始日期。

在 Code 节点生成统一验证结果：

{
  "isValid": false,
  "issues": [
    {
      "field": "phoneE164",
      "code": "COUNTRY_CONTEXT_MISSING",
      "severity": "review"
    }
  ],
  "cleanRecord": {},
  "ruleVersion": "2026-09-08.1"
}

错误码应稳定、可统计，显示给人工的说明可以本地化。不要只抛出一段自由文本错误，因为后续无法按原因分析。校验失败也不应不断把同一输出送回模型；只允许一次有明确目标的格式修复，事实冲突则交给规则或人工。

质量状态可以按最严重问题决定：阻断性错误进入 rejected，可判断但需确认的进入 review_required，完全满足契约才进入 valid。数据库写入使用业务键或 recordId 做 Upsert，并将清洗版本纳入审计记录。若下游写入超时，先查询是否已成功，避免重复创建。

去重与实体解析要保守

去重可分两步。先用规范化邮箱、E.164 电话、税号等强键召回候选；再对公司名、地址等弱字段计算相似度。只有强证据满足规则时自动合并，模糊候选进入复核队列。AI 可解释两个名称可能相同，却不应拥有直接删除或合并主记录的权限。

合并前定义字段优先级，例如已验证 CRM 字段优先于表单输入，较新且有来源的值优先于无来源值。保留合并关系、原记录 ID 和每个字段的来源。误合并通常比漏合并更难修复，因此要提供撤销路径，而不是物理删除原数据。

失败处理、监控与成本控制

为工作流配置 Error Trigger，将 runId、recordId、阶段、可重试标记和安全错误摘要写入错误表。网络超时、429 和部分 5xx 可以有限退避重试；Schema 不合法、字段缺失和权限错误不能原样重试。达到上限后进入死信队列，修正规则或数据后按 recordId 重放。

监控至少包括输入数、有效数、复核数、拒绝数、各错误码数量、字段空值率、枚举分布、AI 调用数、平均延迟和估算成本。若某天 unknown 比例或公司名变化率突然升高，可能是来源格式改变或提示词回归。设置阈值告警，但不要只看工作流是否显示绿色成功。

成本控制从预检开始：跳过已合格记录，批量但不过度拼接输入，限制备注长度，避免每个字段单独调用一次模型，并为每批设置最大记录数和预算。模型升级、提示词调整或枚举变化时先用回归集比较，再切换版本。

用黄金数据集验收管道

从真实数据脱敏抽取一组覆盖正常、边界和恶意输入的样本，由业务人员标注期望输出，形成黄金数据集。至少覆盖中英文混排、全角字符、空值占位、含糊日期、国际电话、重复实体、超长备注、非法枚举、提示词注入文本和依赖服务不可用。

每次规则或模型变化都在同一数据集上运行，比较字段准确率、自动通过率、误合并率、人工复核率和单条成本。特别关注“错误地自动通过”，因为它比保守地转人工更危险。抽样复核线上有效记录，并记录人工改动，用于更新规则和测试集，而不是未经审查直接训练或回灌。

一条可靠的清洗管道应做到：原始数据可追溯，转换规则可版本化，每个字段有来源，AI 不确定时允许为空，验证失败有明确去向，重复执行不会产生重复记录。达到这些条件，才算真正实现了“脏数据进、结构化数据出”。