阅读时间：约 16 分钟
前置知识：理解 Agent 决策循环（P01）、能调用外部工具（P02）、上下文管理（P03）

---

P03 让上下文干净了，Agent 的运行环境稳了。但工具调用还是会失败：网络超时、API 限流、参数出错。这些失败该由谁处理？

---

前言：一个问题，先想清楚

假设你的 Agent 调了 get_weather("北京")，返回了一个错误。你第一反应是什么？

# 加个 retry？
def call_tool_with_retry(func, args, max_retries=3):
    for i in range(max_retries):
        try:
            return func(**args)
        except Exception:
            if i == max_retries - 1:
                raise

这段代码看起来没问题。但如果是参数错误（city name 拼错了），重试 3 次还是错。如果是扣款操作超时，重试可能导致重复扣款。

不是所有失败都适合重试。 核心问题是：什么该由工具层自动处理，什么该交给 Agent 自己决策？

📌 本章核心：失败分两类：低语义错误（工具层自动重试）和语义级错误（Agent 决策下一步）。关键是结构化错误信息：这是两层之间的协议。

---

第一部分：失败的三种颜色

所有工具调用失败，按语义深浅分为三类：

类型	典型场景	谁处理	策略
🟡 短暂性错误	网络超时、API 限流、临时 5xx	工具层	指数退避自动重试
🟠 参数/逻辑错误	参数格式错、工具不存在、权限不足	Agent 层	模型分析错误→改参→重试
🔴 危险操作错误	写操作超时不可能定结果、扣款结果不明	暂停	标记→人工确认

这三类的核心区别是：错误的原因能不能被代码自动修复。

短暂性错误的原因不在你的代码里（网络、服务端），加个延时重试大概率能好。参数错误的原因在调用者身上（参数不对），需要改调用方式。危险操作则涉及副作用，重试可能造成重复执行。

📌 本章要点：失败分三类：短暂性（工具层重试）、参数逻辑类（Agent 决策）、危险操作（暂停确认）。不是所有失败都能无脑重跑。

---

分完类了。先看最简单的一层：工具层。

第二部分：工具层：自动重试的边界

什么该自动重试

工具层的职责是处理低语义、短暂性、可幂等的失败。具体来说：

1. 网络超时（connection timeout, read timeout）
2. 服务端临时故障（HTTP 500, 502, 503, 504）
3. 限流（HTTP 429, rate limit exceeded）
4. DNS/连接重置（connection reset, name resolution failure）

这些错误的特点是：原因在远端，跟你的调用参数无关，等待一会重试大概率能通。

实现：指数退避 + 抖动

import time
import random
import functools

def retry_on_transient(
    max_retries=3,          # 最多重试 3 次
    base_delay=1.0,         # 基础等待 1 秒
    max_delay=30.0,         # 最长等待 30 秒
    jitter=True             # 是否加随机抖动（防止重试风暴）
):
    """
    工具层重试装饰器：只处理短暂性错误
    """
    # 这些 HTTP 状态码意味着"等等再试"
    retryable_statuses = {429, 500, 502, 503, 504}
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    
                    # 最后一次尝试，不再重试
                    if attempt == max_retries:
                        break
                    
                    # 判断错误是否可重试
                    if not is_retryable(e, retryable_statuses):
                        raise  # 不可重试的错误，直接抛给调用方
                    
                    # 计算等待时间
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    if jitter:
                        delay *= random.uniform(0.5, 1.5)  # 加 50% 随机抖动

                    # 从限流响应中读取建议等待时间
                    wait = get_retry_after(e, default=delay)
                    time.sleep(wait)
            
            # 超过最大重试次数，抛出最后一次异常
            raise last_exception
        return wrapper
    return decorator


def is_retryable(exception, retryable_statuses):
    """判断异常是否可重试"""
    # requests / httpx 的 HTTPStatusError
    if hasattr(exception, 'response'):
        code = getattr(exception.response, 'status_code', None)
        if code in retryable_statuses:
            return True
    # urllib3 / socket 的网络层异常
    if hasattr(exception, '__module__'):
        mod = exception.__module__
        if 'urllib3' in mod or 'httpcore' in mod or 'socket' in mod:
            return True
    return False


def get_retry_after(exception, default=None):
    """从限流响应头读取 Retry-After 建议"""
    if hasattr(exception, 'response'):
        headers = getattr(exception.response, 'headers', {})
        after = headers.get('Retry-After')
        if after is not None:
            try:
                return float(after)
            except (TypeError, ValueError):
                pass
    return default


