很多人讨论 ChatGPT 与 Codex，通常把重点放在模型能力上。

能不能理解复杂需求。
能不能生成完整代码。
能不能分析项目结构。
能不能调用工具。
能不能运行测试。
能不能自动修复错误。

这些能力决定了 AI 能否参与任务。

但当 AI Agent 真正进入生产工作流，开始调用接口、修改文件、更新数据库、发送消息和触发部署时，一个更底层的问题会立刻出现：

如果任务执行到一半失败了，系统应该怎么办？

重新执行一遍，看起来是最简单的答案。

但真实系统中，重复执行并不总是安全。

第一次调用可能已经成功，只是响应丢失。
第一次扣款可能已经完成，但 Agent 没有收到结果。
第一次邮件可能已经发送，只是工具超时。
第一次代码修改可能已经写入文件，但后续测试失败。
第一次创建订单可能已经落库，但 Agent 判断任务未完成。

如果系统直接重试，就可能产生：

重复扣款
重复下单
重复发信
重复写入
重复发布
状态错乱

因此，AI Agent 要从“会调用工具”升级为“可靠执行系统”，必须理解几个传统分布式系统中的核心概念：

事务
幂等
重试
补偿
回滚
状态机

这不是一个附加能力。

它决定了 AI 是否有资格修改真实世界。

---

一、语言任务没有副作用，行动任务有副作用

ChatGPT 生成一段文本时，错误成本通常比较低。

回答不好，可以重新生成。
结构不对，可以重新改写。
代码草稿不合适，可以重新调整。

因为这些输出在被正式采用之前，往往没有改变外部系统状态。

但工具调用不同。

例如：

send_email(to="user@example.com", body="...")

一旦执行，邮件就可能已经发出。

再比如：

charge_customer(
    customer_id="c_1024",
    amount=199
)

一旦成功，就产生了真实资金变化。

这类操作具有副作用。

可以把任务分成两类：

Pure Operation
无外部副作用，可安全重复

Side-Effect Operation
改变外部状态，重复执行可能产生不同结果

纯函数式操作：

def summarize(text: str) -> str:
    return model.generate(text)

有副作用操作：

def create_order(user_id: int, items: list) -> str:
    return database.insert_order(user_id, items)

对于前者，失败后重试通常安全。

对于后者，重试必须非常谨慎。

AI Agent 真正进入工程系统之后，首先要学会区分：

当前动作是在生成结果，还是在改变状态？

---

二、为什么“超时”不等于“失败”

分布式系统中，一个非常重要的认知是：

请求超时 ≠ 操作失败

假设 Agent 调用支付工具：

result = payment_tool.charge(
    order_id="order_001",
    amount=199
)

工具返回超时。

Agent 看到的是：

TimeoutError

但支付系统内部可能有三种状态：

1. 请求根本没有到达
2. 请求到达，但执行失败
3. 请求已经成功，只是响应没有返回

Agent 无法仅根据超时判断操作结果。

如果直接重试，就可能产生重复支付。

可以用状态表表示：

Agent 观察	外部系统真实状态	直接重试风险
超时	未执行	重试正确
超时	执行失败	重试可能正确
超时	已执行成功	重试可能重复执行

这说明工具执行结果不能只有：

success / failure

更合理的是：

success
confirmed_failure
unknown

程序结构可以定义为：

from enum import Enum


class ExecutionStatus(Enum):
    SUCCESS = "success"
    CONFIRMED_FAILURE = "confirmed_failure"
    UNKNOWN = "unknown"

其中 UNKNOWN 是最重要的状态。

因为它提醒 Agent：

不能猜，也不能直接重复执行。

---

三、幂等性是 Agent 重试的前提

所谓幂等，简单来说就是：

同一个操作执行一次和执行多次，最终效果相同。

数学表达可以写成：

f(f(x)) = f(x)

例如，把用户状态设置为激活：

set_user_status(user_id=1, status="active")

执行一次和执行十次，最终状态都是 active。

这是幂等操作。

但“给余额增加 100”不是幂等的：

increase_balance(user_id=1, amount=100)

执行一次增加 100。
执行两次增加 200。

所以它不是幂等操作。

在 AI Agent 里，可以为工具增加幂等属性：

from dataclasses import dataclass


@dataclass
class ToolMetadata:
    name: str
    has_side_effect: bool
    idempotent: bool
    reversible: bool
    risk_level: str

例如：

