# Agent 改写客户 CRM，没有撤销键

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

原文: https://fdetimes.net/zh/guides/agent-crm-writes-approved-plans/

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

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

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

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

## 需要准备什么？

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

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

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

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

## 第 1 步：试运行只能读

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

```python
# 简化版：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。他们需要看到每条记录、每个字段、旧值、新值，以及受影响的记录总数。

```python
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 会整体停止，而不是再次调用这个工具。

```python
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 场景里，试运行和审批之间可能相隔数小时，这段时间客户的员工照常在编辑联系人。

```python
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 从没写入过一行他们事先没看过的内容。

**本周动手试试:**

- 在测试项目上运行 terraform plan -out=tfplan，再运行 terraform apply tfplan，体会规划与执行的分离
- 给手头 Agent 的任意一个写操作包一层 dry_run 函数，返回差异和哈希；执行步骤可以以后再做
- 打开客户 CRM 的 API 文档，查看是否支持幂等键；如果不支持，写下自行维护记录表的兜底方案

## 信源

- [terraform plan command reference | Terraform | HashiCorp Developer](https://developer.hashicorp.com/terraform/cli/commands/plan)

- [terraform apply command reference | Terraform | HashiCorp Developer](https://developer.hashicorp.com/terraform/cli/commands/apply)

- [Human-in-the-loop](https://docs.tracecat.com/agents/human-in-the-loop.md)

- [Interrupts](https://docs.langchain.com/oss/python/langgraph/interrupts)

- [Idempotent requests](https://docs.stripe.com/api/idempotent_requests)

- [Building effective agents](https://www.anthropic.com/engineering/building-effective-agents)