更新于

AI Agent 写操作回滚策略：让 Agent 的错误可以被撤销

AI Agent应用工程系统设计安全

给 Agent 开放”只读”权限时，最坏结果是回答错误；一旦开放”写”权限——改文件、写数据库、调用会产生副作用的 API——最坏结果就变成了不可逆的破坏。权限该给多细，《AI Agent 工具权限粒度设计》已经讨论过；本文聚焦另一个问题：权限给了之后，Agent 万一做错了，怎么撤销。

一、为什么”事前审批”不能替代”事后回滚”

很多团队的第一反应是：给危险操作加审批流，人工确认后再执行，这样不就不会出错了吗？审批流确实能拦住一部分风险，但有三个现实问题让它无法单独兜底：

  • 审批本身可能被误判：人工审批时看到的往往是 Agent 给出的操作摘要，摘要和实际执行的操作之间可能存在偏差（比如批量操作里 Agent 生成的 SQL WHERE 条件比摘要描述的范围更大）
  • 高频场景审批会拖垮体验：如果 Agent 一天要执行几百次写操作，逐条人工审批不现实，大部分团队最终会放宽审批粒度（比如”批量审批”而非逐条），这就让单条错误操作有了漏网机会
  • 审批挡不住”正确指令、错误执行”：审批的是意图，不是执行结果——Agent 理解对了、审批也通过了，但工具调用参数拼错、目标资源选错的情况依然会发生

结论是：审批流是第一道防线，回滚机制是第二道防线，两者缺一不可，不能用其中一个替代另一个。

二、回滚机制的四种设计模式

2.1 快照式回滚（Snapshot）

在 Agent 执行写操作前，先对目标资源做一次快照，操作出错后直接恢复快照。这是最直观、实现成本相对可控的方式。

import shutil
import uuid

def execute_with_snapshot(file_path, agent_write_fn):
    snapshot_id = str(uuid.uuid4())
    snapshot_path = f".snapshots/{snapshot_id}_{file_path.name}"
    shutil.copy2(file_path, snapshot_path)

    try:
        result = agent_write_fn(file_path)
        return {"status": "success", "result": result, "snapshot_id": snapshot_id}
    except Exception as e:
        shutil.copy2(snapshot_path, file_path)  # 恢复快照
        return {"status": "rolled_back", "error": str(e)}

适用场景：文件系统操作、配置变更、有明确”当前状态”概念的资源。局限：数据库这类高并发写入的资源不适合整表快照，成本太高。

2.2 补偿事务（Compensating Transaction）

数据库写操作、调用外部 API 这类场景，往往没法简单”复制粘贴”恢复原状，需要为每个写操作设计对应的”反向操作”：

原操作补偿操作
INSERT 一条记录DELETE 该记录（用返回的主键定位）
UPDATE 字段值UPDATE 回旧值（执行前记录旧值）
调用支付 API 扣款调用退款 API（如果平台支持）
发送一封邮件无法补偿——只能事前拦截，事后无法撤销
class CompensatingAction:
    def __init__(self, forward_fn, compensate_fn):
        self.forward_fn = forward_fn
        self.compensate_fn = compensate_fn
        self.executed = False
        self.forward_result = None

    def execute(self):
        self.forward_result = self.forward_fn()
        self.executed = True
        return self.forward_result

    def rollback(self):
        if self.executed:
            self.compensate_fn(self.forward_result)
            self.executed = False

# 一次 Agent 任务里的多步写操作，任意一步失败就逆序回滚已执行的步骤
actions = []
try:
    for action in planned_actions:
        action.execute()
        actions.append(action)
except Exception:
    for action in reversed(actions):
        action.rollback()
    raise

关键提醒：不是所有操作都有补偿操作——发邮件、发短信、调用不支持撤销的第三方接口，这类”不可逆操作”必须在执行前就识别出来，走更严格的审批路径，而不是寄希望于事后回滚。设计系统时先把工具分成”可补偿”和”不可补偿”两类，对不可补偿的操作单独加一层确认。

2.3 Dry Run 预演

在真正执行前，先跑一遍”只计算影响范围、不实际提交”的预演，把预期变更展示给用户或写入日志再决定是否放行：