TOOLS = {
    "read_file": ToolMetadata(
        name="read_file",
        has_side_effect=False,
        idempotent=True,
        reversible=True,
        risk_level="low"
    ),
    "set_user_status": ToolMetadata(
        name="set_user_status",
        has_side_effect=True,
        idempotent=True,
        reversible=True,
        risk_level="medium"
    ),
    "send_email": ToolMetadata(
        name="send_email",
        has_side_effect=True,
        idempotent=False,
        reversible=False,
        risk_level="high"
    ),
    "charge_customer": ToolMetadata(
        name="charge_customer",
        has_side_effect=True,
        idempotent=False,
        reversible=False,
        risk_level="critical"
    )
}

有了这些元数据，Agent 才能判断失败后是否允许自动重试。

---

四、幂等键：把一次任务变成一个确定操作

对于原本不幂等的操作，可以通过幂等键降低重复执行风险。

例如创建订单时，客户端生成唯一请求标识：

create_order(
    user_id=1001,
    items=["sku-a", "sku-b"],
    idempotency_key="task-938-step-4"
)

服务端第一次收到时执行创建：

task-938-step-4 → order_7821

第二次收到同样的幂等键时，不再创建新订单，而是返回原结果：

task-938-step-4 → order_7821

服务端逻辑可以写成：

def create_order(command, idempotency_key):
    existing = idempotency_store.get(idempotency_key)

    if existing is not None:
        return existing.result

    order = order_repository.create(command)

    idempotency_store.save(
        key=idempotency_key,
        result=order
    )

    return order

AI Agent 执行每个有副作用的步骤时，都应该携带稳定的操作标识。

def build_idempotency_key(task_id: str, step_id: str) -> str:
    return f"{task_id}:{step_id}"

例如：

task-938:create-order
task-938:send-confirmation
task-938:update-inventory

关键点是：

重试时必须复用同一个幂等键，不能重新生成。

否则服务端会把重试当成新操作。

---

五、Agent 任务不是函数调用，而是长事务

传统数据库事务通常比较短。

BEGIN;

UPDATE inventory
SET stock = stock - 1
WHERE sku = 'A';

INSERT INTO orders (...);

COMMIT;

要么全部成功。
要么全部回滚。

但 AI Agent 的任务可能持续很久。

例如：

读取需求
分析代码
创建分支
修改文件
运行测试
提交代码
创建合并请求
发送通知

这些步骤跨越：

文件系统
Git 仓库
测试环境
远程平台
消息系统

无法用一个数据库事务包住。

这种任务更像长事务。

长事务的典型问题是：

步骤很多
持续时间长
跨多个系统
无法全局锁定资源
部分操作不可逆
中间状态可能被外部观察

因此，AI Agent 不能简单依赖传统 ACID 事务。

它更需要 Saga 模式。

---

六、Saga：每个正向操作都对应补偿操作

Saga 的核心思想是：

把一个大事务拆成多个局部事务。

T1 → T2 → T3 → T4

如果执行到 T3 失败，就反向执行补偿：

C2 ← C1

其中：

T1：创建临时分支
C1：删除临时分支

T2：应用代码补丁
C2：恢复原始文件

T3：运行测试
失败后触发补偿

可以定义 Saga Step：

from dataclasses import dataclass
from typing import Callable, Any


@dataclass
class SagaStep:
    name: str
    execute: Callable[[], Any]
    compensate: Callable[[Any], None] | None

执行器：

class SagaExecutor:
    def __init__(self):
        self.completed_steps = []

    def run(self, steps: list[SagaStep]):
        try:
            for step in steps:
                result = step.execute()
                self.completed_steps.append((step, result))

            return {
                "status": "completed"
            }

        except Exception as exc:
            self.compensate()

            return {
                "status": "compensated",
                "error": str(exc)
            }

    def compensate(self):
        for step, result in reversed(self.completed_steps):
            if step.compensate is not None:
                step.compensate(result)

例如 Codex 修改项目：

steps = [
    SagaStep(
        name="create_branch",
        execute=create_branch,
        compensate=delete_branch
    ),
    SagaStep(
        name="apply_patch",
        execute=apply_patch,
        compensate=revert_patch
    ),
    SagaStep(
        name="run_tests",
        execute=run_tests,
        compensate=None
    )
]

如果测试失败：

run_tests 失败
    ↓
revert_patch
    ↓
delete_branch

这就是补偿事务。

---

七、补偿不等于回滚

很多人会把补偿理解成回滚。

