趋近智
集成外部工具和API能极大地增强智能体的功能,使其能与动态信息源交互,并执行超出大型语言模型内部知识范围的操作。然而,这种交互也带来了潜在的故障点。网络连接可能不稳定,API可能无预警变更,服务可能出现停机,速率限制可能被突破,智能体本身也可能生成无效请求。构建可靠的智能体系统,需要将错误处理视为核心设计原则,而非可有可无的附加项。忽视这一点会导致智能体脆弱不堪,在遇到外部依赖的必然缺陷时,会不可预测地发生故障。
本节细述了方法,用于在工具执行过程中检测、管理和从故障中恢复,让智能体能更可靠、更具适应性地运行。
在实施处理机制前,识别潜在故障的不同特点是很重要的:
网络问题: 这些问题通常是暂时的,与智能体和API端点之间的连接有关。示例包括:
ConnectionRefusedError)。API服务器错误(HTTP 5xx): 这些表明托管API的服务器端存在问题。
500 内部服务器错误:一个通用服务器错误。502 错误网关:通常与代理或网关问题一同出现。503 服务不可用:服务器暂时过载或因维护而停机。504 网关超时:作为网关的服务器未从上游服务器收到及时响应。客户端错误(HTTP 4xx): 这些表明智能体发送的请求存在问题。
400 错误请求:请求格式不正确或包含无效语法。通常由大型语言模型生成不正确参数引起。401 未经授权:缺少或无效的认证凭据。403 禁止访问:认证成功,但已认证用户缺少请求资源的权限。404 未找到:请求的资源不存在。429 请求过多:智能体已超出API的速率限制。数据处理错误: 在接收到响应后发生的故障。
智能体生成的无效输入: 大型语言模型本身可能为工具调用生成逻辑上不正确或格式错误的参数,这会导致可预测的API错误(通常是400 Bad Request)或意外的工具行为。
有效的处理始于可靠的检测。在您的工具执行封装器中实现检测逻辑:
try-except块: 将外部调用(网络请求、数据解析)封装在try-except块中,以捕获诸如requests.exceptions.Timeout、requests.exceptions.ConnectionError、json.JSONDecodeError 或自定义API客户端异常。一旦检测到故障,智能体或其控制系统需要一个响应方法。
重试对于网络瞬时故障或临时服务器不可用(5xx错误)等瞬时问题有效。
401 未经授权或400 错误请求进行重试通常是徒劳的。重试最适用于超时、连接错误、429 请求过多和5xx服务器错误。import time
import random
import requests
def execute_api_call_with_retry(url, params, headers, max_attempts=5, base_delay=1.0, factor=2.0, jitter=0.1):
"""
执行带指数退避和抖动的API GET请求。仅在特定瞬时错误时重试。
"""
attempts = 0
while attempts < max_attempts:
attempts += 1
try:
response = requests.get(url, params=params, headers=headers, timeout=10) # 示例超时
# 成功或不可重试的客户端错误
if response.status_code < 500 and response.status_code != 429:
response.raise_for_status() # 立即为 4xx 状态码抛出 HTTPError
return response.json() # 或者返回响应对象
# 可重试的服务器错误或速率限制
if response.status_code >= 500 or response.status_code == 429:
print(f"Attempt {attempts}: Received status {response.status_code}. Retrying...")
# 继续执行重试逻辑
except requests.exceptions.Timeout:
print(f"Attempt {attempts}: Request timed out. Retrying...")
# 继续执行重试逻辑
except requests.exceptions.ConnectionError as e:
print(f"Attempt {attempts}: Connection error ({e}). Retrying...")
# 继续执行重试逻辑
except requests.exceptions.RequestException as e:
# 捕获其他与请求相关的异常(例如,由 raise_for_status 抛出的 4xx HTTPError)
print(f"Attempt {attempts}: Non-retriable request error: {e}")
raise # 立即重新抛出异常
if attempts < max_attempts:
# 计算带指数退避和抖动的延迟
delay = base_delay * (factor ** (attempts - 1))
delay = random.uniform(delay - delay * jitter, delay + delay * jitter)
print(f"Waiting {delay:.2f} seconds before next attempt.")
time.sleep(delay)
else:
print(f"Attempt {attempts}: Max retries reached.")
# 选项:抛出自定义异常或返回特定的错误指示
raise MaxRetriesExceededError(f"API call failed after {max_attempts} attempts.")
# 如果抛出了 MaxRetriesExceededError,则不应到达此处
return None
class MaxRetriesExceededError(Exception):
pass
# 示例用法(请替换为实际API详情)
# try:
# data = execute_api_call_with_retry("https://api.example.com/data", params={"id": 123}, headers={"Authorization": "Bearer token"})
# # 处理数据
# except MaxRetriesExceededError as e:
# # 处理重试后的最终故障
# print(e)
# except requests.exceptions.HTTPError as e:
# # 处理不可重试的 4xx 错误
# print(f"Client error: {e.response.status_code}")
# except Exception as e:
# # 处理其他意外错误
# print(f"An unexpected error occurred: {e}")
此Python函数示例展示了API调用执行,其中包含指数退避、抖动以及基于HTTP状态码和请求异常的条件重试。
当重试失败或不适用时,考虑替代操作:
重要的是,故障信息通常必须反馈给大型语言模型核心,以便其进行智能适应:
对于反复发生故障的工具,持续重试会浪费资源和大型语言模型上下文。断路器模式提供了一种更有条理的方法:
实现断路器通常涉及在直接工具调用之外维护状态,这通常在智能体框架或专门的工具管理服务中进行。
断路器模式中的状态转换,用于管理频繁发生故障的工具交互。
execute_api_call_with_retry 示例)、架构验证,并标准化返回给智能体控制器的成功和错误输出格式。requests-mock等库)来模拟不同的故障场景(超时、特定状态码、格式错误的响应),并验证您的重试、备用和错误报告机制按预期工作。考虑在预生产环境中进行故障注入测试。通过系统地处理外部交互中潜在的故障,您可以构建出在动态环境中可靠执行复杂多步任务的、更具韧性和能力的智能体系统。这种韧性是生产就绪型智能体应用的标志。
简洁的语法。内置调试功能。从第一天起就可投入生产。
为 ApX 背后的 AI 系统而构建
这部分内容有帮助吗?
© 2026 ApX Machine Learning用心打造