原文：https://www.mstx.cn/harness/424.html · 创建：2026-09-05 · 更新：2026-09-05 · 阅读：1124

---

工具会失败，这是常态

在理想世界中，工具总是能正确执行：网络畅通、数据库响应迅速、API返回预期的格式。但在真实世界中，工具执行面临各种失败：

• 网络超时：外部API响应慢或不可达
• 数据异常：数据库返回了意外的空结果或格式错误的数据
• 权限问题：API密钥过期或权限不足
• 限流：调用频率超过了服务提供商的限制
• 输入问题：LLM填的参数虽然在Schema校验中通过了，但业务逻辑上不合理
• 依赖故障：工具依赖的下游服务宕机

一个好的Harness不是避免错误（那是不可能的），而是优雅地处理错误。 错误处理策略的好坏，直接决定了Agent在面对真实世界时的鲁棒性。

本节将构建一个完整的工具错误处理与重试系统。核心目标是：让Agent在工具失败时能够恢复，而不是直接崩溃或给出错误回答。

错误的分类

并非所有错误都应该用同样的方式处理。首先需要对错误进行分类：

from enum import Enum
from dataclasses import dataclass
from typing import Any, Dict, Optional, Callable

class ErrorType(Enum):
    """工具错误的分类"""
    # 可重试的瞬时错误
    TIMEOUT = "timeout"                    # 执行超时
    NETWORK = "network"                    # 网络错误
    RATE_LIMIT = "rate_limit"              # 限流
    SERVICE_UNAVAILABLE = "service_unavailable"  # 服务暂时不可用

    # 不可重试的永久错误
    INVALID_ARGUMENT = "invalid_argument"  # 参数不合法
    PERMISSION_DENIED = "permission_denied"  # 权限不足
    NOT_FOUND = "not_found"                # 资源不存在
    CONFLICT = "conflict"                  # 状态冲突
    INTERNAL = "internal"                  # 工具内部错误

    # 特殊错误
    UNKNOWN = "unknown"                    # 未知错误

@dataclass
class ToolError:
    """标准化的工具错误"""
    error_type: ErrorType
    message: str
    retryable: bool
    retry_after_seconds: Optional[int] = None  # 建议的重试等待时间
    details: Dict[str, Any] = None

    def __post_init__(self):
        if self.details is None:
            self.details = {}

    def to_llm_message(self) -> str:
        """转换为给LLM看的错误信息"""
        lines = [
            f"错误类型：{self.error_type.value}",
            f"错误信息：{self.message}",
        ]
        if self.retryable:
            if self.retry_after_seconds:
                lines.append(f"这是可重试的瞬时错误，建议在{self.retry_after_seconds}秒后重试。")
            else:
                lines.append("这是可重试的瞬时错误，可以尝试重试。")
        else:
            lines.append("这是不可重试的永久错误，重试不会解决此问题。建议调整参数或使用其他工具。")

        if self.details:
            lines.append(f"详细信息：{self.details}")

        return "\n".join(lines)

这个分类的核心思想是：区分瞬时错误和永久错误。

• 瞬时错误：重试可能有效。网络超时、限流、服务暂时不可用——这些错误可能在下次尝试时消失。
• 永久错误：重试无效。参数不合法、权限不足、资源不存在——无论重试多少次，结果都一样。

盲目重试永久错误是浪费资源和时间，而不重试瞬时错误则降低了Agent的可靠性。

错误识别：从异常到结构化错误

工具函数在执行时抛出的是普通Python异常。错误处理系统的第一步是将异常转换为结构化的ToolError：

