Agent Function Calling 错误分类:网络超时权限与业务异常 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( categoryErrorCategory.NETWORK_TIMEOUT, retry_strategyRetryStrategy.EXPONENTIAL_BACKOFF, max_retries3, base_delay_ms1000, should_interruptFalse, ), ErrorCategory.AUTH_PERMISSION: ErrorPolicy( categoryErrorCategory.AUTH_PERMISSION, retry_strategyRetryStrategy.NO_RETRY, max_retries0, base_delay_ms0, should_interruptTrue, # 权限错误立即停止工具链 ), ErrorCategory.PARAM_VALIDATION: ErrorPolicy( categoryErrorCategory.PARAM_VALIDATION, retry_strategyRetryStrategy.FIXED_INTERVAL, max_retries1, # 仅重试一次修正参数后 base_delay_ms0, should_interruptFalse, ), ErrorCategory.SERVER_ERROR: ErrorPolicy( categoryErrorCategory.SERVER_ERROR, retry_strategyRetryStrategy.LINEAR_BACKOFF, max_retries2, base_delay_ms2000, should_interruptFalse, ), ErrorCategory.BUSINESS_LOGIC: ErrorPolicy( categoryErrorCategory.BUSINESS_LOGIC, retry_strategyRetryStrategy.NO_RETRY, max_retries0, base_delay_ms0, should_interruptFalse, ), } 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, fHTTP_{http_status}, str(error), 认证或权限不足 ) if http_status 400 or http_status 422: return self._build_result( ErrorCategory.PARAM_VALIDATION, 0.85, fHTTP_{http_status}, str(error), 请求参数有误 ) if http_status 500: return self._build_result( ErrorCategory.SERVER_ERROR, 0.8, fHTTP_{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( categorycategory, confidenceconfidence, error_codeerror_code, original_messageoriginal, sanitized_messagesanitized or f{category.value}, retriablepolicy.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_nametool_name, successTrue, dataresult, retry_countretry_count, elapsed_mselapsed, ) 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_nametool_name, successFalse, errorclassification, retry_countretry_count, elapsed_mselapsed, ) 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 的下一个动作——是重试、是中断、还是降级。

本月热点