Agent Function Calling 错误分类：网络超时权限与业务异常

一、你的 Agent 把所有工具调用失败都当"网络超时"处理，准确率一路下滑

Agent 调用外部工具（API、数据库、文件系统）的失败模式不是"成功"和"失败"两个状态。真实的失败至少可以分解为四类：网络超时、认证权限错误、参数校验失败、业务逻辑异常。这四类的处理方式完全不同——超时需要重试，权限错误需要中断，参数错误需要反馈修正，业务异常需要降级。

如果把所有错误都当成一种来处理，Agent 的行为会不可预知：权限错误被当作超时重试了三次，每次都返回相同的结果；参数校验失败被当作业务异常，Agent 开始自行"修正"参数然后重新调用；业务异常被当作网络问题，Agent 告诉用户"请稍后重试"但实际上这是数据层面的错误，重试一百次也没用。

Function Calling 的错误分类不是可选的优化，是 Agent 行为可预测性的基础。没有分类，Agent 在错误面前的决策就是随机的。

二、底层机制与原理剖析

错误分类的核心是把工具返回的错误映射到一个标准化的"错误处理决策树"中。系统首先根据 HTTP 状态码和响应体内容判定错误类型，随后触发对应的处理策略。具体流程为：若返回 HTTP 2xx 且数据正常则视为成功；若为 408、504 或连接重置则归类为网络超时；若为 401、403 则归类为权限错误；若为 400、422 则归类为参数校验错误；若为 200 但包含错误体则归类为业务逻辑异常。判定完成后，系统将根据错误类型执行重试、中断、修正或降级等操作。

四类错误的处理原则：

网络超时：可重试，但必须有上限和退避。三次重试、指数退避（1s、2s、4s）。超过三次后视为服务不可用，走降级。

权限错误：不可重试，立即终止。权限错误不是临时性的——再试多少次也不会改成 200。继续重试只会浪费 Token 和时间。

参数校验错误：可修正重试一次。如果 API 返回的 error message 中包含参数修正提示（如"city_code 应为 3 位数字"），可以用这个信息自动修正后重试一次。但如果第二次还是参数错误，说明不是简单的修正能解决的，交给用户处理。

业务逻辑异常：不重试，走降级。比如"账户余额不足"、"库存为 0"——这些是正常的业务流程，不是系统故障。需要把业务错误码映射为用户能理解的消息。

三、生产级代码实现

