大模型Agent的系统可靠性：从单次调用到多步推理的端到端保障

一个 LLM 调用的成功率是 99.9%，看起来不错。但一个 Agent 任务包含 8 步推理，每一步都调用 LLM——整个任务的成功率就变成了 0.999^8 ≈ 99.2%。这还没算工具调用失败、输出格式错误和其他非 LLM 故障。Agent 的系统可靠性是一个「乘法定律」问题。

一、Agent 任务的「可靠性乘法灾难」

传统 API 调用链中，可靠性保障的核心手段是重试和降级。但这些手段在 Agent 场景下并不直接适用：

graph LR
    subgraph Chain["Agent 多步推理链（8 步）"]
        S1["Step 1\n工具选择\n99.9%"] --> S2["Step 2\n代码生成\n99.9%"]
        S2 --> S3["Step 3\n代码执行\n99.5%"]
        S3 --> S4["Step 4\n结果分析\n99.9%"]
        S4 --> S5["Step 5\n信息检索\n99.9%"]
        S5 --> S6["Step 6\n数据整合\n99.9%"]
        S6 --> S7["Step 7\n结论生成\n99.9%"]
        S7 --> S8["Step 8\n格式化输出\n99.9%"]
    end

    Chain --> Result["端到端成功率\n0.999^6 × 0.995 = 98.6%"]

    subgraph Problems["Agent 特有的可靠性挑战"]
        P1["部分成功：步骤 1~3 正确但步骤 4 失败\n—— 前面消耗的 Token 成本能否挽回？"]
        P2["级联失败：步骤 2 的一个小错误导致步骤 3~8 全错\n—— 错误在 8 步中被放大 7 倍"]
        P3["状态丢失：进程崩溃后，已完成的 5 步如何恢复？\n—— 从头再来成本巨大"]
    end

核心问题是：Agent 的每一步都可能失败，而且失败的成本是累积的——前 N-1 步消耗的 Token、调用的工具、占用的时间，在最后一步失败时会全部浪费。

二、任务状态的持久化与断点恢复

解决状态丢失的第一原则：每一步完成后立即持久化，永远不依赖内存状态。

package com.example.agent.reliability;

import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Agent 任务状态管理器。
 * 核心设计：每一步执行前先持久化状态，
 * 即使进程崩溃，重启后可以从最近完成的步骤继续执行。
 */
public class AgentTaskStateManager {

    // 任务状态存储 —— 生产环境使用 Redis 或数据库
    // 每个 step 的执行结果以独立 Key 存储，支持部分读取
    private final Map<String, AgentTask> taskStore = new ConcurrentHashMap<>();
    private final Map<String, List<StepExecution>> stepStore = new ConcurrentHashMap<>();