二者并不完全相同。

数据库回滚通常可以恢复到事务开始前的精确状态。

补偿则是通过一个新的业务动作，尽可能抵消之前操作的影响。

例如：

原操作：发送邮件

邮件无法真正撤回。

补偿可能是：

再发送一封更正邮件

再例如：

原操作：创建公开发布

补偿可能是：

下架发布内容

但用户可能已经看到了。

所以补偿只能实现业务上的修正，不一定能消除全部影响。

可以把操作分成三类：

可精确回滚
可业务补偿
不可补偿

工具定义中应该明确补偿等级：

@dataclass
class CompensationPolicy:
    rollback_type: str
    compensation_tool: str | None
    requires_human_review: bool

示例：

POLICIES = {
    "write_file": CompensationPolicy(
        rollback_type="exact",
        compensation_tool="restore_file_version",
        requires_human_review=False
    ),
    "publish_article": CompensationPolicy(
        rollback_type="business_compensation",
        compensation_tool="unpublish_article",
        requires_human_review=True
    ),
    "send_email": CompensationPolicy(
        rollback_type="not_reversible",
        compensation_tool=None,
        requires_human_review=True
    )
}

如果一个操作不可补偿，Agent 就不应该轻易自动执行。

---

八、重试策略不能只写一个循环

最危险的重试实现通常是：

while True:
    try:
        return call_tool()
    except Exception:
        continue

这种无限重试可能造成：

重复副作用
服务压力放大
错误不断传播
任务永久阻塞
外部系统被打爆

一个成熟重试策略至少要考虑：

错误是否可重试
操作是否幂等
最大重试次数
退避时间
抖动
熔断
任务截止时间

可以设计：

import random
import time


class RetryPolicy:
    def __init__(
        self,
        max_attempts: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 30.0
    ):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay

    def delay_for(self, attempt: int) -> float:
        exponential = self.base_delay * (2 ** attempt)
        jitter = random.uniform(0, 0.5)
        return min(exponential + jitter, self.max_delay)

执行逻辑：

def execute_with_retry(tool, arguments, policy):
    metadata = tool.metadata

    if metadata.has_side_effect and not metadata.idempotent:
        raise RuntimeError(
            "Non-idempotent side-effect tool cannot be retried automatically"
        )

    last_error = None

    for attempt in range(policy.max_attempts):
        try:
            return tool.call(arguments)

        except RetryableError as exc:
            last_error = exc

            if attempt == policy.max_attempts - 1:
                break

            time.sleep(policy.delay_for(attempt))

        except NonRetryableError:
            raise

    raise last_error

重试必须是策略，而不是本能。

---

九、错误分类决定恢复方式

并不是所有错误都应该重试。

可以将错误分为：

Transient Error
短暂错误，稍后可能恢复

Permanent Error
永久错误，重试不会解决

Business Error
业务条件不满足

Unknown Error
结果不确定

Policy Error
动作违反系统策略

程序结构：

class AgentExecutionError(Exception):
    pass


class TransientError(AgentExecutionError):
    pass


class PermanentError(AgentExecutionError):
    pass


class BusinessRuleError(AgentExecutionError):
    pass


class UnknownResultError(AgentExecutionError):
    pass


class PolicyViolationError(AgentExecutionError):
    pass

恢复矩阵：

错误类型	推荐处理
Transient Error	有条件重试
Permanent Error	停止并报告
Business Error	重新规划或请求输入
Unknown Error	查询状态，不直接重试
Policy Error	阻止执行

尤其是 Unknown Error。

它不应该直接进入 Retry，而应该进入 Reconcile，也就是状态对账。

---

十、Reconciliation：重新确认真实世界状态

当 Agent 不知道操作是否成功时，正确做法通常不是重试，而是查询。

例如发送订单创建请求后超时：

不要立即再次创建订单

先根据业务键查询：

order = find_order_by_request_id(
    request_id="task-938:create-order"
)

如果找到了，说明第一次已经成功。

如果明确不存在，才能考虑重试。

可以把这个流程写成：

def execute_with_reconciliation(command):
    try:
        return command.execute()

    except TimeoutError:
        status = command.reconcile()

        if status == "completed":
            return command.load_existing_result()

        if status == "not_executed":
            return command.retry()

        raise UnknownResultError(
            "Unable to determine command result"
        )

这是一种非常重要的 Agent 能力：

执行
    ↓
结果不确定
    ↓
查询真实状态
    ↓