def db_update_with_dry_run(query, params, dry_run=True):
    affected_rows = db.execute(f"SELECT COUNT(*) FROM ... WHERE {query}", params)
    if dry_run:
        return {"would_affect": affected_rows, "executed": False}

    if affected_rows > SAFETY_THRESHOLD:
        raise ValueError(f"影响行数 {affected_rows} 超过安全阈值，需要人工确认")

    return {"result": db.execute(f"UPDATE ... WHERE {query}", params), "executed": True}

Dry Run 本质上是把”回滚”提前到”执行前”——与其执行错了再撤销，不如先算出”如果执行会怎样”，用一个阈值判断是否需要额外确认。这个模式对批量操作（批量删除、批量更新）尤其有效，因为批量操作的”错误”通常表现为”影响范围远超预期”，而这正是 Dry Run 能提前发现的。

2.4 幂等性设计

回滚机制经常需要重试（回滚失败了重试回滚，或者原操作失败重试原操作），如果操作本身不是幂等的，重试反而会造成二次伤害。设计 Agent 可调用的写操作工具时，尽量保证：

def create_order_idempotent(order_data, idempotency_key):
    existing = db.query_by_idempotency_key(idempotency_key)
    if existing:
        return existing  # 已存在，直接返回，不重复创建

    return db.insert_order({**order_data, "idempotency_key": idempotency_key})

给 Agent 每次写操作生成一个唯一的 idempotency_key（可以用任务 ID + 步骤序号拼接），工具实现层用这个 key 去重。这样即使回滚逻辑本身出错导致重试，也不会因为重复执行而制造新的脏数据。

三、回滚决策：谁来判断”要不要滚”

回滚机制不是”检测到任何异常就自动回滚”，需要一个决策层区分错误类型：

错误类型处理方式
工具调用本身报错（网络超时、权限拒绝）操作大概率未生效，无需回滚，走重试逻辑
操作执行成功但结果不符合预期（如影响行数异常）需要回滚，走补偿事务
Agent 后续步骤发现之前的操作是错误决策的一部分需要回滚，且要注意回滚顺序（后执行的先回滚）
用户主动喊停立即停止后续步骤 + 回滚已执行但未确认的操作

这里最容易漏掉的是第三种——Agent 在多步任务里，第 3 步执行完才发现第 1 步的判断有误，这时候不是”重试第 3 步”，而是要把第 1、2、3 步按逆序全部回滚，再重新规划。这也是为什么补偿事务要用栈式结构（后进先出）管理已执行操作，而不是简单的列表。

四、审计日志：回滚机制的兜底

任何自动回滚机制都可能有覆盖不到的边界情况，所以审计日志不是可选项——每一次写操作，无论成功、失败还是回滚，都要留下可追溯的记录：

def log_agent_action(action_type, target, params, result, snapshot_id=None):
    audit_log.insert({
        "timestamp": now(),
        "agent_task_id": current_task.id,
        "action_type": action_type,
        "target": target,
        "params": params,
        "result": result,
        "snapshot_id": snapshot_id,
        "rollback_available": snapshot_id is not None,
    })

审计日志的价值不只是事后追责，更重要的是给人工兜底回滚提供依据——当自动回滚逻辑本身失败或覆盖不到某个边界场景时，运维人员能根据日志手动定位并恢复。日志里一定要记录 snapshot_id 或补偿操作所需的关键参数，否则”审计”只是记录了”发生了什么”，却没法支撑”怎么撤销”。

五、把回滚能力当成工具设计的必选项

实际落地时，建议把”这个工具是否可回滚、怎么回滚”作为 Agent 工具注册时的必填元数据，而不是事后补丁：

TOOL_REGISTRY = {
    "update_user_profile": {
        "handler": update_user_profile,
        "rollback_strategy": "compensating",
        "compensate_fn": revert_user_profile,
        "reversible": True,
    },
    "send_notification_email": {
        "handler": send_email,
        "rollback_strategy": "none",
        "reversible": False,
        "requires_approval": True,  # 不可逆操作强制走审批
    },
}

这样设计的好处是：工具接入阶段就强制开发者思考”这个操作出错了怎么办”，而不是等到生产环境出事故了才临时补回滚逻辑。对于确实无法设计回滚方案的操作（发消息、外部不可逆 API），至少要在元数据里显式标记，配合更严格的执行前确认。

六、相关阅读

给 Agent 开放写权限前，先把回滚路径设计好再上生产，YoTradeApi 提供稳定的模型调用中转，让你把更多精力放在 Agent 安全机制的打磨上。

相关阅读