    /**
     * 执行一个 Agent 任务，带断点恢复能力。
     *
     * @param taskId   任务唯一 ID
     * @param plan     执行计划（步骤列表）
     * @param executor 步骤执行器
     * @return 任务最终结果
     */
    public TaskResult executeWithRecovery(
            String taskId,
            List<AgentStep> plan,
            StepExecutor executor) {

        // 1. 检查是否是恢复执行（之前被中断的任务）
        AgentTask task = taskStore.get(taskId);
        if (task == null) {
            // 新任务：持久化任务初始状态
            task = new AgentTask(taskId, AgentTaskStatus.RUNNING, plan, Instant.now());
            taskStore.put(taskId, task);
        } else if (task.status() == AgentTaskStatus.COMPLETED) {
            // 已完成的任务，直接返回结果
            return new TaskResult(taskId, AgentTaskStatus.COMPLETED, task.finalOutput());
        }

        // 2. 找出断点 —— 从第一个未成功执行的步骤开始
        List<StepExecution> history = stepStore.getOrDefault(taskId, new ArrayList<>());
        Set<Integer> completedStepIndexes = new HashSet<>();
        for (StepExecution se : history) {
            if (se.status() == StepStatus.SUCCESS) {
                completedStepIndexes.add(se.stepIndex());
            }
        }

        // 3. 从断点继续执行
        StringBuilder accumulatedContext = new StringBuilder();
        for (int i = 0; i < plan.size(); i++) {
            if (completedStepIndexes.contains(i)) {
                // 已完成的步骤：恢复上下文但跳过执行
                StepExecution prev = getStepExecution(taskId, i);
                accumulatedContext.append(prev.output()).append("\n");
                continue;
            }

            AgentStep step = plan.get(i);

            // 持久化「即将执行」的状态
            StepExecution execution = new StepExecution(
                taskId, i, step.type(), StepStatus.RUNNING,
                null, Instant.now()
            );
            persistStepExecution(taskId, execution);

            try {
                // 执行当前步骤，传入之前所有步骤的上下文
                String stepOutput = executor.execute(
                    step,
                    accumulatedContext.toString(),
                    buildStepTimeout(i, plan.size())
                );

                // 成功：持久化步骤结果
                execution = new StepExecution(
                    taskId, i, step.type(), StepStatus.SUCCESS,
                    stepOutput, Instant.now()
                );
                persistStepExecution(taskId, execution);
                accumulatedContext.append(stepOutput).append("\n");

            } catch (StepTimeoutException e) {
                // 超时：记录但不立即放弃，可能可以通过重试恢复
                execution = new StepExecution(
                    taskId, i, step.type(), StepStatus.TIMEOUT,
                    null, Instant.now()
                );
                persistStepExecution(taskId, execution);

                // 超时是否致命取决于步骤类型：
                // - MUST_SUCCEED: 任务标记为失败
                // - BEST_EFFORT: 跳过当前步骤继续执行
                if (step.failureMode() == FailureMode.MUST_SUCCEED) {
                    markTaskFailed(taskId, "步骤 " + i + " 超时: " + step.description());
                    return new TaskResult(taskId, AgentTaskStatus.FAILED, null);
                }
                // BEST_EFFORT 模式下跳过，使用空输出继续
                accumulatedContext.append("[步骤跳过: 超时]").append("\n");

            } catch (Exception e) {
                // 执行异常：记录并处理
                // logger.error("Step {} execution failed for task {}", i, taskId, e);

                execution = new StepExecution(
                    taskId, i, step.type(), StepStatus.FAILED,
                    e.getMessage(), Instant.now()
                );
                persistStepExecution(taskId, execution);

                if (step.failureMode() == FailureMode.MUST_SUCCEED) {
                    markTaskFailed(taskId, "步骤 " + i + " 失败: " + e.getMessage());
                    return new TaskResult(taskId, AgentTaskStatus.FAILED, null);
                }
            }
        }

        // 4. 所有步骤完成
        String finalOutput = accumulatedContext.toString();
        task = new AgentTask(taskId, AgentTaskStatus.COMPLETED, finalOutput,
            task.createdAt(), Instant.now());
        taskStore.put(taskId, task);

        return new TaskResult(taskId, AgentTaskStatus.COMPLETED, finalOutput);
    }

    /**
     * 构建步骤级别的超时时间。
     * 采用动态超时策略：靠前的步骤给更多时间（通常是信息收集），
     * 靠后的步骤给较少时间（格式化输出等简单操作）。
     */
    private Duration buildStepTimeout(int stepIndex, int totalSteps) {
        if (stepIndex < totalSteps * 0.3) {
            return Duration.ofSeconds(120);  // 前 30% 步骤：2 分钟
        } else if (stepIndex < totalSteps * 0.7) {
            return Duration.ofSeconds(60);   // 中间 40% 步骤：1 分钟
        } else {
            return Duration.ofSeconds(30);   // 最后 30% 步骤：30 秒
        }
    }

    private void persistStepExecution(String taskId, StepExecution execution) {
        stepStore.computeIfAbsent(taskId, k -> new ArrayList<>()).add(execution);
    }

    private StepExecution getStepExecution(String taskId, int stepIndex) {
        return stepStore.getOrDefault(taskId, List.of()).stream()
            .filter(se -> se.stepIndex() == stepIndex)
            .findFirst().orElse(null);
    }

    private void markTaskFailed(String taskId, String reason) {
        taskStore.put(taskId, new AgentTask(taskId, AgentTaskStatus.FAILED,
            reason, taskStore.get(taskId).createdAt(), Instant.now()));
    }

    // --- 内部数据结构 ---
    record AgentTask(String id, AgentTaskStatus status, Object finalOutput,
                     Instant createdAt, Instant completedAt) {
        AgentTask(String id, AgentTaskStatus status, List<AgentStep> plan, Instant createdAt) {
            this(id, status, null, createdAt, null);
        }
    }

    enum AgentTaskStatus { RUNNING, COMPLETED, FAILED }
    enum StepStatus { RUNNING, SUCCESS, FAILED, TIMEOUT }
    enum FailureMode { MUST_SUCCEED, BEST_EFFORT }

    record AgentStep(String description, String type, FailureMode failureMode) {}
    record StepExecution(String taskId, int stepIndex, String stepType,
                         StepStatus status, String output, Instant executedAt) {}
    record TaskResult(String taskId, AgentTaskStatus status, Object output) {}
}

