Agent 可靠性与故障防护

服务台助手最麻烦的故障，往往不是模型完全不回答，而是它一直“很努力”地做错事：工单接口超时了，它重试十几次；查不到资料，它换几个相似关键词继续查；用户已经点了停止，它还在后台烧 token。

这篇文章不讨论“回答中途断线后如何接着显示”。那是传输和任务恢复问题，已经放在流式回复中断后的恢复和Response Job、事件日志与 SSE 重放中。本篇只关心另一件事：一次 Agent 运行正在执行时，怎样让它在有限时间、有限费用和有限副作用内结束。

我们沿用企业 IT 服务台助手的例子。它可以查询知识库、查询工单，也可以在用户确认后创建工单。每一步都只增加一项防护，你可以把短代码复制到自己的实验目录中，观察行为变化。

先看没有防护的版本

这一小步要完成什么

先写一个最小的 Agent 循环，故意留下问题。模型决定下一步动作，工具返回结果，循环继续，直到模型给出最终回答。

async def run_agent(user_text: str) -> str:
    messages = [{"role": "user", "content": user_text}]

    while True:
        decision = await call_model(messages, tools=[search_kb, get_ticket, create_ticket])
        if decision.type == "final":
            return decision.text

        result = await execute_tool(decision.tool_name, decision.arguments)
        messages.append(decision.as_message())
        messages.append({
            "role": "tool",
            "content": result.to_json(),
        })

运行观察

给它一个正常问题，例如“帮我查一下 VPN 申请流程”。它大概率会正常结束。再让 get_ticket 永远等待，或者让模型重复请求同一个查询，你会发现这个函数没有任何出口：没有最大轮数，没有总超时，也没有停止重试的条件。

先不要急着补 try/except。可靠性要先有一份明确的运行合同：这次任务最多走几步、最多花多少钱、什么时候必须停。

第 1 步：给一次运行建立预算

这一小步要完成什么

把“不能无限执行”写成数据，而不是散落在各个函数里的魔法数字。

from dataclasses import dataclass
from time import monotonic


@dataclass
class RunBudget:
    max_steps: int = 8
    max_tool_calls: int = 5
    max_output_tokens: int = 2_000
    max_cost_usd: float = 0.05
    timeout_seconds: float = 30.0

    started_at: float = 0.0
    steps: int = 0
    tool_calls: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0

    def start(self) -> None:
        self.started_at = monotonic()

    def elapsed(self) -> float:
        return monotonic() - self.started_at

    def exceeded(self) -> str | None:
        if self.steps >= self.max_steps:
            return "max_steps"
        if self.tool_calls >= self.max_tool_calls:
            return "max_tool_calls"
        if self.output_tokens >= self.max_output_tokens:
            return "max_output_tokens"
        if self.cost_usd >= self.max_cost_usd:
            return "max_cost_usd"
        if self.elapsed() >= self.timeout_seconds:
            return "timeout"
        return None

这里的 max_steps 是模型决策轮数，不是文本 token 数；max_tool_calls 单独限制工具次数，因为一个工具可能在一轮里执行很久。max_cost_usd 是估算值，实际项目应从统一的 usage 记录累加，字段设计可以参考可观测性文章中的成本计算。

把预算放进循环

async def run_agent(user_text: str) -> str:
    budget = RunBudget()
    budget.start()
    messages = [{"role": "user", "content": user_text}]

    while True:
        reason = budget.exceeded()
        if reason:
            return f"任务未完成，已达到运行上限：{reason}"

        budget.steps += 1
        decision = await call_model(messages, tools=[search_kb, get_ticket, create_ticket])
        budget.output_tokens += decision.usage.output_tokens
        budget.cost_usd += decision.usage.estimated_cost_usd

        if decision.type == "final":
            return decision.text

        budget.tool_calls += 1
        result = await execute_tool(decision.tool_name, decision.arguments)
        messages.extend([decision.as_message(), result.as_message()])

运行观察

把 max_steps 改成 2，然后输入一个需要检索、再查询工单的问题。第三次模型决策不会再发出去，用户会收到一个可解释的停止结果。这个结果不一定漂亮，但比后台继续运行更安全。

常见错误

• 只在调用模型前检查预算，工具超时后仍可能把总时长拖过上限；工具前后都要检查。
• 只限制平均成本，不限制单次任务成本；平均值掩盖不了一个失控请求。
• 把预算超限当成普通异常重试；预算是业务决策，不是网络故障。