""" Agent Function Calling 错误分类与处理策略 四类错误：网络超时、权限、参数校验、业务异常 每种错误有独立的退避、重试和降级策略 """ from dataclasses import dataclass, field from typing import Dict, List, Optional, Any, Callable from enum import Enum import time import re import asyncio import aiohttp class ErrorCategory(Enum): """错误大类""" NETWORK_TIMEOUT = "network_timeout" # 网络超时/连接问题 AUTH_PERMISSION = "auth_permission" # 认证/权限 PARAM_VALIDATION = "param_validation" # 参数校验失败 SERVER_ERROR = "server_error" # 服务端内部错误 BUSINESS_LOGIC = "business_logic" # 业务逻辑异常 UNKNOWN = "unknown" # 未分类 class RetryStrategy(Enum): """重试策略""" NO_RETRY = "no_retry" # 不重试 LINEAR_BACKOFF = "linear_backoff" # 线性退避 EXPONENTIAL_BACKOFF = "exponential" # 指数退避 FIXED_INTERVAL = "fixed_interval" # 固定间隔 @dataclass class ErrorPolicy: """单类错误的处理策略""" category: ErrorCategory retry_strategy: RetryStrategy max_retries: int base_delay_ms: int # 基础延迟（毫秒） should_interrupt: bool # 是否中断工具链（不继续调后续工具） @dataclass class ErrorClassification: """错误分类结果""" category: ErrorCategory confidence: float # 分类置信度 0-1 error_code: str # 标准化错误码 original_message: str sanitized_message: str # 净化后的消息（可安全展示给 LLM） retriable: bool auto_fixable: bool # 是否可以自动修正 @dataclass class ToolCallResult: """工具调用结果""" tool_name: str success: bool data: Optional[Any] = None error: Optional[ErrorClassification] = None retry_count: int = 0 elapsed_ms: float = 0.0 # 错误分类规则表 ERROR_CLASSIFICATION_RULES = { # === 网络超时 === "timeout": ErrorCategory.NETWORK_TIMEOUT, "timed out": ErrorCategory.NETWORK_TIMEOUT, "connection refused": ErrorCategory.NETWORK_TIMEOUT, "connection reset": ErrorCategory.NETWORK_TIMEOUT, "no route to host": ErrorCategory.NETWORK_TIMEOUT, "temporary failure in name resolution": ErrorCategory.NETWORK_TIMEOUT, "eof": ErrorCategory.NETWORK_TIMEOUT, "broken pipe": ErrorCategory.NETWORK_TIMEOUT, # === 权限错误 === "unauthorized": ErrorCategory.AUTH_PERMISSION, "forbidden": ErrorCategory.AUTH_PERMISSION, "access denied": ErrorCategory.AUTH_PERMISSION, "permission denied": ErrorCategory.AUTH_PERMISSION, "invalid api key": ErrorCategory.AUTH_PERMISSION, "token expired": ErrorCategory.AUTH_PERMISSION, "insufficient scope": ErrorCategory.AUTH_PERMISSION, # === 参数校验 === "invalid parameter": ErrorCategory.PARAM_VALIDATION, "missing required": ErrorCategory.PARAM_VALIDATION, "validation error": ErrorCategory.PARAM_VALIDATION, "bad request": ErrorCategory.PARAM_VALIDATION, "unprocessable entity": ErrorCategory.PARAM_VALIDATION, "invalid format": ErrorCategory.PARAM_VALIDATION, # === 服务端错误 === "internal server error": ErrorCategory.SERVER_ERROR, "service unavailable": ErrorCategory.SERVER_ERROR, "bad gateway": ErrorCategory.SERVER_ERROR, } # 各类错误的处理策略 DEFAULT_POLICIES = { ErrorCategory.NETWORK_TIMEOUT: ErrorPolicy( category=ErrorCategory.NETWORK_TIMEOUT, retry_strategy=RetryStrategy.EXPONENTIAL_BACKOFF, max_retries=3, base_delay_ms=1000, should_interrupt=False, ), ErrorCategory.AUTH_PERMISSION: ErrorPolicy( category=ErrorCategory.AUTH_PERMISSION, retry_strategy=RetryStrategy.NO_RETRY, max_retries=0, base_delay_ms=0, should_interrupt=True, # 权限错误：立即停止工具链 ), ErrorCategory.PARAM_VALIDATION: ErrorPolicy( category=ErrorCategory.PARAM_VALIDATION, retry_strategy=RetryStrategy.FIXED_INTERVAL, max_retries=1, # 仅重试一次（修正参数后） base_delay_ms=0, should_interrupt=False, ), ErrorCategory.SERVER_ERROR: ErrorPolicy( category=ErrorCategory.SERVER_ERROR, retry_strategy=RetryStrategy.LINEAR_BACKOFF, max_retries=2, base_delay_ms=2000, should_interrupt=False, ), ErrorCategory.BUSINESS_LOGIC: ErrorPolicy( category=ErrorCategory.BUSINESS_LOGIC, retry_strategy=RetryStrategy.NO_RETRY, max_retries=0, base_delay_ms=0, should_interrupt=False, ), } class ErrorClassifier: """错误分类器 基于规则匹配 + 关键词提取，将工具返回的错误归类。 规则是声明式的可维护的表格，不是硬编码的 if-else。 """ def classify(self, error: Exception, http_status: Optional[int] = None) -> ErrorClassification: """分类一个错误""" error_str = str(error).lower() error_type = type(error).__name__ # Step 1: HTTP 状态码分类 if http_status: if http_status == 401 or http_status == 403: return self._build_result( ErrorCategory.AUTH_PERMISSION, 0.95, f"HTTP_{http_status}", str(error), "认证或权限不足" ) if http_status == 400 or http_status == 422: return self._build_result( ErrorCategory.PARAM_VALIDATION, 0.85, f"HTTP_{http_status}", str(error), "请求参数有误" ) if http_status >= 500: return self._build_result( ErrorCategory.SERVER_ERROR, 0.8, f"HTTP_{http_status}", str(error), "服务端内部错误" ) # Step 2: 异常类型分类 if isinstance(error, asyncio.TimeoutError): return self._build_result( ErrorCategory.NETWORK_TIMEOUT, 0.95, "TIMEOUT", str(error), "请求超时" ) if isinstance(error, (ConnectionError, ConnectionRefusedError, ConnectionResetError)): return self._build_result( ErrorCategory.NETWORK_TIMEOUT, 0.9, "CONNECTION_ERROR", str(error), "网络连接异常" ) # Step 3: 关键词匹配 for keyword, category in ERROR_CLASSIFICATION_RULES.items(): if keyword in error_str: return self._build_result(category, 0.8, keyword.upper().replace(" ", "_"), str(error), "") # Step 4: 降级为 UNKNOWN return self._build_result( ErrorCategory.UNKNOWN, 0.3, "UNKNOWN", str(error), "未分类的错误" ) def _build_result( self, category: ErrorCategory, confidence: float, error_code: str, original: str, sanitized: str ) -> ErrorClassification: policy = DEFAULT_POLICIES.get(category) return ErrorClassification( category=category, confidence=confidence, error_code=error_code, original_message=original, sanitized_message=sanitized or f"{category.value}", retriable=policy.retry_strategy != RetryStrategy.NO_RETRY if policy else False, auto_fixable=(category == ErrorCategory.PARAM_VALIDATION), ) class RetryExecutor: """重试执行器 根据 ErrorPolicy 执行不同策略的重试。 """ def __init__(self): self.classifier = ErrorClassifier() async def execute_with_retry( self, tool_name: str, call_fn: Callable, *args, **kwargs ) -> ToolCallResult: """带错误分类的重试执行""" start_time = time.time() retry_count = 0 last_error = None while True: try: result = await call_fn(*args, **kwargs) elapsed = (time.time() - start_time) * 1000 return ToolCallResult( tool_name=tool_name, success=True, data=result, retry_count=retry_count, elapsed_ms=elapsed, ) except Exception as e: retry_count += 1 last_error = e # 提取 HTTP 状态码（如果有） http_status = getattr(e, 'status', None) or getattr(e, 'status_code', None) # 分类错误 classification = self.classifier.classify(e, http_status) policy = DEFAULT_POLICIES.get(classification.category) if not policy: break # 未知错误不重试 # 权限错误：立即中断 if policy.should_interrupt: break # 达到最大重试次数 if retry_count > policy.max_retries: break # 参数校验：尝试自动修正 if classification.auto_fixable and retry_count == 1: kwargs = self._attempt_auto_fix(e, kwargs) if kwargs is None: break # 无法自动修正 # 计算退避延迟 delay = self._calculate_delay(policy, retry_count) await asyncio.sleep(delay / 1000.0) # 所有重试已耗尽 elapsed = (time.time() - start_time) * 1000 classification = self.classifier.classify( last_error or Exception("unknown"), getattr(last_error, 'status', None) if last_error else None ) return ToolCallResult( tool_name=tool_name, success=False, error=classification, retry_count=retry_count, elapsed_ms=elapsed, ) def _calculate_delay(self, policy: ErrorPolicy, retry_count: int) -> int: """计算退避延迟""" base = policy.base_delay_ms if policy.retry_strategy == RetryStrategy.EXPONENTIAL_BACKOFF: return base * (2 ** (retry_count - 1)) elif policy.retry_strategy == RetryStrategy.LINEAR_BACKOFF: return base * retry_count elif policy.retry_strategy == RetryStrategy.FIXED_INTERVAL: return base return 0 def _attempt_auto_fix(self, error: Exception, kwargs: dict) -> Optional[dict]: """尝试从错误消息中自动修正参数""" # 简化实现：实际应使用 LLM 来解析错误消息中的修正建议 # 例如：API 返回 {"error": "city_code must be 3 digits, got '12'"} # LLM 可以解析出正确的 city_code 应该是 "012" return None # 暂时不支持自动修正 def build_response_for_llm(self, result: ToolCallResult) -> str: """为 LLM 构建工具调用的响应消息""" if result.success: return f"工具 [{result.tool_name}] 调用成功。" if result.error: cat = result.error.category if cat == ErrorCategory.NETWORK_TIMEOUT: return f"工具 [{result.tool_name}] 网络超时，已重试 {result.retry_count} 次。请稍后重试或跳过此步骤。" elif cat == ErrorCategory.AUTH_PERMISSION: return f"工具 [{result.tool_name}] 权限不足，无法继续。请检查 API Key 或权限配置。" elif cat == ErrorCategory.PARAM_VALIDATION: return f"工具 [{result.tool_name}] 参数校验失败：{result.error.sanitized_message}。请检查调用参数。" elif cat == ErrorCategory.SERVER_ERROR: return f"工具 [{result.tool_name}] 服务端异常，已重试 {result.retry_count} 次。暂时不可用。" else: return f"工具 [{result.tool_name}] 调用失败。" return f"工具 [{result.tool_name}] 调用失败。"

