超时、重试与补偿：先确认一次操作的结果

更新 2026-10-05

我的学习 →

正在读取学习记录…

1. 超时发生在等待的一端

超时表示调用方等到时间上限后仍未取得完整响应。读取资料时，可以有限重试；提交写入时，需要考虑另一种情况：远端已经完成，只是回来的响应丢了。

像网购付款后页面一直转圈：你还没看到支付成功页，但账户可能已经扣款。此时查询订单状态能帮助确认结果。

请求到达远端并完成写入，返回响应丢失；相同操作键再次查询时取回原回执 — 写入发生在响应之前。调用方看到未知结果时，用原操作键回查。

重试是再次尝试请求；结果确认是查询前次操作是否完成；补偿是用新的业务动作修正已发生的效果。例如取消已创建的预订是补偿，回查预订编号是结果确认。

2. 观察响应丢失后，再次提交的结果

在学习文件夹新建 timeout_lab.py，保存下面完整程序。从该文件夹打开 PowerShell，运行代码后的命令。

先在 PowerShell 创建本篇的独立目录并进入；路径可换成你自己的非系统盘位置：

mkdir -Force "F:\学习\agentdepth-l2-timeout"
cd "F:\学习\agentdepth-l2-timeout"

将本篇的 timeout_lab.py、test_timeout_lab.py 保存到这个目录。保存全部文件后运行 dir *.py，核对上述文件名，再执行下文命令；新开窗口时先回到该目录，并按 Python 环境篇核对 python 的解释器路径。

"""观察响应丢失后的结果确认与重复提交。"""
from __future__ import annotations


class Remote:
    def __init__(self) -> None:
        self.applied = 0
        self.records: dict[str, tuple[str, str]] = {}

    def submit(self, key: str, payload: str, lose_response: bool = False) -> dict:
        if key in self.records:
            old_payload, receipt = self.records[key]
            if old_payload != payload:
                return {"status": "conflict"}
            return {"status": "succeeded", "receipt": receipt, "replayed": True}
        self.applied += 1
        receipt = f"receipt-{self.applied}"
        self.records[key] = (payload, receipt)
        if lose_response:
            return {"status": "unknown", "reason": "response_lost"}
        return {"status": "succeeded", "receipt": receipt, "replayed": False}


# 第一次写入后丢失响应，再用相同操作键取回已保存回执。
def demo() -> dict:
    remote = Remote()
    first = remote.submit("op-1", "create-tag:todo", lose_response=True)
    repeated = remote.submit("op-1", "create-tag:todo")
    conflict = remote.submit("op-1", "create-tag:urgent")
    return {"first": first, "repeated": repeated, "conflict": conflict, "effects": remote.applied}


if __name__ == "__main__":
    import json
    print(json.dumps(demo(), ensure_ascii=False, indent=2))

python timeout_lab.py

输出：

{
  "first": {
    "status": "unknown",
    "reason": "response_lost"
  },
  "repeated": {
    "status": "succeeded",
    "receipt": "receipt-1",
    "replayed": true
  },
  "conflict": {
    "status": "conflict"
  },
  "effects": 1
}

第一次返回 unknown，但 effects 已增加。第二次同键同内容取回 receipt-1，replayed: true 表示返回保存的结果。第三次同键换内容返回 conflict。最后效果计数为1。

records 把操作键连接到原内容和回执。submit 先检查旧记录，同内容返回回执，不同内容冲突；新键先记录效果和回执，再按 lose_response 设置返回未知。这个开关固定制造了“写入完成但响应丢失”的窗口，便于每次运行观察同一条路径。

3. 换操作键会怎样

把第二次提交中的 "op-1" 改为 "op-2"，运行后效果计数变2。因为服务把它视为新操作。恢复键后，计数回到1。

再将首次 lose_response 改为 False，第一条直接成功，后面的同键请求仍回放原回执。这说明结果去重与响应是否丢失分别处理。

4. 运行检查用例

把下方测试文件保存到程序所在目录：

下面的测试检查响应丢失后，用同一业务键重发能否取得原结果，并区分参数错误与可继续处理的故障。

测试代码：test_timeout_lab.py

# 核对超时、重试与补偿的正常结果和错误分支；断言失败时会显示对应的测试。
import unittest
from timeout_lab import Remote, demo


class TimeoutLabTests(unittest.TestCase):
    def test_response_loss_is_unknown_to_client(self):
        remote = Remote()
        result = remote.submit("k1", "x", lose_response=True)
        self.assertEqual(result["status"], "unknown")
        self.assertEqual(remote.applied, 1)

    def test_same_key_replay_returns_receipt(self):
        result = demo()
        self.assertEqual(result["repeated"]["receipt"], "receipt-1")
        self.assertEqual(result["effects"], 1)

    def test_changed_payload_is_conflict(self):
        self.assertEqual(demo()["conflict"]["status"], "conflict")


if __name__ == "__main__":
    unittest.main()

python -m unittest test_timeout_lab -v

测试同时检查正常和失败路径。先恢复刚才的改动，再运行；若出现导入错误，核对文件位置与运行目录。

5. 把重试与补偿安排成明确流程

观察	优先处理	原因
读取超时	有限重试并记录次数	读取通常容易重做
提交前校验失败	修正输入	请求尚未合法进入执行
写入后响应未知	用原操作编号查询结果	远端可能已完成
结果确认成功	返回已保存回执	保留同一次操作身份
业务效果需要撤销	发起明确的补偿操作	撤销有独立语义和权限

重试间隔可以逐步增大，并加少量随机抖动，减少多客户端一起重发。总次数和总时间都要限制，最终保留未确认状态。一次任务的时间预算包含初次等待和每次重试等待。

补偿能撤销的程度取决于业务：预订可以取消，已发送邮件只能再发更正说明。补偿也可能失败，所以保存补偿请求、结果和人工处理入口。对同一资源操作时，还应检查它是否已经被其他任务改变。

真实系统可将操作记录与业务变化在同一事务内提交，并提供回执查询接口。可进一步阅读 AWS 重试与幂等实践。总结：等待超时、结果未知、确认成功和业务撤销分别表示，下一步根据已知事实决定。继续读 MCP架构。

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“工具超时、重试与补偿”，哪种做法更稳妥？

对应知识点：L2-timeout

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L2-timeout

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L2-timeout

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L2-timeout

你的回答