interface StepExecutor {
    String execute(AgentTaskStateManager.AgentStep step,
                   String context, java.time.Duration timeout)
        throws StepTimeoutException, Exception;
}

class StepTimeoutException extends Exception {
    StepTimeoutException(String message) { super(message); }
}

三、部分成功的补偿机制

当任务在中间步骤失败时，「从头再来」是最粗暴也最浪费的选择。更好的策略是分层补偿：

flowchart TB
    Failure[步骤执行失败] --> Choice{步骤的 FailureMode?}
    
    Choice -->|MUST_SUCCEED| Retry{可重试?}
    Choice -->|BEST_EFFORT| Skip["跳过当前步骤\n使用空输出或默认值继续"]
    
    Retry -->|是| Backoff["指数退避重试\n(最多 3 次)"]
    Retry -->|否| Compensate["启动补偿事务\n(撤销已完成的副作用)"]
    
    Backoff -->|成功| Continue["继续下一步"]
    Backoff -->|3 次全部失败| Compensate
    
    Compensate -->|补偿成功| MarkFailed["任务标记为失败\n释放已占用资源"]
    Compensate -->|补偿失败| Escalate["升级到人工处理\n(Human-in-the-Loop)"]

    Skip --> Continue

    style Escalate fill:#f96,stroke:#333,stroke-width:3px
    style Compensate fill:#fc6,stroke:#333,stroke-width:2px

补偿事务的核心思想：Agent 操作的每一步都应当是可逆的或可补偿的。如果 Agent 在步骤 3 发送了一封邮件，但在步骤 7 失败了，补偿逻辑应该发送一封「抱歉，上一封邮件有误」的撤销邮件，而不是让用户困惑。

/**
 * 带补偿的回滚框架。
 * 每执行一个步骤，同时注册该步骤的补偿操作。
 * 如果后续步骤失败，按逆序执行所有已注册的补偿操作。
 */
public class CompensatingTransaction {

    // 补偿栈：后进先出（最后成功的步骤最先补偿）
    private final Deque<Runnable> compensations = new ArrayDeque<>();

    /**
     * 执行一个带补偿的步骤。
     *
     * @param stepName    步骤名称（用于日志）
     * @param action      正向操作
     * @param compensation 补偿（回滚）操作
     */
    public void executeStep(
            String stepName,
            Runnable action,
            Runnable compensation) {
        try {
            action.run();
            // 正向操作成功后，将补偿操作入栈
            compensations.push(compensation);
        } catch (Exception e) {
            // 当前步骤失败，触发补偿链
            // logger.error("Step {} failed, triggering compensation chain", stepName, e);
            rollback();
            throw new StepFailedException(stepName, e);
        }
    }

    /**
     * 按逆序执行所有补偿操作（LIFO）。
     * 每个补偿操作失败不会阻止其他补偿的执行。
     */
    public void rollback() {
        while (!compensations.isEmpty()) {
            Runnable compensation = compensations.pop();
            try {
                compensation.run();
            } catch (Exception e) {
                // 补偿失败：记录并继续执行其他补偿
                // 这是最坏情况 —— 需要 Human-in-the-Loop
                // logger.error("Compensation failed, requires manual intervention", e);
            }
        }
    }
}

四、Human-in-the-Loop 的触发条件

不是所有失败都值得召唤人类。触发 Human-in-the-Loop（HITL）需要满足特定条件：

P0 级别的 HITL 应当是同步阻塞的：Agent 暂停执行，等待人类确认后才继续。这意味着你的系统需要一个「等待队列」——Agent 任务被挂起，状态为 AWAITING_HUMAN，通过 Webhook 或轮询获取人类的审批结果。

五、总结

Agent 系统可靠性的四个核心原则：

• 持久化是地基：每一步执行前先写状态，崩溃恢复从最近成功的步骤继续。内存状态在 Agent 场景下是最大敌人。
• 补偿事务是安全网：Agent 操作的每一步都应当注册回滚操作。发送邮件、创建工单、写入数据库——这些有副作用的操作必须有撤销路径。
• 部分成功不等于全部失败：BEST_EFFORT 步骤的失败应该被容忍，MUST_SUCCEED 步骤的失败应该触发补偿而非静默丢弃。
• HITL 是最后防线：定义清晰的触发条件（补偿失败、资金操作、反复失败），不要让人类淹没在低价值的审批请求中。

衡量 Agent 系统可靠性的两个核心指标：任务成功率（目标 > 95%）和平均完成步数/总步数比（目标 > 90%）。如果一个 Agent 经常在最后几步失败，要么是步骤设计不合理（把复杂操作放太靠后），要么是补偿机制缺失。