# 使用示例
@retry_on_transient(max_retries=3, base_delay=1.0)
def call_external_api(endpoint, params):
    response = httpx.get(endpoint, params=params)
    response.raise_for_status()
    return response.json()

三条红线

工具层重试有三条不能逾越的规则：

1. 不重试参数错误：参数错在调用方，重试 100 次也不会对。需要把错误返回给 Agent 让它自己调整。
2. 不重试写操作：如果超时发生在 POST/PUT/DELETE 上，你不知道服务端有没有执行成功。重试可能导致重复创建、重复扣款。
3. 不重试权限错误：401/403 代表认证或授权失败，重试不会改变结果。

📌 本章要点：工具层只处理短暂性故障（超时、限流、5xx），用指数退避 + 抖动防止重试风暴。参数错、权限错、写操作不能自动重试。

---

工具层已经处理了网络层的错误。剩下的失败：参数错、工具选错、权限不足：需要 Agent 自己决策。

第三部分：Agent 层：语义级恢复

Agent 看到的不是"失败了"，而是"为什么失败"

当工具层无法自动恢复时，错误信息要原封不动地传给 Agent。但原始异常不是给模型看的好材料：ConnectionError: [Errno 111] Connection refused 对模型来说跟 FileNotFoundError 几乎没有区别。

正确的做法：结构化错误信息。 这是一种工具层和 Agent 层之间的协议。

结构化错误协议

class ToolErrorCode:
    """错误码枚举：每种失败一个明确标识"""
    TIMEOUT         = "timeout"          # 网络超时
    RATE_LIMITED    = "rate_limited"     # 被限流
    SERVER_ERROR    = "server_error"     # 服务端 5xx
    INVALID_PARAM   = "invalid_param"    # 参数格式错误
    NOT_FOUND       = "not_found"        # 资源不存在
    PERMISSION_DENY = "permission_deny"  # 权限不足
    BUSINESS_REJECT = "business_reject"  # 业务规则拒绝


def format_tool_error(error_code, message, partial_result=None, suggested_action=None):
    """
    把原始异常转成 Agent 能理解的结构化错误
    这不是给代码看的，是给模型看的
    """
    error = {
        "error_type": error_code,                  # 错误类型
        "retryable": error_code in (               # 模型是否可以自己尝试恢复
            ToolErrorCode.INVALID_PARAM,
            ToolErrorCode.NOT_FOUND
        ),
        "idempotent_safe": error_code != ToolErrorCode.BUSINESS_REJECT,  # 写操作不能盲目重试
        "message": message                         # 人类可读的说明
    }
    if partial_result is not None:
        error["partial_result"] = partial_result   # 部分成功的数据（模型可能还能用）
    if suggested_action is not None:
        error["suggested_action"] = suggested_action  # 建议的恢复路径
    return json.dumps(error, ensure_ascii=False)


# 工具函数内部使用
def get_weather(city):
    try:
        response = httpx.get(f"https://api.weather.com/v1/{city}", timeout=5.0)
        response.raise_for_status()
        return response.json()
    except httpx.TimeoutException:
        return format_tool_error(
            ToolErrorCode.TIMEOUT,
            f"查询{city}天气超时，建议换城市或稍后重试",
            suggested_action="retry_with_different_city"
        )
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return format_tool_error(
                ToolErrorCode.NOT_FOUND,
                f"未找到城市'{city}'，请检查城市名称拼写",
                suggested_action="correct_city_name"
            )
        if e.response.status_code == 429:
            return format_tool_error(
                ToolErrorCode.RATE_LIMITED,
                "API 调用频率过高，请等待 30 秒后重试"
            )
        return format_tool_error(
            ToolErrorCode.SERVER_ERROR,
            f"天气服务暂时不可用 (HTTP {e.response.status_code})"
        )

Agent 收到错误后怎么决策

关键：错误信息不是让模型看一眼就完了。模型要根据 error_type 和 suggested_action 决定下一步。不同错误类型对应不同恢复路径：