动手改一改

给 RunBudget 增加 max_retries 和 max_recoveries。想一想：重试模型请求和恢复流式输出是不是一回事？前者属于本篇的执行预算，后者属于流式恢复教程的任务恢复预算，最好不要共用一个字段。

第 2 步：把错误分成“该重试”和“不该重试”

这一小步要完成什么

except Exception 看起来省事，却会把权限错误、参数错误和网络抖动混在一起。先定义内部错误类型和处理策略。

from dataclasses import dataclass
from typing import Literal


ErrorClass = Literal[
    "transient",       # 临时网络错误、429、上游 5xx
    "timeout",         # 本次调用超时
    "invalid_argument",# 工具参数不符合 schema
    "permission_denied",# 用户没有权限
    "not_found",       # 工单或知识不存在
    "rate_limited",    # 限流，需要等待或降级
    "cancelled",       # 用户主动取消
    "unknown",         # 未分类错误
]


@dataclass
class ClassifiedError:
    kind: ErrorClass
    message: str
    retryable: bool
    user_message: str

分类函数可以先写得朴素一些：

def classify_error(exc: Exception) -> ClassifiedError:
    if isinstance(exc, TimeoutError):
        return ClassifiedError("timeout", str(exc), True, "服务响应较慢，请稍后再试。")
    if isinstance(exc, PermissionError):
        return ClassifiedError("permission_denied", str(exc), False, "你没有执行这项操作的权限。")
    if isinstance(exc, ValueError):
        return ClassifiedError("invalid_argument", str(exc), False, "工具参数不正确，需要重新确认请求。")
    if getattr(exc, "status_code", None) == 429:
        return ClassifiedError("rate_limited", str(exc), True, "当前请求较多，我先暂停一下。")
    if getattr(exc, "status_code", 0) >= 500:
        return ClassifiedError("transient", str(exc), True, "依赖服务暂时不可用。")
    return ClassifiedError("unknown", str(exc), False, "这次操作没有完成，请转人工处理。")

运行观察

分别让工具抛出 TimeoutError、PermissionError 和参数 ValueError，打印 kind 与 retryable。超时可以进入有限重试，权限错误必须马上结束；如果两者都进入重试，日志里很快会出现“每隔一秒重复拒绝”的噪声。

供应商的 HTTP 状态码只能作为输入，不能直接作为业务结论。比如某些 API 用 400 表示参数错误，另一些服务会把业务拒绝包装成 200 响应，最终还要结合工具返回的业务错误码判断。

第 3 步：给单次调用加超时

这一小步要完成什么

总预算不能替代单次调用超时。知识库搜索、工单 API 和模型调用应有自己的时间上限，否则一个工具会独占整个任务。

Python 3.11 可以使用 asyncio.timeout：

import asyncio


async def call_with_timeout(coro, seconds: float):
    try:
        async with asyncio.timeout(seconds):
            return await coro
    except TimeoutError as exc:
        raise TimeoutError(f"调用超过 {seconds:.1f}s") from exc

在工具入口使用它：

TOOL_TIMEOUTS = {
    "search_kb": 3.0,
    "get_ticket": 5.0,
    "create_ticket": 8.0,
}


async def execute_tool_safely(
    name: str,
    arguments: dict,
    *,
    cancel_event: asyncio.Event | None = None,
):
    if cancel_event and cancel_event.is_set():
        raise asyncio.CancelledError("user_cancelled_before_tool")
    tool = TOOL_REGISTRY[name]
    coro = tool(**arguments)
    result = await call_with_timeout(coro, TOOL_TIMEOUTS[name])
    if cancel_event and cancel_event.is_set():
        raise asyncio.CancelledError("user_cancelled_after_tool")
    return result

运行观察

让 get_ticket await asyncio.sleep(10)，应在 5 秒左右得到一个可分类的超时错误，而不是等到任务总超时才知道是哪一步卡住。注意 asyncio.timeout 只能取消当前协程；如果底层 HTTP 客户端没有把取消传给 socket，仍要配置客户端自己的连接和读取超时。

动手改一改

给模型调用、工具连接、工具读取分别设置超时。为什么“连接超时 1 秒、读取超时 30 秒、整个任务 30 秒”比只设置一个 30 秒更容易排查？把三个耗时记录到日志中，观察失败到底发生在建立连接还是等待响应。

