本篇目标与场景说明

在 OpenClaw教程：从入门到实战的分层学习路线 这个系列里，本篇聚焦一个非常“实战向”的能力：多源数据聚合。当你把 OpenClaw 用在真实业务（舆情、竞品、线索、风控、知识库构建、BI 指标）时，几乎一定会遇到：数据来自多个接口（REST、GraphQL、数据库、第三方 SaaS、爬取结果、内部服务），字段名不一致、粒度不同、主键缺失、时间戳混乱、同一条记录在不同源里出现“冲突”。

本篇会围绕三件事展开：

1. 跨接口对齐：把不同源的字段、单位、时间、枚举值、ID 统一到同一套规范。
2. 合并规则：明确 join/union、去重、权重、窗口、优先级等可执行规则。
3. 冲突处理：当多个源给出不同值时，如何决策（可信度、最新优先、人工复核、保留多版本）。

文中会给出可落地的步骤、示例数据结构与规则模板，便于你直接搬到 OpenClaw 项目中。

多源聚合的“标准管线”：建议的四层结构

把多源聚合拆成四层，会显著降低复杂度：

1）采集层（Source Fetch）

• 每个数据源单独拉取，保留原始字段。
• 记录元信息：source_name、fetched_at、request_id、raw_payload_hash。

2）标准化层（Normalize / Canonicalize）

• 做字段对齐、类型转换、单位换算、枚举映射、时间统一。
• 输出统一的“规范模型”（Canonical Model）。

3）合并层（Merge / Dedup / Link）

• 对同一实体进行关联（identity resolution），可能需要模糊匹配。
• 执行 union/join 聚合，做去重与冲突决策。

4）审计与回写层（Audit / Trace / Sink）

• 输出最终结果，同时保留可追溯信息：每个字段来自哪个源、使用了哪个规则。
• 可回写到数据库、索引、对象存储或知识库。

这一套结构的核心思想是：先对齐，再合并；先保留证据，再下结论。

跨接口对齐：从字段、类型、单位、时间、枚举开始

跨接口对齐的难点不在“能不能取到数据”，而在于“取到的数据能不能用在同一张表/同一实体上”。下面给一个通用的对齐清单。

字段对齐：建立 Canonical Model（规范模型）

先定义你最终需要的统一字段。例如做“用户/客户”聚合，你可以定义：

• entity_id：内部统一 ID（聚合后生成）
• external_ids：各来源的 ID 映射
• name、email、phone
• country、city
• created_at、updated_at（统一为 UTC ISO8601）
• attributes（动态扩展字段）
• provenance（字段级溯源）

然后为每个源建立映射表：

• Source A：userId -> external_ids.sourceA，mail -> email
• Source B：id -> external_ids.sourceB，emailAddress -> email

建议把映射写成“可配置”的规则，而不是散落在代码里。这样当第三方 API 字段变更时，只改配置不改逻辑。

类型对齐：不要相信字符串

常见问题：

• 数字被当成字符串："00123"、"12.0"
• 布尔值多种形式：true/false、1/0、"Y"/"N"
• JSON 内嵌字符串：字段里塞了一个 JSON 字符串

建议在标准化层做严格转换并记录失败：

• 转换失败不要直接丢弃，保留在 attributes._parse_errors 并记录原值。

单位对齐：金额、距离、比例特别容易错

典型例子：

• 金额：源 A 给 amount_cents=1999，源 B 给 amount=19.99
• 距离：源 A 米，源 B 公里
• 比例：源 A 0.12 表示 12%，源 B 12 表示 12%

做法：

1. Canonical Model 里规定唯一单位，例如金额统一为“分”（整数）或“元”（保留两位）。
2. 每个源在标准化时换算。
3. 把换算信息写入溯源，便于审计。

时间对齐：统一时区、统一格式、明确语义

多源时间最容易出事故的点：

• 有的接口返回本地时间但不带时区。
• 有的是 Unix 秒，有的是毫秒。
• updated_at 有时代表“源系统记录更新时间”，有时是“抓取时间”。