ERROR_RECOVERY_MAP = {
    # 短暂性错误：工具层已重试失败 → Agent 换策略
    ToolErrorCode.TIMEOUT: "change_params",        # 改参数（缩小范围、换城市）
    ToolErrorCode.RATE_LIMITED: "wait_or_degrade", # 等待或降级
    ToolErrorCode.SERVER_ERROR: "degrade_or_stop", # 降级或停止
    
    # 参数/逻辑错误：Agent 自己修正
    ToolErrorCode.INVALID_PARAM: "fix_params",     # 修正参数格式
    ToolErrorCode.NOT_FOUND: "correct_input",      # 纠正输入
    
    # 不可自动恢复
    ToolErrorCode.PERMISSION_DENY: "ask_user",     # 请求用户授权
    ToolErrorCode.BUSINESS_REJECT: "stop_report",  # 停止并报告
}

Agent 在执行时的逻辑就变成了一个带分支的决策树：不是"失败就重试"，而是"失败→看错误类型→选恢复策略"。

📌 本章要点：结构化错误信息是工具层和 Agent 层的契约。error_type 让模型知道失败类别，suggested_action 给恢复方向。不同错误类型映射到不同恢复策略。

---

错误分类和恢复路径讲完了。但有一类错误特别危险，需要单独讲。

第四部分：危险操作：写完就不能撤回

幂等性：重试的真正红线

读操作（GET、查询、搜索）天然幂等：调 10 次和调 1 次结果一样，安全重试。但写操作不是。

# 这个函数如果超时了，你敢重试吗？
def transfer_money(from_account, to_account, amount):
    response = httpx.post("https://bank.example.com/transfer", json={
        "from": from_account,
        "to": to_account,
        "amount": amount
    }, timeout=5.0)
    # 如果这里超时了：钱转没转？不知道。
    return response.json()

危险就在这里：POST 请求发出去了，但响应没回来。服务端可能已经执行了转账，重试等于再转一次。

解决方案：去重键

import uuid

def transfer_money_safe(from_account, to_account, amount):
    # 为本次操作生成唯一去重键
    idempotency_key = str(uuid.uuid4())
    
    try:
        response = httpx.post("https://bank.example.com/transfer", json={
            "from": from_account,
            "to": to_account,
            "amount": amount
        }, headers={
            "Idempotency-Key": idempotency_key   # 服务端用这个键去重
        }, timeout=5.0)
        response.raise_for_status()
        return response.json()
    except httpx.TimeoutException:
        # 超时了：不重试转账，而是查询状态
        status = httpx.get(f"https://bank.example.com/transfer/{idempotency_key}")
        if status.status_code == 200:
            return status.json()                        # 已成功，返回结果
        elif status.status_code == 404:
            return {"error": "transfer_failed_timeout"} # 没查到，可能是真的失败了

三条写操作安全规则：

1. 每个写操作必须有去重键（Idempotency-Key）
2. 超时不重试，改用查询接口验证结果
3. 无法验证时，标记为"结果不明"并请求人工确认

📌 本章要点：写操作的超时不能无脑重试：用去重键防止重复执行，超时后查询状态而不是再发一次。无法确认结果时暂停并标记。

---

理论讲完了。把工具层重试、Agent 层决策、结构化错误、幂等保护串起来。

第五部分：实战：完整错误恢复管理器

import time
import json
import uuid
import random
import functools
import httpx
from dataclasses import dataclass, field
from typing import Optional, Callable, Any


# ═══════════════════════════════════════════
# 完整错误恢复管理器
# 串联：工具层重试 → 结构化错误 → Agent 决策 → 幂等保护
# ═══════════════════════════════════════════

class ErrorCode:
    TIMEOUT         = "timeout"
    RATE_LIMITED    = "rate_limited"
    SERVER_ERROR    = "server_error"
    INVALID_PARAM   = "invalid_param"
    NOT_FOUND       = "not_found"
    PERMISSION_DENY = "permission_deny"
    DANGEROUS_WRITE = "dangerous_write"    # 新增：危险写操作超时


@dataclass
class StructuredError:
    """结构化错误：工具层和 Agent 层之间的协议"""
    error_type: str
    message: str
    retryable: bool = False         # Agent 是否可以尝试恢复
    idempotent_safe: bool = True    # 重试是否安全
    partial_result: Any = None      # 部分结果（可能仍有价值）
    suggested_action: str = ""      # 建议恢复动作
    idempotency_key: str = ""       # 去重键（写操作场景）