决定接受、重试或人工处理

AI Agent 不能只依赖工具返回值。

它还需要理解外部系统的真实状态。

---

十一、状态机比自然语言记忆更可靠

复杂 Agent 任务不能只靠聊天记录判断执行到了哪里。

自然语言历史可能很长。
模型可能遗漏步骤。
重启后上下文可能丢失。
重试时可能重复已完成动作。

更可靠的方式是显式状态机。

例如代码修改任务：

CREATED
    ↓
CONTEXT_COLLECTED
    ↓
PLAN_APPROVED
    ↓
PATCH_APPLIED
    ↓
TESTS_PASSED
    ↓
REVIEW_APPROVED
    ↓
COMPLETED

失败状态：

PATCH_FAILED
TESTS_FAILED
COMPENSATING
COMPENSATED
MANUAL_INTERVENTION

程序结构：

from enum import Enum


class TaskState(Enum):
    CREATED = "created"
    CONTEXT_COLLECTED = "context_collected"
    PLAN_APPROVED = "plan_approved"
    PATCH_APPLIED = "patch_applied"
    TESTS_PASSED = "tests_passed"
    REVIEW_APPROVED = "review_approved"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"
    MANUAL_INTERVENTION = "manual_intervention"

状态转换必须受约束：

ALLOWED_TRANSITIONS = {
    TaskState.CREATED: {
        TaskState.CONTEXT_COLLECTED
    },
    TaskState.CONTEXT_COLLECTED: {
        TaskState.PLAN_APPROVED,
        TaskState.MANUAL_INTERVENTION
    },
    TaskState.PLAN_APPROVED: {
        TaskState.PATCH_APPLIED
    },
    TaskState.PATCH_APPLIED: {
        TaskState.TESTS_PASSED,
        TaskState.COMPENSATING
    },
    TaskState.TESTS_PASSED: {
        TaskState.REVIEW_APPROVED,
        TaskState.COMPENSATING
    },
    TaskState.REVIEW_APPROVED: {
        TaskState.COMPLETED
    }
}

这样 Agent 即使重启，也能从持久化状态继续，而不是重新猜测任务进度。

---

十二、Exactly Once 通常只是工程幻觉

很多人希望 Agent 的每个操作都“恰好执行一次”。

即：

Exactly Once

但在分布式系统中，这通常很难真正保证。

网络可能丢包。
响应可能超时。
服务可能崩溃。
消息可能重复投递。
Agent 可能在持久化前退出。

更现实的语义通常是：

At Most Once
最多执行一次

At Least Once
至少执行一次

At Most Once 的风险是：

可能没有执行，但不会重复

At Least Once 的风险是：

保证尝试执行，但可能重复

实际工程通常通过：

At Least Once + Idempotency

来模拟接近 Exactly Once 的业务效果。

例如消息处理器：

def handle_message(message):
    if processed_store.exists(message.id):
        return processed_store.get_result(message.id)

    result = process(message)

    processed_store.save(
        message.id,
        result
    )

    return result

Agent 设计中也应该接受一个事实：

不要假设工具只会执行一次，要假设任何步骤都可能被重复调用。

---

十三、Codex 修改代码也存在事务问题

很多人觉得事务语义只与支付、订单、数据库有关。

其实 Codex 修改代码同样存在。

假设 Codex 的计划是：

1. 修改接口定义
2. 修改 Service
3. 修改 Controller
4. 修改测试
5. 运行测试

执行到第三步时进程崩溃。

此时代码库处于中间状态：

接口已修改
Service 已修改
Controller 修改一半
测试尚未更新

如果下一次重新执行整个计划，可能重复修改，甚至产生冲突。

因此代码修改也需要事务边界。

一种方式是在临时工作区执行：

主工作区
    ↓
创建临时分支或临时目录
    ↓
执行全部修改
    ↓
运行验证
    ↓
通过后再合并

程序结构：

class CodeChangeTransaction:
    def __init__(self, repository):
        self.repository = repository
        self.workspace = None

    def begin(self):
        self.workspace = self.repository.create_isolated_workspace()

    def apply(self, patch):
        self.workspace.apply_patch(patch)

    def verify(self):
        return self.workspace.run_tests()

    def commit(self):
        self.repository.merge_workspace(self.workspace)

    def rollback(self):
        self.repository.delete_workspace(self.workspace)

执行：

transaction = CodeChangeTransaction(repository)

transaction.begin()

