如何统一设计 Agent 的重试、取消、超时和成本预算？

实现与取舍

代码示例

同时限制重试次数与总 deadline

需要 Python 3.11+。只对显式暂时性异常重试；单次超时默认直接失败，取消不会被吞掉。远端副作用与计费预算另行实现。

import asyncio
class TransientError(Exception): pass
async def run(tool, max_attempts=3, total_seconds=1):
    async with asyncio.timeout(total_seconds):
        for attempt in range(1, max_attempts + 1):
            try:
                async with asyncio.timeout(0.2):
                    return await tool(attempt)
            except TransientError:
                if attempt == max_attempts: raise
                await asyncio.sleep(min(0.01 * 2**(attempt-1), 0.05))
    raise RuntimeError('unreachable')
async def tool(attempt):
    if attempt == 1: raise TransientError('temporary outage')
    return 'ok after 2 attempts'
print(asyncio.run(run(tool)))

预期输出

ok after 2 attempts

工程推演

场景

假设工程场景：研究 Agent 调用搜索 API 遇到限流，用户随后取消。

设计决策

共享截止时间和取消信号，退避可取消，已发出请求保留计费预留并核对。

验证目标

验收目标是在取消后不再发起新搜索，任务保留已完成证据和真实尝试次数。

适用边界

服务端不支持撤销时，已发出的搜索仍可能完成并计费。

易错点

• 对权限错误和取消一律重试
• 每次重试重置总 deadline
• 取消后立即认定远端操作没有发生
• 预算先检查后调用，没有预留原子性

参考资料

依据公开技术资料设计；参考资料支持技术机制，场景与评分标准为本站设计，不代表某公司面试原题。 新增问答与迁移案例用于原理讲解，来源核查与案例运行验证分别记录。

• Python：Coroutines and tasks
• Temporal Retry Policies
• Temporal：Detecting Activity failures 与 heartbeat
• AWS：Retry with backoff pattern

检查自己理解到哪一步

读完后可以对照这些标准解释原理、边界和取舍。掌握程度由你自评；需要进一步验证时，再完成下方小任务。

基础达标

能按错误类型区分可重试、永久失败与用户取消。

中高级信号

统一截止时间、退避、次数和费用预算。

资深信号

处理并发预算预留、跨层重试放大与外部动作核对。