class ErrorClassifier:
    """将Python异常分类为结构化的ToolError"""

    def __init__(self):
        # 注册异常类型到错误类型的映射
        self.exception_mappings = {}
        self.message_patterns = []

    def register_exception(self, exception_type, error_type: ErrorType,
                           retryable: bool):
        """注册异常类型映射"""
        self.exception_mappings[exception_type] = (error_type, retryable)

    def register_pattern(self, pattern: str, error_type: ErrorType,
                         retryable: bool):
        """注册消息模式映射（通过正则表达式匹配错误消息）"""
        import re
        self.message_patterns.append(
            (re.compile(pattern, re.IGNORECASE), error_type, retryable)
        )

    def classify(self, exception: Exception) -> ToolError:
        """将异常分类为结构化的ToolError"""
        # 第一优先：精确的异常类型匹配
        for exc_type, (error_type, retryable) in self.exception_mappings.items():
            if isinstance(exception, exc_type):
                return ToolError(
                    error_type=error_type,
                    message=str(exception),
                    retryable=retryable,
                )

        # 第二优先：消息模式匹配
        error_message = str(exception)
        for pattern, error_type, retryable in self.message_patterns:
            if pattern.search(error_message):
                return ToolError(
                    error_type=error_type,
                    message=error_message,
                    retryable=retryable,
                )

        # 默认：未知错误（不重试）
        return ToolError(
            error_type=ErrorType.UNKNOWN,
            message=error_message,
            retryable=False,
        )

# 配置默认的分类器
default_classifier = ErrorClassifier()

# 注册常见的异常类型
default_classifier.register_exception(TimeoutError, ErrorType.TIMEOUT, True)
default_classifier.register_exception(ConnectionError, ErrorType.NETWORK, True)
default_classifier.register_exception(PermissionError, ErrorType.PERMISSION_DENIED, False)
default_classifier.register_exception(ValueError, ErrorType.INVALID_ARGUMENT, False)
default_classifier.register_exception(KeyError, ErrorType.NOT_FOUND, False)
default_classifier.register_exception(FileNotFoundError, ErrorType.NOT_FOUND, False)

# 注册常见的错误消息模式
default_classifier.register_pattern(r"rate.?limit", ErrorType.RATE_LIMIT, True)
default_classifier.register_pattern(r"too many requests", ErrorType.RATE_LIMIT, True)
default_classifier.register_pattern(r"service unavailable", ErrorType.SERVICE_UNAVAILABLE, True)
default_classifier.register_pattern(r"503", ErrorType.SERVICE_UNAVAILABLE, True)
default_classifier.register_pattern(r"timeout", ErrorType.TIMEOUT, True)
default_classifier.register_pattern(r"timed out", ErrorType.TIMEOUT, True)
default_classifier.register_pattern(r"unauthorized", ErrorType.PERMISSION_DENIED, False)
default_classifier.register_pattern(r"forbidden", ErrorType.PERMISSION_DENIED, False)
default_classifier.register_pattern(r"not found", ErrorType.NOT_FOUND, False)
default_classifier.register_pattern(r"does not exist", ErrorType.NOT_FOUND, False)
default_classifier.register_pattern(r"already exists", ErrorType.CONFLICT, False)
default_classifier.register_pattern(r"conflict", ErrorType.CONFLICT, False)

错误分类的价值：它让后续的重试策略和LLM反馈策略能够根据错误类型做出差异化的决策。一个TIMEOUT错误和PERMISSION_DENIED错误的处理方式完全不同。

重试策略的设计

重试不是"再试一次"那么简单。一个好的重试策略需要考虑：重试什么、重试多少次、间隔多久、什么条件下停止。

策略一：固定间隔重试

最简单的策略：每次重试之间等待固定的时间。

class FixedIntervalRetry:
    """固定间隔重试策略"""

    def __init__(self, max_attempts: int = 3, interval_seconds: float = 1.0):
        self.max_attempts = max_attempts
        self.interval_seconds = interval_seconds

    def get_delay(self, attempt: int) -> Optional[float]:
        """返回第attempt次重试前的等待时间。返回None表示不应重试。"""
        if attempt >= self.max_attempts:
            return None
        return self.interval_seconds

适用场景：对延迟不敏感的场景，简单直接。 缺点：在服务过载时，固定间隔的重试可能加重服务负担。