try:
    transaction.apply(patch)

    report = transaction.verify()

    if not report.passed:
        raise RuntimeError("Tests failed")

    transaction.commit()

except Exception:
    transaction.rollback()
    raise

这比直接在主工作区连续修改安全得多。

---

十四、补偿动作本身也可能失败

Saga 不是魔法。

补偿操作也可能失败。

例如：

已创建云资源
后续配置失败
系统尝试删除云资源
删除操作也失败

这时任务就进入异常状态。

不能简单标记为已回滚。

需要一个新的状态：

COMPENSATION_FAILED

然后交给人工或后台修复任务。

def compensate_step(step, result):
    try:
        step.compensate(result)

    except Exception as exc:
        incident_store.create({
            "step": step.name,
            "result": result,
            "error": str(exc),
            "status": "manual_recovery_required"
        })

        raise CompensationFailedError(
            step.name
        )

所以每个补偿动作也需要：

幂等性
重试策略
审计记录
失败升级

否则补偿链本身会成为新的错误来源。

---

十五、Agent 需要一个事务日志

为了在崩溃后恢复，Agent 必须持久化执行日志。

不能只把过程放在模型上下文里。

事务日志至少包含：

任务 ID
步骤 ID
操作类型
幂等键
执行前状态
执行结果
外部引用
补偿状态
时间戳

数据结构：

from dataclasses import dataclass


@dataclass
class TransactionLog:
    task_id: str
    step_id: str
    action: str
    idempotency_key: str
    status: str
    external_reference: str | None
    compensation_status: str | None
    timestamp: str

示例：

task-938 | step-1 | create_branch | success
task-938 | step-2 | apply_patch   | success
task-938 | step-3 | run_tests     | failed
task-938 | step-2 | revert_patch  | success
task-938 | step-1 | delete_branch | success

有了日志，系统才能回答：

任务执行到哪一步？
哪些动作已经完成？
哪些动作需要补偿？
哪些步骤可以安全重试？

这也是可恢复 Agent 的基础。

---

十六、事务协调器不应该完全交给模型

模型可以帮助规划事务。

但事务控制不能完全依赖自然语言推理。

例如：

是否已经执行过？
这个步骤是否幂等？
当前状态能不能进入下一状态？
是否应该补偿？

这些问题应该由确定性代码控制。

可以设计一个 Transaction Coordinator：

class TransactionCoordinator:
    def __init__(self, log_store, policy):
        self.log_store = log_store
        self.policy = policy

    def execute_step(self, task, step):
        existing = self.log_store.find(
            task_id=task.id,
            step_id=step.id
        )

        if existing and existing.status == "success":
            return existing.external_reference

        if (
            existing
            and existing.status == "unknown"
            and not step.idempotent
        ):
            return self.reconcile(step, existing)

        self.policy.validate(step)

        result = step.execute(
            idempotency_key=f"{task.id}:{step.id}"
        )

        self.log_store.save_success(
            task_id=task.id,
            step_id=step.id,
            result=result
        )

        return result

模型负责：

理解目标
生成计划
解释失败
提出补偿方案

协调器负责：

状态转换
幂等键
重试限制
日志持久化
补偿调度

生成式智能与确定性控制必须分层。

---

十七、高风险操作应该采用 Prepare-Commit 模式

对于特别高风险的动作，可以采用类似两阶段提交的思路。

第一阶段只准备，不真正提交。

Prepare

第二阶段经过验证或人工批准后再执行。

Commit

例如部署任务：

1. 生成部署计划
2. 构建发布制品
3. 检查变更范围
4. 运行预发布验证
5. 请求人工确认
6. 正式部署

程序结构：

class TwoPhaseAction:
    def prepare(self):
        raise NotImplementedError

    def commit(self, prepared_result):
        raise NotImplementedError

    def abort(self, prepared_result):
        raise NotImplementedError

部署实现：

class DeploymentAction(TwoPhaseAction):
    def prepare(self):
        artifact = build_artifact()
        report = run_staging_checks(artifact)

        return {
            "artifact": artifact,
            "report": report
        }

    def commit(self, prepared_result):
        if not prepared_result["report"].passed:
            raise RuntimeError(
                "Prepared deployment failed verification"
            )

        deploy_to_production(
            prepared_result["artifact"]
        )

    def abort(self, prepared_result):
        delete_artifact(
            prepared_result["artifact"]
        )

这种模式的核心是：

先让 Agent证明动作已经准备好，再允许产生真实副作用。

---