第 4 步：有限重试，并且让退避时间有上限

这一小步要完成什么

只对临时错误重试，并限制次数、总等待时间和单次退避。指数退避没有上限时，用户会等很久；所有实例同时重试时，又会把上游压垮。

import asyncio
import random


async def retry_transient(operation, *, attempts: int = 3, base_delay: float = 0.4,
                          max_delay: float = 3.0):
    for attempt in range(1, attempts + 1):
        try:
            return await operation()
        except Exception as exc:
            error = classify_error(exc)
            if not error.retryable or attempt == attempts:
                raise

            delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
            delay = delay * (0.8 + random.random() * 0.4)  # 加一点抖动
            await asyncio.sleep(delay)

    raise RuntimeError("unreachable")

运行观察

让模拟接口前两次抛出 503、第三次成功。日志应该显示三次调用，等待时间逐渐增加；把异常换成 PermissionError，它应该只调用一次。记录 attempt、delay_ms 和最终错误，后面才能判断重试是在救场还是在制造流量。

常见错误

• 对创建工单这种副作用操作直接重试，可能创建两张工单；先看第 7 步的幂等键。
• 把 429 无条件立即重试；应优先读取 Retry-After，并服从任务预算。
• 每层都实现一套重试：HTTP 客户端重试一次、工具层重试三次、Agent 层再重试三次，实际可能调用九次。

第 5 步：上游持续失败时打开熔断器

这一小步要完成什么

重试适合偶发故障，熔断适合持续故障。连续失败时先停止向故障服务发送请求，过一段时间只放一个探测请求。

from time import monotonic


class CircuitBreaker:
    def __init__(self, threshold: int = 3, cooldown: float = 10.0):
        self.threshold = threshold
        self.cooldown = cooldown
        self.failures = 0
        self.opened_at: float | None = None
        self.half_open_probe = False

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if monotonic() - self.opened_at >= self.cooldown and not self.half_open_probe:
            self.half_open_probe = True  # 半开：只放行一个探测
            return True
        return False

    def success(self) -> None:
        self.failures = 0
        self.opened_at = None
        self.half_open_probe = False

    def failure(self) -> None:
        self.failures += 1
        self.half_open_probe = False
        if self.failures >= self.threshold:
            self.opened_at = monotonic()

调用工具前先检查：

kb_breaker = CircuitBreaker(threshold=3, cooldown=15.0)


async def search_kb_guarded(query: str):
    if not kb_breaker.allow():
        raise RuntimeError("knowledge_base_circuit_open")
    try:
        result = await retry_transient(lambda: search_kb(query))
    except Exception:
        kb_breaker.failure()
        raise
    else:
        kb_breaker.success()
        return result

这里的熔断状态保存在进程内，多个实例无法共享。多实例服务应把失败计数、打开时间和半开竞争放进共享状态，并用原子更新避免多个实例同时放行探测请求。

运行观察

让知识库连续失败三次。第四次请求不会真正触发 HTTP 调用，而是立即进入降级分支。等待冷却时间后再试一次，成功则关闭熔断器，失败则继续保持打开。

第 6 步：副作用工具必须使用幂等键

这一小步要完成什么

用户说“帮我创建 VPN 工单”时，模型可能超时，网络也可能在服务端已经创建成功后断开。重试前不能凭感觉认为“刚才没成功”。

import hashlib


def make_idempotency_key(run_id: str, action: str, arguments: dict) -> str:
    normalized = repr(sorted(arguments.items()))
    raw = f"{run_id}:{action}:{normalized}".encode()
    return hashlib.sha256(raw).hexdigest()

工具调用记录至少要有：

key = make_idempotency_key(
    run_id="run-1001",
    action="create_ticket",
    arguments={"category": "vpn", "reason": "新员工入职"},
)

existing = action_store.get(key)
if existing:
    return existing.result

result = await create_ticket(**arguments, idempotency_key=key)
action_store.save(key, status="succeeded", result=result)
return result

运行观察

让 create_ticket 在服务端先写入工单，再故意在响应前抛出超时。第二次执行时，action_store 应该返回第一次的结果，而不是新增第二张工单。

幂等键要和业务动作绑定，不能只用每次请求都变化的随机 UUID。也不能把幂等键当权限检查：恢复或重试前仍然要重新验证当前用户是否有创建权限。

第 7 步：检测死循环和重复工具调用