策略二：指数退避重试

每次重试的等待时间按指数增长。

class ExponentialBackoffRetry:
    """指数退避重试策略"""

    def __init__(self, max_attempts: int = 5,
                 base_delay: float = 0.5,
                 max_delay: float = 30.0,
                 multiplier: float = 2.0):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.multiplier = multiplier

    def get_delay(self, attempt: int) -> Optional[float]:
        if attempt >= self.max_attempts:
            return None

        import random
        # 指数增长 + 随机抖动
        delay = self.base_delay * (self.multiplier ** attempt)
        # 添加抖动，避免多个客户端同时重试
        jitter = random.uniform(0, 0.1 * delay)
        delay = delay + jitter
        # 限制最大延迟
        return min(delay, self.max_delay)

适用场景：网络服务调用、限流恢复等待。指数退避是分布式系统中的最佳实践——它给服务恢复留出时间，同时随机抖动防止"惊群效应"。

策略三：不重试

某些工具根本不应该重试：

class NoRetry:
    """不重试策略"""

    def get_delay(self, attempt: int) -> Optional[float]:
        return None

不应该重试的场景：

• 非幂等操作：发送邮件、执行支付、创建订单。重试可能导致重复执行。
• 权限错误：重试不会改变权限状态。
• 参数错误：重试不会修正错误的参数。

重试策略的选择逻辑

def get_retry_strategy(tool_def, error: ToolError):
    """根据工具定义和错误类型选择合适的重试策略"""

    # 永久错误不重试
    if not error.retryable:
        return NoRetry()

    # 工具自身配置了不重试
    if tool_def.max_retries <= 0:
        return NoRetry()

    # 非幂等操作不重试（即使错误是瞬时的）
    if tool_def.category in (ToolCategory.COMMUNICATE, ToolCategory.DANGEROUS):
        if tool_def.metadata.get("idempotent", False) is False:
            return NoRetry()

    # RATE_LIMIT错误使用指数退避（尊重服务端的限流）
    if error.error_type == ErrorType.RATE_LIMIT:
        return ExponentialBackoffRetry(
            max_attempts=tool_def.max_retries,
            base_delay=1.0,
            max_delay=error.retry_after_seconds or 60.0,
        )

    # TIMEOUT错误使用指数退避（服务可能暂时繁忙）
    if error.error_type == ErrorType.TIMEOUT:
        return ExponentialBackoffRetry(
            max_attempts=tool_def.max_retries,
            base_delay=0.5,
            max_delay=10.0,
        )

    # 其他可重试错误使用固定间隔
    return FixedIntervalRetry(
        max_attempts=tool_def.max_retries,
        interval_seconds=1.0,
    )

幂等性：重试安全的关键

幂等性是重试策略中最关键的概念之一。一个操作是幂等的，意味着执行一次和执行多次的效果相同。

• 幂等操作：查询订单状态（查100次结果一样）、更新用户昵称（设置同一个昵称10次效果一样）
• 非幂等操作：发送邮件（发2次用户收到2封）、创建订单（执行2次生成2个订单）、增加余额（加100元执行2次变成加200元）

对于非幂等操作，任何自动重试都是危险的。这就是为什么send_email工具在4.1节中被配置为max_retries=1——它不应该被自动重试。

如果你的工具函数天然不幂等，但你需要某种形式的重试，有两个选择：

选择一：让工具函数幂等。

在设计工具时就考虑幂等性。比如create_order工具可以接受一个idempotency_key参数：

def create_order(order_id: str, product_id: str, quantity: int,
                 idempotency_key: Optional[str] = None) -> str:
    """创建订单（幂等）"""
    # 如果提供了幂等键，先检查是否已存在相同键的订单
    if idempotency_key:
        existing = find_order_by_idempotency_key(idempotency_key)
        if existing:
            return f"订单已存在：{existing.id}"
    # 创建新订单...

