FDE Times 数据在招 FDE 职位 826▼35 近 7 天在招职位数的变化(同口径比较):不计入本周新开始追踪的公司的职位近 7 天新增 77在招公司 154支持远程 25%美国薪资中位数 $202k招聘最多 Databricks 104
ENVI

前线部署工程师(FDE)的专业媒体

技能指南 · 指南

Agent 改写客户 CRM,没有撤销键

借鉴 Terraform 保存计划的规则,Agent 先试运行生成差异,只执行人工批准的那份计划。

分享微博

要点速览

  • Terraform 早就解决了这个问题:plan -out 把计划保存到文件,把这个文件交给 apply 就等于批准
  • 审批步骤提供三个选项:批准、修改或拒绝。拒绝后 Agent 彻底停止,不再重试
  • 执行步骤必须检查数据自试运行以来是否有变化,并发送幂等键,避免重试变成重复写入
四个相连步骤的示意图:1 试运行(只读),2 差异(一分钟内读完),3 审批(批准、修改或拒绝),4 执行(橙色高亮),旁注:哈希一致才写入。一条虚线箭头从审批步骤指回试运行,标注:修改 = 新哈希,重新检查。
只有当计划的哈希与审批人看到的版本一致时,Agent 才会写入客户的 CRM。每次修改都会生成新的哈希,必须重新批准。

在 Terraform 里,把保存好的计划文件交给 apply 就够了,因为工具把这个动作视为批准。HashiCorp 在文档里说得很直白:传入计划文件,就意味着它已经获得批准。

这只是一个小约定,却回答了每个构建 Agent 的前线部署工程师(FDE)都会遇到的问题:怎么确认实际执行的内容,正是有人审阅过的内容?

当 Agent 要写入客户的 CRM 时,这个问题最棘手。设想一次数据清理之后,Agent 要更新几百个联系人的状态。一旦写错,客户丢掉的不只是数据,还有对项目的信任。

本文介绍如何给每一次写入加上四道保护:生成计划的试运行(dry-run)、屏幕上的差异(diff)、等待人工审批,最后原样执行已批准的计划。代码用 Python 编写,做了简化。crm 是一个假想的客户端,实际使用时换成客户真实的 SDK。

需要准备什么?

需要 Python 3 和一个 CRM 沙箱,用客户的或自己的都行。想亲手试一下原始模式,还要安装 Terraform。第一个版本不要直接跑在生产数据上:Anthropic 建议在沙箱中充分测试,并为自主运行的 Agent 加上适当的护栏;Agent 拥有写权限时,这条建议更加重要。

写代码之前,先在 Terraform 里体验一下原始模式:

terraform plan -out=tfplan   # 把计划保存到文件,此时什么都不会改变
terraform apply tfplan       # 传入文件 = 批准,原样执行这份计划

不加 -out 时,Terraform 只会生成推测性计划:它描述变更效果,但不用于执行。执行已保存的计划时,也不能再附加任何其他规划模式或选项。审阅的是哪份计划,执行的就是哪份计划,写入 CRM 的 Agent 需要的正是这个特性。

第 1 步:试运行只能读

试运行读取每条记录的当前状态,与 Agent 想写入的值比较,只保留真正发生变化的字段。结果用 SHA-256 哈希,让计划有一个唯一标识。

# 简化版:crm 是一个假想的客户端
import hashlib, json