十八、事务语义应该进入 Agent 计划阶段

Agent 生成计划时，不应该只输出步骤名称。

每个步骤都应该声明事务属性。

例如：

{
  "step_id": "send-confirmation",
  "tool": "send_email",
  "has_side_effect": true,
  "idempotent": false,
  "reversible": false,
  "retry_policy": "manual_only",
  "compensation": null,
  "requires_approval": true
}

再比如修改文件：

{
  "step_id": "apply-patch",
  "tool": "write_file",
  "has_side_effect": true,
  "idempotent": false,
  "reversible": true,
  "retry_policy": "reconcile_then_retry",
  "compensation": "restore_file_snapshot",
  "requires_approval": false
}

可以定义计划步骤：

@dataclass
class TransactionalPlanStep:
    step_id: str
    tool_name: str
    has_side_effect: bool
    idempotent: bool
    reversible: bool
    retry_policy: str
    compensation_tool: str | None
    requires_approval: bool

这样事务治理不是执行时临时补救。

而是在计划阶段就被设计进去。

---

十九、一个可落地的 Agent 事务运行时

完整结构可以抽象为：

User Goal
    ↓
Planner
    ↓
Transactional Plan
    ↓
Policy Check
    ↓
State Machine
    ↓
Tool Execution
    ↓
Transaction Log
    ↓
Verification
    ↓
Commit / Retry / Reconcile / Compensate

简化代码：

class TransactionalAgentRuntime:
    def __init__(
        self,
        planner,
        policy,
        log_store,
        verifier
    ):
        self.planner = planner
        self.policy = policy
        self.log_store = log_store
        self.verifier = verifier

    def run(self, task):
        plan = self.planner.create_transactional_plan(task)

        completed = []

        for step in plan.steps:
            self.policy.validate(step)

            try:
                result = self.execute_step(task, step)

                verification = self.verifier.verify(
                    step,
                    result
                )

                if not verification.passed:
                    raise VerificationError(
                        verification
                    )

                completed.append((step, result))

            except UnknownResultError:
                resolved = self.reconcile(task, step)

                if not resolved:
                    return self.require_manual_intervention(
                        task,
                        step
                    )

            except Exception as exc:
                self.compensate(completed)

                return {
                    "status": "failed",
                    "error": str(exc)
                }

        return {
            "status": "completed"
        }

这个运行时和普通 Agent 最大的区别是：

它不只是考虑下一步做什么。

它还考虑：

这一步能不能重复？
失败后怎么确认状态？
已经执行的步骤如何撤销？
补偿失败怎么办？
任务重启后如何继续？

---

二十、写在最后：Agent 的可靠性，不在于永不失败，而在于失败后系统仍然一致

ChatGPT 与 Codex 让 AI 更容易理解任务、生成代码和调用工具。

但只要 AI 开始改变外部系统，错误就不再只是回答质量问题。

它会变成状态一致性问题。

一次超时，可能导致重复操作。
一次重试，可能导致重复扣款。
一次中断，可能留下半成品代码。
一次补偿失败，可能让系统停留在未知状态。
一次任务重放，可能把同一动作执行两遍。

所以，AI Agent 真正的工程门槛，不是：

能不能自动完成任务

而是：

能不能在失败、重试、中断和恢复过程中保持系统正确

这要求 Agent 理解：

事务边界
幂等键
错误分类
重试条件
状态对账
补偿动作
持久化日志
人工接管

ChatGPT 可以帮助理解复杂业务意图。

Codex 可以帮助生成代码、修改项目并运行验证。

但真正让它们成为可靠执行系统的，是确定性的事务运行时。

模型负责智能。

事务系统负责秩序。

模型可以提出下一步。
状态机决定是否允许进入下一步。

模型可以建议重试。
幂等策略决定能不能重试。

模型可以设计补偿。
事务日志决定哪些步骤需要补偿。

模型可以解释失败。
协调器负责恢复系统一致性。

未来 AI Agent 的真正成熟，不是它表现得像一个永远不会犯错的专家。

而是即使它犯错、工具超时、服务异常、进程中断，整个系统仍然知道：

发生了什么
完成了什么
没有完成什么
哪些状态需要确认
哪些动作需要补偿
什么时候必须停止

这才是工程级可靠性。

生成能力决定 Agent 能走多快。

事务语义决定 Agent 出错之后，系统会不会失控。

当 ChatGPT 与 Codex 真正进入复杂工作流后，后者可能比前者更加重要。