四、边界分析与架构权衡

错误分类的准确性边界：

基于关键词匹配的分类精度在 80-85% 左右。有些 API 返回的错误消息比较模糊（如"Something went wrong"），分类器无法区分是超时还是服务端错误。另外，有些 API 的 4xx 错误信息中包含了可修正的提示——比如 OpenAI 的 429（限流）需要用不同的退避策略处理。分类器需要持续根据 API 特性做规则调整。

参数自动修正的风险：

自动修正参数是一个有风险的操作。修正后的参数可能引入新的错误。比如 API 返回"city_code 应为 3 位数字，当前为 '12'"，自动修正为 "012" 看起来正确——但如果 API 期望的 city_code 根本不是 3 位只是恰好名字相似，自动修正反而带来了更大的偏差。建议自动修正在内部环境或开发阶段使用，生产环境由 Agent 提示用户确认。

适用边界：

最适合调用多个不同外部 API 的 Agent。API 种类越多，标准化的错误分类越有价值——避免了为每个 API 单独实现错误处理逻辑。

禁用场景：

不适合只有一两个确定性工具的简单 Agent。也不适合通过中间件（如 API Gateway）已经在入口层统一处理了错误的场景。

五、总结

Function Calling 的错误处理不能一刀切。四类错误对应四种策略：网络超时用指数退避重试（3 次上限）、权限错误立即中断（不能把 403 当 504 处理）、参数校验可尝试自动修正后重试一次、服务端异常可选降级。分类不是用于展示给用户的，而是用于指导 Agent 的下一个动作——是重试、是中断、还是降级。