def plan_hash(items):
    body = json.dumps(items, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(body.encode()).hexdigest()

def dry_run(changes, crm):
    items = []
    for c in changes:
        current = crm.get_contact(c["id"])   # 只读,不写入
        diff = {f: [current.get(f), v] for f, v in c["fields"].items()
                if current.get(f) != v}
        if diff:
            items.append({"id": c["id"], "diff": diff})
    return {"hash": plan_hash(items), "items": items}

检查:对沙箱连续运行两次这个函数。两次的哈希必须一致,CRM 日志里也不能出现任何写请求。如果数据没变而哈希变了,常见原因是漏了 sort_keys。

第 2 步:差异要能在一分钟内读完

审阅人不会去读 JSON。他们需要看到每条记录、每个字段、旧值、新值,以及受影响的记录总数。

def render(plan):
    for item in plan["items"]:
        for field, (old, new) in item["diff"].items():
            print(f'{item["id"]}  {field}: {old!r} -> {new!r}')
    print(f'{len(plan["items"])} 条记录 | 计划 {plan["hash"][:12]}')

def save_plan(plan, path):          # 作用与 tfplan 文件相同
    with open(path, "w", encoding="utf-8") as f:
        json.dump(plan, f, ensure_ascii=False)

和 tfplan 一样,计划文件是唯一传递到审批和执行步骤的东西。检查:把 render 的输出拿给不写代码的人看,比如客户的 CRM 管理员。如果对方回答不了“会改什么?”,说明差异展示还不够好。

第 3 步:批准、修改还是拒绝?

Tracecat 为 Agent 设计的审批关卡值得参考。Agent 在需要审批的工具调用前暂停,审阅人在收件箱里选择批准、修改或拒绝。被拒绝的调用不会执行,Agent 会整体停止,而不是再次调用这个工具。

def decide(plan, path, reviewer, decision):
    assert decision in ("approve", "edit", "deny")
    approval = {"plan_hash": plan["hash"], "reviewer": reviewer,
                "decision": decision}
    with open(path + ".approval.json", "w", encoding="utf-8") as f:
        json.dump(approval, f)       # 紧挨着计划文件保存
    return approval

这个简化版遵循两条规则。“修改”指改动输入的变更列表并重新试运行,因此新计划有新哈希,必须重新批准。“拒绝”则结束这次运行:如果 Agent 换个说法再次请求批准,审阅人会疲于应付,最后为了省事直接点批准。

如果用 LangGraph 构建 Agent,这里就是放置 interrupt 的位置。LangGraph 文档把“在关键操作前暂停并请求批准”列为 interrupt 最常见的用途之一。checkpointer 会保存图的精确状态,所以审阅人几小时后才批准,运行仍能从停下的地方原样恢复。

LangGraph 还提醒了一个陷阱:interrupt 之前的代码在恢复时会再运行一遍,所以那里不能有任何非幂等操作。这就是试运行只能读的原因。哪怕它只往 CRM 写一行日志,恢复时也会再写一次。

第 4 步:原样执行已批准的计划

这一步加载计划文件及旁边的审批记录,然后按顺序做三件事:把哈希与已批准的哈希核对,检查试运行之后是否有人改过数据,最后才写入。加载时重新计算 items 的哈希,可以发现审批后被改动过的计划文件。

第二项检查是针对 CRM 场景的推理,并非 Terraform 文档的说法。HashiCorp 只建议在执行前始终审阅最终计划,确认它仍符合你的意图。而在 CRM 场景里,试运行和审批之间可能相隔数小时,这段时间客户的员工照常在编辑联系人。

def load(path):
    with open(path, encoding="utf-8") as f:
        plan = json.load(f)
    if plan_hash(plan["items"]) != plan["hash"]:
        raise RuntimeError("计划文件在保存后被修改")
    with open(path + ".approval.json", encoding="utf-8") as f:
        return plan, json.load(f)

def apply(path, crm):
    plan, approval = load(path)
    if approval["decision"] != "approve" or approval["plan_hash"] != plan["hash"]:
        raise RuntimeError("计划未获批准或已被更改")
    # 先全部检查,再写入任何一条记录
    for item in plan["items"]:
        current = crm.get_contact(item["id"])
        if any(current.get(f) != old for f, (old, _) in item["diff"].items()):
            raise RuntimeError(f'{item["id"]} 在试运行后发生变化,需要生成新计划')
    for item in plan["items"]:
        crm.update_contact(
            item["id"],
            {f: new for f, (_, new) in item["diff"].items()},
            idempotency_key=f'{plan["hash"]}:{item["id"]}',  # 假想参数
        )

幂等键借鉴自 Stripe。在 Stripe 中,连接失败时可以用同一个键重发请求,不必担心创建出第二个对象或把更新执行两次。Stripe 还会比较重试请求与原始请求的参数,不一致就返回错误,因此每个键只对应一份请求内容。

上面代码中的 idempotency_key 参数是假想的。并非所有 CRM 都支持,要查阅客户的 API 文档。如果 CRM 没有这个功能,就自己维护一张表,记录已成功执行的哈希与 id 组合,重新运行时跳过它们。

检查:在写入循环中途断网,再次运行 apply,确认没有记录被写入两次。然后在沙箱里手动修改一个联系人,再运行 apply,函数必须在写入任何内容之前停下。

常见错误

最危险的错误是批准一份计划、执行另一份:在执行步骤里,Agent 重新试运行并写入新结果。plan -out 和 apply tfplan 这对组合正是为了杜绝这种情况。第二个错误是把漂移检查和写入交错进行,等到第 150 条记录发现不一致时,前面 149 条已经写进去了。

第三个错误是差异太大。涉及几千条记录的计划,最终会在没人仔细看的情况下被批准。应拆成小批次,或者展示按字段汇总的统计,并附上几条样例记录。

这项技能在客户项目中如何体现?

第一次和客户开会之前,先准备好那些与模型无关的问题的答案:谁有权点批准,谁接收差异,写错了数据如何恢复。

带着这四个步骤去开会,就能把“我们敢不敢让 Agent 写数据?”变成一套由客户 CRM 运营团队自己掌控的流程。

简历上不要只写“构建了集成 CRM 的 Agent”。要写设计了一套“试运行、差异、审批、幂等执行”的写入流程,并给出数字,比如有多少次写入因数据漂移被拦下。

阅读 FDE 岗位描述时,留意“human-in-the-loop”“production data”“guardrails”这类说法。被问到系统设计时,可以把这套四步流程作为具体例子讲清楚。

客户会忘记你用的是哪个模型,但会记住:Agent 从没写入过一行他们事先没看过的内容。

这篇文章对你有帮助吗?

6 个信源
路线图上的下一篇 · 阶段 5: 部署上线在 Clay 做数据补全,真正的难点是写回 CRM前线部署 GTME 负责从需求调研到交付的全流程,RevOps 副总裁的一句需求背后至少藏着五个技术决策。