建议：

• Canonical Model 里强制使用 UTC 的 ISO8601，例如 2026-03-19T10:21:00Z。
• 同时保留：source_event_time（源系统时间）与 fetched_at（采集时间），不要混用。

枚举对齐：先建字典，再做映射

比如状态：

• Source A：ACTIVE / SUSPENDED / DELETED
• Source B：1 / 2 / 9
• Source C：正常 / 冻结 / 注销

做法：

1. 规范枚举：active | suspended | deleted
2. 每个源映射到规范值。
3. 未知值进入 unknown 并打告警，避免静默错误。

多源合并规则：先“关联”，再“聚合”

合并不是简单的 union。你要先解决：不同来源的两条记录是不是同一个实体。

关联（Identity Resolution）：三种常见策略

1）强主键关联（最可靠）

• 有统一 ID：例如都包含同一个 crm_id。
• 直接 join：crm_id 相同即同一实体。

2）弱主键+辅助字段（常见于企业内部多系统）

例如：

• Source A 只有 email
• Source B 只有 phone
• Source C 有 name + company

可建立“候选匹配规则”优先级：

1. email 精确匹配
2. phone 精确匹配
3. name + company（标准化后）匹配

3）模糊匹配（谨慎使用）

适用于线索、舆情、商品等。

• 需要先做文本标准化：去空格、统一大小写、去特殊符号、同义词映射。
• 建议引入“匹配置信度”：例如 0~1。
• 低于阈值不自动合并，进入人工复核队列。

聚合（Merge）：定义清晰的合并策略

建议把聚合拆成“字段级”策略，而不是整条记录一刀切。常用策略：

1）优先级策略（Source Priority）

• 例如内部 CRM > 支付系统 > 第三方数据
• 规则：同字段冲突时，优先用优先级更高的源。

适用：权威字段（客户等级、合同状态）。

2）最新优先（Last Write Wins）

• 比较 source_event_time 或源的 updated_at。
• 取最新值。

适用：频繁变化字段（地址、昵称）。

3）可信度加权（Weighted Confidence）

• 给每个源一个可信度权重，例如：

  • 内部系统 0.9
  • 第三方 0.6
  • 爬取 0.4
• 同时考虑“新鲜度衰减”：越旧权重越低。

适用：指标类、画像类字段（行业、规模、职位）。

4）集合并集（Set Union）

• 对多值字段：标签、兴趣、来源渠道
• 合并为集合去重。

适用：tags、channels、aliases。

5）窗口聚合（Time Window Aggregation）

• 对事件流：订单、点击、日志
• 以窗口聚合：按天/周/月统计。

适用：BI 指标、行为分析。

冲突处理：不仅要“选一个”，还要“说清楚为什么”

冲突处理要回答两个问题：
1) 最终值是什么？
2) 为什么是它？（可追溯）

冲突分类与对应策略

类型 A：事实类字段冲突（应有唯一真值）

例：birth_date、legal_name、contract_status

推荐策略：

• 权威源优先（例如 HR/合同系统）
• 若非权威源冲突：保留多个候选 + 打上 needs_review=true
• 建立人工复核入口（至少输出冲突清单）

类型 B：时效类字段冲突（随时间变化）

例：address、title、status

推荐策略：

• 最新优先（按 source_event_time）
• 但保留历史：history.address[]（包含时间与来源）

类型 C：指标类字段冲突（口径不同）

例：revenue（是否含税）、active_users（日活定义不同）

推荐策略：

• 不要强行合并为一个数
• 在 Canonical Model 中引入口径字段：metric_name + metric_definition_id
• 或拆成多个字段：revenue_gross、revenue_net

类型 D：文本类字段冲突（可共存）

例：description、bio、product_summary

推荐策略：

• 多版本并存，排序：优先级 + 时间
• 需要“最终展示”时再选一个（展示层决策）

字段级溯源（Provenance）建议结构

最终输出里建议包含：

