埋点和指标需求承接方最头疼的事，从来不是写 SQL 本身——而是把散落在各处的信息重新拼起来。需求文档里写的动作到底要不要采集？历史上有没有类似点位？指标口径是否还在被下游使用？新增字段要改哪几层表？发布前该由谁确认？这些问题的答案分布在文档、元数据平台、群聊记录和历史工单里，靠人肉拼图既慢又容易漏。得物数仓团队用 Hermes Agent 重构了这条工作流，把反复出现的判断逻辑沉淀成规则资产，让后续需求可以自动校验和复用。
埋点需求承接的碎片化困局
一个典型埋点需求从提出到上线，至少要经过四个环节的"拼图"：
- 需求解析——产品文档里列了十几个动作，哪些真正需要采集、哪些只是描述性文字，需要逐条判断。
- 历史查重——同一个动作可能去年已经埋过，只是命名不同；不做查重就会产生重复点位，下游指标口径分裂。
- 影响面评估——新增字段要改 ODS、DWD、DWS 多层表，还要确认下游报表和看板是否依赖旧字段。
- 发布确认——改动涉及哪些团队、谁负责最终审批，流程靠人记忆和口头确认。
每个环节都在不同系统里找信息，做完之后判断逻辑留在人的脑子里，下次遇到类似需求又得从头来一遍。这就是"碎片化"的核心：信息分散、判断不可复用、流程不可追溯。
为什么选 Hermes Agent 而不是 OpenClaw
团队在选型时对比了 OpenClaw 和 Hermes Agent，最终选择后者的关键差异在于三点：
- 持续在线——Agent 不是一次性执行完就退出，而是长驻运行，可以随时接收新需求、触发校验流程。
- 持久记忆——每次处理需求时积累的判断结果、历史查重结论、字段映射关系都会写入记忆库，后续需求可以直接检索复用，而不是重新推理。
- 技能沉淀——反复执行的判断逻辑（比如"某个动作是否需要采集"的判定规则）可以被抽象成技能，下次遇到同类需求时自动调用，不再依赖人工逐条分析。
OpenClaw 更偏向一次性任务编排，缺乏长驻记忆和技能自动沉淀机制，在需要持续积累规则的数仓场景里不够贴合。
从需求到规则：Agent 工作流拆解
Hermes Agent 处理一个埋点需求的核心流程可以拆成以下步骤：
需求文档 → 解析动作列表 → 历史查重（检索记忆库）
↓
未命中 → 新建点位规则 → 影响面评估（查元数据）
已命中 → 复用已有规则 → 校验口径一致性
↓
生成改动清单（涉及表、字段、下游依赖）
↓
推送确认任务给相关团队
关键在于"历史查重"和"规则沉淀"这两个环节：查重不是简单的关键词匹配，而是基于语义和上下文的相似度判断；沉淀也不是存一条记录，而是把判断逻辑本身变成可调用的技能。
实践：用配置定义一个埋点查重技能
下面是一个简化版的 Hermes Agent 技能配置示例，展示如何把"埋点查重"的逻辑定义成可复用的规则资产。实际项目中可以根据团队元数据平台的具体接口做适配。
# hermes_skills.yaml — 埋点查重技能定义
skill:
name: tracking_point_dedup
description: "对需求文档中的动作列表做历史查重，返回已有点位或建议新建"
trigger:
event: new_tracking_requirement
# 当新需求进入系统时自动触发
steps:
- name: parse_actions
type: llm_extract
prompt: |
从以下需求文档中提取所有需要采集的用户动作，
输出 JSON 数组，每个元素包含 action_name、page、trigger_condition。
不要包含纯描述性文字，只提取可采集的动作。
input_key: requirement_doc
output_key: action_list
- name: dedup_check
type: memory_search
# 在持久记忆库中检索相似点位
search_type: semantic
similarity_threshold: 0.85
input_key: action_list
output_key: dedup_result
# dedup_result 结构：
# { matched: [{new_action, existing_point, similarity}], unmatched: [...] }
- name: impact_assessment
type: metadata_query
# 对未匹配的动作，评估新增字段的影响面
query_template: |
SELECT table_name, layer, downstream_count
FROM metadata.field_dependencies
WHERE field_name IN ({new_fields})
input_key: dedup_result.unmatched
output_key: impact_report
- name: generate_checklist
type: template_render
template: |
## 埋点改动清单
**复用点位：**
{% for m in dedup_result.matched %}
- {{ m.new_action }} → 已有点位 {{ m.existing_point }}（相似度 {{ m.similarity }}）
{% endfor %}
**新增点位：**
{% for u in dedup_result.unmatched %}
- {{ u.action_name }}：需新增，影响表 {{ u.affected_tables }}
{% endfor %}
input_keys: [dedup_result, impact_report]
output_key: change_checklist
- name: notify_teams
type: message_push
channels: [dingtalk, jira_comment]
input_key: change_checklist
上面的配置定义了完整的查重流程：从需求文档提取动作、在记忆库做语义查重、查元数据评估影响面、生成改动清单并推送给相关团队。每次执行后，匹配结果和新增点位的判定逻辑都会写入记忆库，下次遇到相似需求时 dedup_check 步骤可以直接命中。
如果团队暂时没有 Hermes Agent 的完整部署环境，也可以用 Python 脚本模拟核心查重逻辑，先跑通最小闭环：
"""最小化埋点查重脚本 — 先跑通语义匹配闭环"""
import json
from pathlib import Path
# 假设已有历史点位库（实际可从元数据平台拉取）
MEMORY_FILE = Path("tracking_memory.json")
def load_memory():
if MEMORY_FILE.exists():
return json.loads(MEMORY_FILE.read_text())
return []
def save_memory(memory):
MEMORY_FILE.write_text(json.dumps(memory, ensure_ascii=False, indent=2))
def dedup_check(new_actions: list[dict], memory: list[dict], threshold: float = 0.85):
"""简易语义查重：基于 action_name + page 的文本相似度
生产环境应替换为向量检索（如 Milvus / pgvector）"""
matched, unmatched = [], []
for action in new_actions:
best_match = None
best_score = 0.0
for existing in memory:
# 简化相似度：名称完全匹配或包含关系
score = compute_simple_similarity(
action["action_name"], existing["action_name"]
)
if score > best_score:
best_score = score
best_match = existing
if best_score >= threshold and best_match:
matched.append({
"new_action": action,
"existing_point": best_match,
"similarity": best_score,
})
else:
unmatched.append(action)
return matched, unmatched
def compute_simple_similarity(a: str, b: str) -> float:
"""生产环境请替换为 embedding cosine similarity"""
a_lower, b_lower = a.lower(), b.lower()
if a_lower == b_lower:
return 1.0
if a_lower in b_lower or b_lower in a_lower:
return 0.9
overlap = len(set(a_lower) & set(b_lower))
total = len(set(a_lower) | set(b_lower)) or 1
return overlap / total
# --- 使用示例 ---
new_requirement_actions = [
{"action_name": "点击购买按钮", "page": "商品详情页", "trigger_condition": "tap"},
{"action_name": "加入购物车", "page": "商品详情页", "trigger_condition": "tap"},
{"action_name": "分享商品链接", "page": "商品详情页", "trigger_condition": "tap"},
]
memory = load_memory()
matched, unmatched = dedup_check(new_requirement_actions, memory)
print(f"复用点位：{len(matched)} 个")
for m in matched:
print(f" {m['new_action']['action_name']} → {m['existing_point']['action_name']} (相似度 {m['similarity']})")
print(f"新增点位：{len(unmatched)} 个")
for u in unmatched:
print(f" {u['action_name']} @ {u['page']}")
# 将新增点位写入记忆库，下次可复用
for u in unmatched:
memory.append(u)
save_memory(memory)
运行前把 tracking_memory.json 放在同目录下，初始内容可以是空数组 [] 或从现有元数据导出的点位列表。脚本跑通后，下一步就是把 compute_simple_similarity 替换为真正的向量相似度检索，再逐步接入元数据平台的影响面查询。
规则资产的长期价值与边界
把判断逻辑沉淀成规则资产，带来的不只是效率提升，还有几个容易被忽略的长期收益：
- 口径一致性——当查重命中已有点位时，下游指标天然沿用同一口径，避免了"同一个动作两个名字、两个指标"的分裂问题。
- 流程可追溯——每个需求的查重结论、影响面评估、确认记录都留在 Agent 记忆库中，出问题时可以回溯，而不是翻群聊记录。
- 新人上手加速——规则资产本身就是文档，新同事不需要靠"问老员工"来了解哪些动作已有点位、哪些字段有下游依赖。
但也有边界需要正视：
- 语义查重的准确率依赖 embedding 质量——动作描述越模糊，误匹配风险越高，需要人工审核兜底。
- 规则资产需要定期清理——已废弃的点位、不再使用的指标口径如果不从记忆库中标记失效，查重结果会越来越噪音。
- Agent 不能替代业务判断——"要不要采集"的最终决策仍然需要产品和数据团队确认，Agent 只是把信息拼齐、把重复判断自动化。
上手建议
如果你所在团队也在埋点需求承接上反复拼图，可以按这个顺序推进：
- 先用脚本跑通查重闭环（如上面的 Python 示例），验证语义匹配的准确率。
- 把查重结果和判断逻辑持久化，形成第一版记忆库。
- 引入 Agent 框架（Hermes 或同类方案），把脚本逻辑迁移为技能配置，接入元数据平台和消息推送。
- 设置人工审核节点——查重命中时自动复用，未命中时生成清单推送确认，避免 Agent 全自动带来的误判风险。
- 每季度清理记忆库中的废弃点位和过期规则，保持资产质量。
从拼图到规则，核心转变不是工具升级，而是把"留在人脑子里的判断"变成"可检索、可复用、可追溯的资产"。Hermes Agent 提供了持续在线和技能沉淀的机制，但真正让规则资产发挥价值的是团队愿意把判断逻辑写下来、持续维护它。