面向生产环境AI智能体的分层重试机制

面向生产环境AI智能体的分层重试机制：包括大语言模型调用延迟处理、工具错误恢复、检查点机制以及持久化工作流，从而避免不可靠的API和程序崩溃导致任务失败。

发布者：REACTAPP.TOP加入本站 2026年9月26日阅读约 10 分钟2377 词

来源: Building Retries in Agents: How to Build AI Agents That Survive Failures

能抵御真实故障的代理重试机制

一个拥有两个工具和循环结构的简单代理在遇到首次速率限制、不稳定的API或计划执行中途崩溃之前表现良好。而实际应用中的代理需要多层容错机制——而非仅在LLM周围使用一个try/except结构。

故障类型各不相同

在重试之前先对故障进行分类：

临时的LLM/API错误（429、503、超时）
模型可以修复的工具错误（参数错误、字段缺失）
需要重新规划的正确性错误的逻辑死路
导致内存状态丢失的进程/基础设施崩溃

对错误类型判断失误后的盲目重试只会浪费资源并加剧问题。

第一层——重新尝试LLM调用

幂等性的完成请求会采用带抖动的指数退避策略：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import anthropic

llm_retry = retry(
    stop=stop_after_attempt(2),           # 2 total attempts (1 retry)
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type((
        anthropic.RateLimitError,          # 429
        anthropic.APITimeoutError,         # Timeout
        anthropic.APIConnectionError,      # Network errors
        anthropic.InternalServerError,     # 500s (often transient)
    )),
)

@llm_retry
async def call_llm(messages, tools):
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        messages=messages,
        tools=tools,
    )
    return response

限制尝试次数；当预算用尽时抛出结构化的错误信息。

第二层——让模型处理工具错误

将工具故障作为观察结果而非严重崩溃上报，这样智能体就能修正参数：

# Inside the agentic loop
for block in response.content:
    if block.type == "tool_use":
        try:
            result = run_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result),
            })
        except Exception as e:
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": f"Error: {str(e)}",
                "is_error": True,
            })

Claude Code风格的循环会将标准错误/标准输出视为下一轮思考的上下文。

@agent.tool
async def process_document(ctx, file_path: str) -> dict:
    """Process a document through the enrichment pipeline."""
    try:
        result = await run_enrichment(file_path)

        if result.status == "error":
            raise RuntimeError(f"Processing failed: {result.error}")

        return {
            "status": "success",
            "records_processed": result.count,
            "output_path": result.output_path,
        }
    except Exception as e:
        return {
            "status": "error",
            "error": str(e),
            "traceback": traceback.format_exc(),
        }

第三层——智能体级弹性

渐进式检查点机制

保存计划步骤、工具结果以及对话片段，以便重启后能继续处理，而无需重复已完成的工作：

async def execute_with_resilience(agent, prompt, deps, max_retries=3):
    """Execute agent with checkpoint-based retry."""
    message_history = []

    for attempt in range(max_retries):
        try:
            result = await agent.run(
                prompt,
                deps=deps,
                message_history=message_history,  # Resume from last checkpoint
            )
            return result

        except (RateLimitError, APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise  # Final attempt failed

            # Checkpoint: save what we have so far
            # How you access partial messages depends on your framework.
            # PydanticAI exposes all_messages() on the result; other frameworks
            # may store it differently. The key is: grab whatever history exists.
            if hasattr(e, '_partial_messages'):
                message_history = e._partial_messages

            # Also save context that tools have accumulated
            save_checkpoint({
                "message_history": message_history,
                "context_fields": deps.get_context_snapshot(),
            })

            wait_time = min(2 ** attempt * 5, 60)
            logger.warning(
                f"Agent run failed (attempt {attempt + 1}/{max_retries}), "
                f"retrying in {wait_time}s with {len(message_history)} messages"
            )
            await asyncio.sleep(wait_time)

尤其是在工具操作成功完成后立即创建检查点。

第四层——基础设施崩溃与Temporal框架

当各步骤需要在工作节点失效后依然正常运行时，持久化工作流才能发挥优势：

# After each successful tool call, save a checkpoint
async def on_tool_complete(tool_name, result, current_messages):
    """Called after each tool completes successfully."""
    save_checkpoint({
        "message_history": current_messages,
        "last_completed_tool": tool_name,
        "context_fields": get_current_context(),
    })
    logger.info(f"Checkpoint saved after {tool_name}")

Temporal框架的适用场景

需要长时间处理的审批流程、需人工干预的场景以及耗时数小时的工具链。

其局限性

简单的聊天智能体可能无需工作流引擎，因为相应的运营成本可能会超过其带来的风险收益。

何时使用它

当需要确保副作用仅发生一次且其持续时间超过进程生命周期时，可使用Temporal（或类似工具）。对于那些具备低成本重试机制的短周期请求/响应处理程序，则无需使用该工具。

更轻量级的生产模式

许多团队会结合使用以下方法：LLM重试机制、工具错误检测功能、Redis/数据库检查点，以及能够从上一步完成状态继续运行的监督进程——从而无需在一开始就实现完整的工作流编排。

from temporalio.common import RetryPolicy

TOOL_RETRY_POLICY = RetryPolicy(
    maximum_attempts=2,
    initial_interval=timedelta(seconds=5),
    backoff_coefficient=2.0,
)

检查清单

SandboxRestrictions.default.with_passthrough_modules(
    "anthropic", "httpx", "httpcore", "os", "anyio",
    "anyio._backends", "anyio._backends._asyncio", "threading",
    "pydantic_ai", "pydantic_ai.agent",
    "pydantic_ai.durable_exec", "pydantic_ai.durable_exec.temporal",
    "pydantic_graph", "pydantic_graph.beta", "fitz",  # ... and more
)

针对临时性的LLM错误采用退避策略
将工具错误以消息形式反馈，而非直接终止进程
在修改工具后生成检查点
设置超时时间与最大步骤限制
在写入操作时使用幂等性键
监控指标：重试率、步骤延迟、放弃率

下一步

应添加混沌测试：在工具执行过程中终止进程、注入429错误响应，并验证系统的恢复行为。弹性是处理程序的核心功能，而非事后才考虑的问题。

为每次运行设定步数上限和金额上限；紧急停止按钮应位于重试旋钮旁边。

为每次运行设定步数上限和金额上限；紧急停止按钮应设置在重试旋钮旁边。

← 全部文章发送简报