Agent 改写客户 CRM,没有撤销键
借鉴 Terraform 保存计划的规则,Agent 先试运行生成差异,只执行人工批准的那份计划。
要点速览
- Terraform 早就解决了这个问题:plan -out 把计划保存到文件,把这个文件交给 apply 就等于批准
- 审批步骤提供三个选项:批准、修改或拒绝。拒绝后 Agent 彻底停止,不再重试
- 执行步骤必须检查数据自试运行以来是否有变化,并发送幂等键,避免重试变成重复写入
在 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 从没写入过一行他们事先没看过的内容。
这篇文章对你有帮助吗?
感谢你的反馈!