class RecoveryManager:
    """错误恢复管理器：统一管理工具层和 Agent 层的重试逻辑"""
    
    def __init__(self):
        self.pending_confirmations = []  # 需要人工确认的操作
    
    # ── 工具层：指数退避重试 ──
    def should_retry_at_tool_level(self, error: StructuredError) -> bool:
        """工具层只重试短暂性错误"""
        return error.error_type in (
            ErrorCode.TIMEOUT, ErrorCode.RATE_LIMITED, ErrorCode.SERVER_ERROR
        )
    
    def retry_with_backoff(self, func, max_retries=3, base_delay=1.0):
        """工具层自动重试（指数退避 + 抖动）"""
        retryable_statuses = {429, 500, 502, 503, 504}
        last_error = None
        
        for attempt in range(max_retries + 1):
            try:
                return func()
            except Exception as e:
                last_error = e
                if attempt == max_retries:
                    break
                if not is_retryable(e, retryable_statuses):
                    raise
                delay = min(base_delay * (2 ** attempt), 30) * random.uniform(0.5, 1.5)
                time.sleep(delay)
        
        # 工具层重试全部失败 → 转成结构化错误交给 Agent
        return StructuredError(
            error_type=ErrorCode.TIMEOUT if "timeout" in str(last_error).lower()
            else ErrorCode.SERVER_ERROR,
            message=f"工具层重试 {max_retries} 次后仍失败: {last_error}",
            retryable=False,
            suggested_action="degrade_or_stop"
        )
    
    # ── Agent 层：根据错误类型选恢复策略 ──
    def get_agent_recovery_strategy(self, error: StructuredError) -> str:
        """返回 Agent 应采取的恢复策略"""
        strategies = {
            ErrorCode.TIMEOUT:         "change_params",    # 改参数再试
            ErrorCode.RATE_LIMITED:    "wait_or_degrade",  # 等待或降级
            ErrorCode.SERVER_ERROR:    "degrade_or_stop",  # 降级或停止
            ErrorCode.INVALID_PARAM:   "fix_params",       # 修正参数
            ErrorCode.NOT_FOUND:       "correct_input",    # 纠正输入
            ErrorCode.PERMISSION_DENY: "ask_user",         # 请求授权
            ErrorCode.DANGEROUS_WRITE: "verify_or_stop",   # 验证或停止
        }
        return strategies.get(error.error_type, "stop_report")
    
    # ── 幂等保护：写操作只做一次 ──
    def safe_write(self, operation_name, write_func, verify_func=None):
        """安全的写操作：带去重键保护"""
        idempotency_key = str(uuid.uuid4())
        try:
            result = write_func(idempotency_key)
            return result
        except Exception as e:
            if "timeout" in str(e).lower():
                # 超时→尝试查询结果
                if verify_func:
                    try:
                        return verify_func(idempotency_key)
                    except Exception:
                        pass
                # 无法验证→标记为"结果不明"
                self.pending_confirmations.append({
                    "operation": operation_name,
                    "idempotency_key": idempotency_key,
                    "status": "unknown"
                })
                return StructuredError(
                    error_type=ErrorCode.DANGEROUS_WRITE,
                    message=f"操作'{operation_name}'执行结果不明(超时)，请确认",
                    idempotency_key=idempotency_key,
                    suggested_action="verify_or_stop"
                )
            raise
    
    def summary(self):
        """报告当前未确认的危险操作"""
        return {
            "pending_count": len(self.pending_confirmations),
            "pending": self.pending_confirmations
        }


# ── 完整使用示例 ──
rm = RecoveryManager()

# 场景一：查询天气（读操作，工具层自动重试）
try:
    weather = rm.retry_with_backoff(
        lambda: httpx.get("https://api.weather.com/v1/beijing", timeout=5.0).json()
    )
except Exception:
    # 工具层重试 3 次全部失败，返回结构化错误给 Agent
    error = StructuredError(
        ErrorCode.SERVER_ERROR, "天气服务不可用",
        retryable=False, suggested_action="degrade_or_stop"
    )
    strategy = rm.get_agent_recovery_strategy(error)
    # Agent 根据 strategy 决定：降级用缓存数据，或告诉用户暂时查不到

# 场景二：转账（写操作，幂等保护）
def do_transfer(key):
    return httpx.post("https://bank.example.com/transfer",
        json={"amount": 100}, headers={"Idempotency-Key": key}, timeout=5.0
    ).json()

def check_transfer(key):
    return httpx.get(f"https://bank.example.com/transfer/{key}").json()

result = rm.safe_write("transfer_money", do_transfer, check_transfer)
if isinstance(result, StructuredError):
    print(f"危险操作未确认: {result.message}")
    print(f"去重键: {result.idempotency_key}")