• provenance.field_name.source
• provenance.field_name.rule（用了哪个策略）
• provenance.field_name.source_time
• provenance.field_name.confidence

这样当业务问“为什么这个电话不是 CRM 里的那个？”你能直接回答：来自哪个源、按哪个规则覆盖、当时比较了哪些候选。

一个可直接套用的实操流程（以“客户资料聚合”为例）

下面给出从 0 到 1 的执行步骤，你可以在 OpenClaw 项目里按模块落地。

步骤 1：列出数据源与字段清单

写一个表格（建议放在仓库 docs 里），至少包含：

• 源名称（CRM、支付、工单、第三方画像）
• 获取方式（API/DB/文件）
• 主键字段（如果没有写“无”）
• 可用的关联字段（email/phone/企业名等）
• 更新频率与延迟
• 权威性（高/中/低）

步骤 2：定义 Canonical Model（强约束）

明确：

• 哪些字段必须有（如 entity_id、external_ids）
• 类型与单位
• 时间统一规则
• 枚举字典

这一步建议“宁可少，也要稳”。先把最关键的 20% 字段定义清楚，再逐步扩展。

步骤 3：为每个源实现 Normalize（可配置映射）

Normalize 输出两部分：

• canonical_fields：已对齐字段
• raw_snapshot：原始数据（用于审计/重放）

同时生成：

• quality_flags：缺失关键字段、格式异常、时间不合法等

步骤 4：实体关联（Linking）生成聚合键

建议的做法：

1. 优先用强主键：如 crm_id

2. 没有强主键则生成“候选键集合”

   • email_hash
   • phone_e164
   • name_company_key
3. 通过候选键命中已有实体则关联，否则创建新实体。

注意：模糊匹配一定要可回滚，且要记录置信度与证据。

步骤 5：字段级合并（Merge Policies）

为每个字段指定策略，例如：

• email：权威源优先（CRM）
• phone：最新优先 + 格式化为 E.164
• address：最新优先 + 保留历史
• tags：集合并集
• industry：可信度加权（第三方画像 + 内部填写）

输出时附带 provenance。

步骤 6：冲突输出与告警

至少产出两份结果：

1. customers_merged：最终聚合表
2. customers_conflicts：冲突清单（字段、候选值、来源、时间、决策结果/待复核）

当 customers_conflicts 超过阈值或出现未知枚举值时，触发告警（日志、Webhook、邮件都行）。

步骤 7：回归测试与重放机制

多源聚合最怕“规则一改全变”。建议准备：

• 固定的样本数据集（包含典型冲突）
• 每次变更规则后跑对比：输出差异报告
• 支持按 request_id 或时间范围重放（利用 raw_snapshot）

常见坑与规避建议（实战经验导向）

1）把“抓取时间”当成“业务更新时间”

规避：明确 fetched_at 与 source_event_time，合并时默认比较后者。

2）过早丢弃原始字段

规避：标准化后仍保留 raw_snapshot 或至少保留关键原字段与哈希。

3）只做记录级覆盖，不做字段级策略

规避：合并策略以字段为单位配置，必要时不同字段不同源。

4）模糊匹配没有置信度与可解释性

规避：输出匹配证据（命中规则、相似度、参与字段），低置信度进入复核。

5）指标口径混用

规避：指标必须带“定义”，不要把不同口径数字强行归一。

小结：把多源聚合做成“可配置、可追溯、可回滚”的能力

在 OpenClaw 的多源数据聚合中，真正决定质量的不是“接了多少接口”，而是：

• 是否有严格的 Canonical Model；
• 是否把对齐、合并、冲突处理做成可配置规则；
• 是否具备字段级溯源与冲突清单；
• 是否能用样本集回归与重放来控制变更风险。

下一步你可以在自己的 OpenClaw 项目里先挑一个最小场景（例如“客户基本信息聚合”或“商品信息聚合”），按本文四层结构搭出第一版管线，再逐步扩展字段与数据源。只要可追溯与可回滚打牢，规模扩张时就不会失控。