这一小步要完成什么

最大轮数只能限制“多久停”，还要知道 Agent 是否一直在做同一件事。为每次工具调用生成一个规范化指纹。

import json
import hashlib


def call_fingerprint(name: str, arguments: dict) -> str:
    payload = json.dumps(arguments, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(f"{name}:{payload}".encode()).hexdigest()


seen_calls: dict[str, int] = {}


def record_call(name: str, arguments: dict) -> int:
    fingerprint = call_fingerprint(name, arguments)
    seen_calls[fingerprint] = seen_calls.get(fingerprint, 0) + 1
    return seen_calls[fingerprint]

在循环里设置不同的处置：

count = record_call(decision.tool_name, decision.arguments)
if count >= 2 and decision.tool_name in {"search_kb", "get_ticket"}:
    messages.append({
        "role": "system",
        "content": "这个工具调用已经重复，请改用已有结果或向用户说明信息不足。",
    })
if count >= 3:
    return "任务检测到重复调用，已停止并转人工。"

完全相同的调用可以直接复用结果；参数只差一个无意义空格时，也应在规范化阶段归一化。不要把所有重复都视为错误：查询“工单状态”在后台状态可能变化，业务上可能允许刷新，但应设置单独的刷新次数和间隔。

运行观察

让模拟模型每一轮都返回 search_kb({"query": "VPN"})。第二次调用后追加提醒，第三次直接停止。日志里应包含指纹、次数和停止原因，便于评估文章中的轨迹指标。

第 8 步：准备可解释的降级路径

这一小步要完成什么

“工具失败”不等于“整段对话只能报错”。不同故障要有不同降级：知识库不可用时可以给出不带内部政策的澄清；工单查询失败时可以让用户稍后重试；创建操作失败时不能假装成功。

async def answer_with_fallback(question: str) -> str:
    try:
        docs = await search_kb_guarded(question)
    except Exception as exc:
        error = classify_error(exc)
        if error.kind in {"timeout", "transient", "rate_limited"}:
            return "知识库暂时不可用。我可以先告诉你需要准备哪些信息，稍后再帮你查询具体政策。"
        raise

    return await compose_answer(question, docs)

降级回答必须说明信息边界，不能在没有资料时编造政策。涉及创建、删除、退款等副作用操作时，失败出口应明确写成“未确认执行”，而不是“应该已经完成”。

动手改一改

为三种场景写不同的用户文案：只读查询超时、创建工单结果未知、用户没有权限。比较它们是否都告诉了用户“下一步能做什么”，但没有承诺系统并未确认的结果。

第 9 步：加入人工接管和高风险出口

这一小步要完成什么

自动化系统需要一个明确的“停在这里，交给人”。它不是把异常吞掉，而是保存上下文、原因和已执行动作。

async def handoff_to_human(run_id: str, reason: str, messages: list[dict]) -> str:
    case_id = await human_queue.create({
        "run_id": run_id,
        "reason": reason,
        "conversation": redact_messages(messages),
        "status": "waiting_for_agent",
    })
    return f"这项操作需要人工确认，已创建服务单 {case_id}。"

以下情况优先接管：

• 预算耗尽但问题仍未解决；
• 同一工具连续失败；
• 检测到重复调用或状态矛盾；
• 创建、删除、权限变更等高风险动作结果未知；
• 用户明确要求人工处理。

人工队列里的内容也要脱敏。不要因为“转人工”就把完整 system prompt、API Key 或其他租户的检索结果一并复制过去。

第 10 步：把取消信号传到每一层

这一小步要完成什么

用户点击“停止”时，不能只让前端隐藏加载动画。服务端要把取消意图传给 Agent、模型请求、工具请求和后台任务。

import asyncio


async def run_agent_with_cancel(user_text: str, cancel_event: asyncio.Event):
    budget = RunBudget()
    budget.start()

    while True:
        if cancel_event.is_set():
            raise asyncio.CancelledError("user_cancelled")

        reason = budget.exceeded()
        if reason:
            return await handoff_or_fail(reason)

        decision = await call_model(
            user_text,
            cancel_event=cancel_event,
            timeout=5.0,
        )
        if cancel_event.is_set():
            raise asyncio.CancelledError("user_cancelled")

        result = await execute_tool_safely(
            decision.tool_name,
            decision.arguments,
            cancel_event=cancel_event,
        )

模型 SDK 和 HTTP 客户端是否支持取消，需要查你实际使用的库；如果底层不支持，就至少在每个事件边界检查取消，并停止提交新的工具动作。用户取消与网络断开不是一回事：浏览器断开可能只是临时网络问题，不能自动把任务标成 cancelled。这点在Response Job 文章的任务与订阅解耦中有更完整的说明。

运行观察

让工具执行一个 10 秒任务，在第 2 秒设置 cancel_event。理想结果是：当前可取消的请求尽快结束，后续工具不再提交，任务最终状态记录为 cancelled，而不是 completed 或 failed。

第 11 步：统一最终失败出口

这一小步要完成什么

所有分支最终都要落到有限的状态集合，方便前端展示、日志统计和评估。

FINAL_STATUSES = {
    "completed",
    "cancelled",
    "budget_exceeded",
    "dependency_unavailable",
    "permission_denied",
    "needs_human",
    "failed",
}


def finish(run_id: str, status: str, *, reason: str, user_message: str):
    if status not in FINAL_STATUSES:
        raise ValueError(f"不是最终状态：{status}")
    audit_log.write({
        "run_id": run_id,
        "status": status,
        "reason": reason,
    })
    return {"status": status, "message": user_message}

建议把内部原因和用户文案分开：timeout 可以展示为“服务响应较慢”，但日志里仍要保留具体是哪一个工具、哪一次 attempt、等待了多久。最终失败也要返回已完成的安全部分，例如已经查到的工单号，但不能把未确认的副作用写成成功。

第 12 步：做几次故障注入

只看正常路径，很难知道防护是否真的有效。给实验代码加一个可控的故障开关：

class Faults:
    model_timeout = False
    kb_503_times = 0
    ticket_duplicate_response = False
    repeated_tool_call = False


faults = Faults()

然后按顺序测试：

1. model_timeout = True：确认单次模型调用超时，任务不会无限等待。
2. kb_503_times = 2：确认只对临时错误有限重试，成功后关闭熔断器。
3. kb_503_times = 5：确认熔断打开，后续请求快速降级。
4. ticket_duplicate_response = True：模拟服务端已创建工单但响应丢失，确认幂等键不会重复创建。
5. repeated_tool_call = True：确认重复指纹触发提醒和停止。
6. 在工具执行中设置取消事件：确认后续动作不再提交。
7. 把 max_cost_usd 调成极小值：确认结果是预算超限，而不是普通系统错误。

每次实验记录 run_id、步骤数、工具次数、重试次数、熔断状态、最终状态和估算成本。这样才能看出“防护生效了”，而不是只看到页面上出现了一句错误。

和流式恢复怎么分工

两个问题经常被混在一起：

问题	本篇故障防护	流式恢复文章
工具卡住	设置单次超时、重试和熔断	不负责决定工具是否继续执行
模型调用超预算	停止当前 Agent 运行	不负责扩大恢复预算
浏览器断线	不应自动把任务取消	保存事件并允许重连、重放
副作用重试	用幂等键避免重复	恢复时先查询执行记录
Worker 崩溃	记录失败并进入接管或重试	通过 Response Job 和事件日志恢复任务

可以这样记：可靠性防护决定“这次运行何时停、失败后做什么”；流式恢复决定“已经生成的内容如何保存和再次交付”。 两者都需要状态和幂等，但状态的目的不同。

常见误区

• 把最大轮数当成完整可靠性方案：一个工具可能一轮就卡死，还需要单次超时和总超时。
• 所有异常都重试：权限错误、参数错误和用户取消通常不该重试。
• 重试副作用工具却没有幂等键：响应丢失不代表动作没发生。
• 只在内存里做熔断：多实例下每个进程都有自己的失败计数，状态会互相打架。
• 降级时编造资料：检索不可用时宁可说明无法确认，也不要生成看似具体的公司政策。
• 把浏览器断开当成取消：网络连接和后台任务要分开建模。
• 只记录最终失败：没有步骤、工具、重试和预算日志，事后无法解释成本和责任边界。

做到这里，一个 Agent 的失败就不再只是“抛了异常”：它会在预算、超时、重试、熔断、幂等、循环检测、降级和人工接管之间走一条可解释的路径。下一步再把这些事件接入评估文章和可观测性文章，就能统计哪些故障最常见、哪些防护真的降低了成本，以及哪些任务应该交给人处理。