选择二：标记为非幂等，禁止自动重试。

在工具定义中明确标记：

@registry.tool(
    description="发送邮件...",
    category=ToolCategory.COMMUNICATE,
    max_retries=0,  # 不自动重试
    metadata={"idempotent": False},  # 标记为非幂等
)
def send_email(to: str, subject: str, body: str) -> str:
    ...

将错误处理集成到执行流程中

更新ToolRegistry.execute方法，集成错误分类和重试：

class ToolRegistry:
    def __init__(self,
                 default_timeout: int = 10,
                 default_max_memory: int = 512,
                 default_max_output: int = 10000,
                 error_classifier: Optional[ErrorClassifier] = None):
        # ... 之前的初始化 ...
        self.error_classifier = error_classifier or default_classifier

    def execute(self, name: str, arguments: Dict[str, Any],
                user_confirmed: bool = False,
                user_role: str = "user") -> ToolResult:
        """执行工具，带错误处理和重试"""
        import time

        start = time.time()

        if name not in self._tools:
            return ToolResult(
                success=False, content="",
                error=f"未知工具：{name}",
                tool_name=name
            )

        tool_def = self._tools[name]

        # 安全检查（第4.3节的内容）
        permission_error = self.permission_manager.check_permission(
            tool_def, arguments, user_confirmed, user_role
        )
        if permission_error:
            return ToolResult(
                success=False, content="",
                error=str(permission_error),
                tool_name=name
            )

        validation_errors = self.validator.validate(name, arguments)
        if validation_errors:
            return ToolResult(
                success=False, content="",
                error=f"参数校验失败：{'; '.join(validation_errors)}",
                tool_name=name
            )

        # 执行（带重试）
        attempt = 0
        last_error: Optional[ToolError] = None

        while True:
            try:
                with self.limiter.time_limit(tool_def.timeout_seconds or self.default_timeout):
                    result = tool_def.func(**arguments)

                result_str = self.limiter.limit_output(str(result))
                duration = int((time.time() - start) * 1000)

                return ToolResult(
                    success=True,
                    content=result_str,
                    tool_name=name,
                    duration_ms=duration,
                    metadata={"attempts": attempt + 1, "retried": attempt > 0}
                )

            except Exception as e:
                # 分类错误
                tool_error = self.error_classifier.classify(e)
                last_error = tool_error
                attempt += 1

                # 选择重试策略
                retry_strategy = get_retry_strategy(tool_def, tool_error)
                delay = retry_strategy.get_delay(attempt - 1)  # attempt-1因为第一次尝试不算重试

                if delay is not None:
                    print(f"  [工具重试] {name} 第{attempt}次尝试失败："
                          f"{tool_error.error_type.value}。"
                          f"将在{delay:.1f}秒后重试（第{attempt + 1}次尝试）")
                    time.sleep(delay)
                else:
                    # 不再重试
                    duration = int((time.time() - start) * 1000)
                    return ToolResult(
                        success=False,
                        content="",
                        error=tool_error.to_llm_message(),
                        tool_name=name,
                        duration_ms=duration,
                        metadata={
                            "attempts": attempt,
                            "error_type": tool_error.error_type.value,
                            "retryable": tool_error.retryable,
                        }
                    )

将错误信息反馈给LLM

工具失败后，错误信息会被放入对话历史，供LLM查看。错误信息的质量直接影响LLM能否有效恢复。

糟糕的错误信息：

工具执行失败：Error: Connection refused

LLM看到这个信息后不知道该做什么。是重试？换工具？还是放弃？

良好的错误信息（使用ToolError.to_llm_message()）：

错误类型：network
错误信息：Connection refused
这是可重试的瞬时错误，可以尝试重试。

更完整的版本还应该包含上下文信息：