print(f"未确认操作: {rm.summary()}")

📌 本章要点：一个完整的错误恢复管理器需要四层：工具层指数退避、结构化错误协议、Agent 决策映射表、写操作幂等保护。四层协作才能让 Agent 在真实环境中稳定运行。

---

第六部分：DeepSeek V4 的坑：工具调用无限循环

2026 年实测 DeepSeek V4 时，一个常见陷阱是工具调用无限重试。当工具返回空结果（如 git_diff 为空），模型可能反复调用同一个工具，直到 token 耗尽。

# 触发无限循环的场景
轮次 1: 调用 git_diff → 返回空
轮次 2: 调用 git_diff → 返回空（模型不认为空是"失败"）
轮次 3: 调用 git_diff → 返回空
...
轮次 37: 上下文窗口溢出

解决方案：在 Agent 循环中增加工具调用次数上限，并在工具结果中明确标注状态：

MAX_TOOL_CALLS = 5  # 单次任务最多调用 5 次工具

def agent_loop(messages, tools, max_calls=MAX_TOOL_CALLS):
    call_count = 0
    while call_count < max_calls:
        response = call_llm(messages, tools)
        if response.content:        # 模型返回了最终答案
            return response.content
        if response.tool_calls:     # 模型请求继续调工具
            call_count += 1
            if call_count >= max_calls:
                return "超过最大工具调用次数，任务中止"
            messages.extend(execute_tools(response.tool_calls))
    return "达到上限未获结果"

DeepSeek V4 的思维模式（thinking mode）在多轮工具调用时还要求将 reasoning_content 回传，否则会触发校验失败。这些经验说明：即使是最先进的模型，在生产环境中也需要在工具层和 Agent 层都做足保护。

📌 核心观点：DeepSeek V4 的 Agent 能力很强，但工具调用无限循环是真实存在的坑。解决方案是设置调用上限，并让工具返回明确的状态标识。

---

总结：读完这篇，你应该带走这几件事

1. 三层错误分类：短暂性错误（工具层重试）→ 参数逻辑错误（Agent 决策）→ 危险操作（暂停确认）。
2. 工具层职责：低语义、短暂性、可幂等的错误：用指数退避 + 抖动自动重试。不碰参数错、权限错、写操作。
3. Agent 层职责：接收结构化错误信息，根据 error_type 和 suggested_action 决定下一步：改参、换工具、降级、请求授权、停止。
4. 结构化错误是契约：error_type + retryable + idempotent_safe + suggested_action：让模型从"看到异常"变成"理解失败原因"。
5. 写操作必须幂等保护：去重键 + 超时查询 + 结果不明标记。重复扣款只需要一次超时。

🤔 思考一下：你的 Agent 项目里，工具调用失败时是直接抛异常还是返回结构化错误？如果是转账超时，你的代码会怎么做？

---

思维导图

• 重试策略：工具层 vs Agent 层
  • 三层错误分类
    • 🟡 短暂性：网络超时、限流、5xx → 工具层重试
    • 🟠 参数逻辑：格式错、不存在、权限 → Agent 决策
    • 🔴 危险操作：写操作超时 → 暂停确认
  • 工具层
    • 指数退避 + 抖动（防重试风暴）
    • 读 Retry-After 响应头
    • 不碰参数错、权限错、写操作
  • Agent 层
    • 结构化错误协议（error_type/suggested_action）
    • 6 种恢复策略映射
    • 从"看到异常"到"理解失败原因"
  • 幂等保护
    • 去重键（Idempotency-Key）
    • 超时查状态，不重试
    • 无法确认就标记 + 人工
  • DeepSeek V4 实战
    • 工具调用上限防无限循环
    • reasoning_content 回传要求
  • 核心 Takeaways
    • 不是所有失败都能重试
    • 结构化错误是两层间的协议
    • 写操作必须有去重键

---

下一篇预告

P05. Memory 系统：从会话记忆到长期记忆

重试策略让 Agent 稳定了，但 Agent 记不住事怎么办？用户说"上次那个北京的温度"，Agent 一脸茫然。短期记忆靠上下文，长期记忆靠外部存储。三种记忆模式：会话记忆、摘要记忆、向量记忆，各自适用什么场景？

🤔 思考一下：你的 Agent 关掉重启后，还记得用户上一轮对话说了什么吗？如果不记得，怎么让它记住？