def build_error_feedback(call: ToolCallRequest, tool_error: ToolError,
                         attempts: int) -> str:
    """构建给LLM的完整错误反馈"""
    lines = [
        f"工具 '{call.name}' 执行失败。",
        f"调用参数：{json.dumps(call.arguments, ensure_ascii=False)}",
        f"尝试次数：{attempts}",
        "",
        tool_error.to_llm_message(),
        "",
        "你可以选择：",
        "1. 修改参数后重试",
        "2. 使用其他工具完成此任务",
        "3. 告知用户当前无法完成此操作，并说明原因",
    ]
    return "\n".join(lines)

关键设计：错误反馈中包含了调用参数和可选的后续行动。这让LLM不仅能理解"发生了什么"，还能知道"接下来可以怎么做"。

一个完整的错误处理流程示例

用户："帮我查一下订单ORD12345678的状态"

Agent循环：
1. LLM决定调用check_order_status(order_id="ORD12345678")
2. Runtime执行工具
3. 工具执行：网络请求超时
4. 错误分类：TIMEOUT，可重试
5. 重试策略：指数退避，等待0.5秒后重试
6. 第2次尝试：仍然超时
7. 重试策略：等待1.0秒后重试
8. 第3次尝试：仍然超时
9. 重试策略：达到最大次数，停止重试
10. 错误反馈给LLM：
    工具 'check_order_status' 执行失败。
    调用参数：{"order_id": "ORD12345678"}
    尝试次数：3
    错误类型：timeout
    错误信息：连接超时
    这是可重试的瞬时错误...
    你可以选择：
    1. 修改参数后重试
    2. 使用其他工具完成此任务
    3. 告知用户当前无法完成此操作

11. LLM看到错误后决定：
    "抱歉，查询订单状态的服务暂时不可用。这可能是因为网络问题。
    请您稍后再试，或者提供更多信息让我尝试其他查询方式。"

注意LLM在步骤11中的行为：它理解了错误（服务不可用），向用户解释了情况（网络问题），提供了后续选项（稍后再试或换方式）。这就是良好错误反馈的价值。

错误处理的几个设计原则

原则一：区分"工具错误"和"Agent错误"

• 工具错误：工具本身执行失败（网络超时、权限不足、参数不合法）
• Agent错误：Agent的决策有问题（选错了工具、填错了参数、推理偏差）

工具错误可以在Harness层面处理（重试、分类、反馈），Agent错误需要通过反馈让LLM自我纠正。不要试图在Harness层面修复Agent的决策错误——那是LLM的职责。

原则二：错误信息是给LLM看的，不是给开发者看的

工具执行失败时，错误信息会被放入对话历史，由LLM来理解并应对。因此，错误信息应该：

• 使用LLM能理解的语言
• 包含足够的上下文（参数、尝试次数、错误类型）
• 给出可能的后续行动建议

原则三：记录原始错误，展示结构化错误

在日志中记录原始异常（包括堆栈），在对话中展示结构化错误。两者用途不同：原始异常用于开发者调试，结构化错误用于LLM理解和决策。

原则四：失败是信息，不是终点

工具失败后，Agent的循环不应该终止。失败信息应该被反馈给LLM，让它决定下一步。这是Agent优于传统程序的关键特性——传统程序遇到错误就崩溃，Agent遇到错误会想办法。

本节小结

本节构建了完整的工具错误处理与重试系统：

• 错误分类：区分瞬时错误（可重试）和永久错误（不可重试）
• 重试策略：固定间隔、指数退避、不重试三种策略，根据错误类型和工具特性选择
• 幂等性意识：非幂等操作不自动重试
• 错误反馈：结构化错误信息让LLM理解发生了什么并决定下一步

核心原则：错误处理的目标不是"避免所有错误"，而是"让Agent在错误发生时能够优雅地恢复或合理地放弃"。一个好的Harness应该让工具失败成为可管理的常态，而非致命的事件。

在下一节中，我们将实践构建一组完整的工具，将前面讨论的所有原则——定义、Schema、安全、错误处理——应用到实际代码中。