深入自修正机制：当智能体发现工具报错时，如何自动重试并修正参数
各位同仁，大家好。
在构建智能体的过程中，我们常常追求其自主决策、自主执行的能力。然而，现实世界复杂多变，智能体所依赖的工具 API 并非总是完美无缺。网络波动、参数错误、权限不足、外部服务故障，这些都是工具调用中常见的“拦路虎”。当智能体在执行任务时，如果一个工具调用失败，仅仅简单地报告错误并终止任务，无疑会大大降低其可用性和鲁棒性。
今天的讲座，我们将深入探讨一个核心问题：当智能体发现工具报错时，如何让它自动重试并智能地修正参数？ 这不仅仅是简单的错误处理，更是一种高级的自修正能力，它赋予了智能体从失败中学习、适应并最终完成任务的韧性。我们将从错误类型识别、重试策略、参数修正逻辑，到最终的智能体架构实现，层层剖析，并辅以详尽的代码示例。
1. 智能体与工具的协同挑战
智能体（Agent）通常被设计为能够理解用户意图、规划任务、并利用一系列工具（Tools）来执行这些任务的实体。这些工具可以是查询数据库、发送邮件、调用外部API、执行代码等。智能体与工具的协同工作流通常如下：
- 理解与规划：智能体接收任务，分解为子任务，并决定需要调用哪些工具。
- 工具选择与参数生成：根据子任务，智能体选择合适的工具，并根据当前上下文生成调用该工具所需的参数。
- 工具执行：智能体调用工具执行器，将生成的参数传递给工具。
- 结果解析与反馈：工具返回执行结果，智能体解析结果并决定下一步行动。
然而，在第3步“工具执行”环节，各种意外情况层出不穷。一个健壮的智能体，必须能够优雅地处理这些异常，甚至能够主动地纠正错误，而不是仅仅将问题抛给用户。这正是“自修正”机制的核心价值所在。
2. 工具故障的解剖学：错误类型与识别
要实现智能的自修正，首先要做的就是理解错误的本质。不同的错误类型需要不同的修正策略。我们可以将常见的工具错误归纳为以下几类：
-
瞬时性错误（Transient Errors）：
- 特点：通常与网络波动、服务暂时不可用、并发限制（Rate Limit）等有关。这些错误在短时间内可能会自动恢复。
- 示例：HTTP 503 Service Unavailable, Connection Timeout, API Rate Limit Exceeded。
- 修正策略：简单重试（Simple Retry）是首选，通常采用指数退避（Exponential Backoff）策略。
-
参数验证错误（Validation Errors）：
- 特点：智能体提供的参数不符合工具预期的格式、类型、范围或枚举值。
- 示例：
{"error": "Invalid parameter 'category', expected one of ['electronics', 'books', 'clothing'] but received 'food'"}，{"error": "Parameter 'max_results' must be an integer, got 'ten'"}，{"error": "Missing required parameter 'query'"}。 - 修正策略：需要智能体解析错误信息，识别问题参数，并根据错误提示和工具规范重新生成或修改参数。这通常需要LLM的参与。
-
语义错误（Semantic Errors）：
- 特点：参数在语法上是正确的，但在业务逻辑上是无效的。工具能够接收参数，但无法完成操作。
- 示例：
{"error": "Product with ID '12345' not found"}，{"error": "Unauthorized access to resource 'reports'"}，{"error": "File 'report.pdf' does not exist"}。 - 修正策略：比参数验证错误更复杂。可能需要LLM重新思考原始意图，或者执行其他工具来获取正确的信息（例如，在“文件不存在”时，先调用“列出文件”工具）。
-
服务内部错误（Internal Service Errors）：
- 特点：工具服务自身发生的未预期错误，通常是后端代码缺陷或基础设施问题。
- 示例：HTTP 500 Internal Server Error，
{"error": "An unexpected error occurred on the server side."}。 - 修正策略：通常无法通过修改参数解决。可以重试，如果持续失败，则可能需要报告给用户或尝试其他工具。
为了让智能体能够智能地处理这些错误，我们必须能够从工具返回的原始错误信息中提取出有用的结构化数据。理想情况下，工具应该返回结构化的错误响应（如JSON），包含错误码、错误消息、以及可能的问题参数字段。
| 错误类型 | 典型描述 | 修正策略 | LLM 参与 |
|---|---|---|---|
| 瞬时性错误 | 网络超时、服务暂时不可用、限流 | 延迟重试（指数退避） | 否 |
| 参数验证错误 | 参数类型不匹配、格式错误、值超出范围、缺失参数 | 解析错误，识别问题参数，LLM修正参数 | 是 |
| 语义错误 | ID不存在、权限不足、资源状态不正确 | 解析错误，LLM重新思考意图，可能调用其他工具获取信息 | 是 |
| 服务内部错误 | 服务器内部逻辑错误 | 有限次重试，如果持续失败，报告或尝试替代方案 | 有限 |
3. 初级自修正：基于策略的重试机制
对于瞬时性错误，最直接有效的策略就是重试。然而，盲目地立即重试往往只会加剧问题（例如，对限流API进行即时重试会更快达到上限）。因此，我们需要一个智能的重试机制。
指数退避（Exponential Backoff）是一种常用的策略。它在每次失败后等待更长的时间再进行重试，从而给服务恢复或限流解除留出时间。
import time
import random
from functools import wraps
def retry_with_exponential_backoff(max_attempts: int = 3, initial_delay: float = 1.0, max_delay: float = 60.0,
factor: float = 2.0, jitter: bool = True):
"""
一个装饰器，用于在函数执行失败时进行指数退避重试。
Args:
max_attempts (int): 最大重试次数。
initial_delay (float): 第一次重试前的初始等待时间（秒）。
max_delay (float): 最大等待时间（秒）。
factor (float): 每次重试时延迟增加的因子。
jitter (bool): 是否添加随机抖动，以避免“惊群效应”。
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
print(f"尝试 {attempt + 1}/{max_attempts} 失败: {e}")
if attempt + 1 == max_attempts:
raise # 最后一次尝试失败，向上抛出异常
sleep_time = delay
if jitter:
sleep_time = delay * (1 + random.uniform(-0.25, 0.25)) # 添加 +/- 25% 抖动
print(f"等待 {sleep_time:.2f} 秒后重试...")
time.sleep(min(sleep_time, max_delay))
delay *= factor
return func(*args, **kwargs) # 理论上不会执行到这里，因为会先抛出异常
return wrapper
return decorator
# 示例：一个可能随机失败的工具
class MockService:
def __init__(self):
self.call_count = 0
@retry_with_exponential_backoff(max_attempts=5, initial_delay=0.5, jitter=True)
def fetch_data(self, item_id: str):
self.call_count += 1
print(f"--- 尝试调用 fetch_data({item_id}), 第 {self.call_count} 次 ---")
if self.call_count % 3 != 0 and random.random() < 0.7: # 模拟70%的瞬时失败率，但每3次会成功一次
if random.random() < 0.5:
raise ConnectionError(f"网络连接中断，无法获取 item {item_id}")
else:
raise TimeoutError(f"请求超时，无法获取 item {item_id}")
print(f"--- 成功获取 item {item_id} ---")
return {"id": item_id, "data": "some_data_payload"}
# 使用示例
if __name__ == "__main__":
service = MockService()
try:
result = service.fetch_data("product_123")
print(f"最终结果: {result}")
except (ConnectionError, TimeoutError) as e:
print(f"多次重试后仍然失败: {e}")
print("n--- 再次尝试不同的ID ---")
service = MockService() # 重置计数器
try:
result = service.fetch_data("user_abc")
print(f"最终结果: {result}")
except (ConnectionError, TimeoutError) as e:
print(f"多次重试后仍然失败: {e}")代码解析：
retry_with_exponential_backoff是一个装饰器，可以方便地应用于任何可能失败的工具函数。- 它通过一个循环来管理重试次数。
time.sleep()实现延迟，delay *= factor实现指数增长。jitter参数通过添加随机抖动来分散重试请求，避免所有客户端在同一时刻重试，这在分布式系统中尤为重要。- 当达到最大尝试次数后，如果仍然失败，则重新抛出原始异常。
这种重试机制对于处理瞬时性、临时性的错误非常有效，但对于参数错误或语义错误则无能为力，因为错误的根源在于输入本身，而不是外部环境。
4. 核心机制：参数修正与LLM的智能决策
现在，我们进入讲座的核心部分：如何让智能体在遇到参数错误时，不仅重试，还能智能地修正参数。这正是大语言模型（LLM）大显身手的地方。LLM的强大理解、推理和生成能力，使其成为解析错误消息、推断正确参数的理想选择。
4.1 LLM在参数修正中的角色
当工具返回参数相关的错误时，LLM需要完成以下任务：
- 错误信息解析与理解：将工具返回的原始、有时晦涩的错误消息，转化为对其根源的清晰理解。
- 问题参数识别：准确指出哪个或哪些参数导致了问题。
- 参数修正建议：根据错误信息、工具的预期（如果知道）和上下文，生成新的、正确的参数值。
- 修正理由说明：解释为什么提出这些修正，这有助于智能体或开发者理解决策过程。
4.2 LLM的输入：构建修正提示
为了让LLM高效地完成这些任务，我们需要精心设计传递给它的提示（Prompt）。一个好的提示应该包含足够的信息，并明确指示LLM的输出格式。
核心信息要素：
- 当前任务目标：智能体最初想做什么？这提供了修正的宏观上下文。
- 原始工具调用信息：工具名称、原始参数，让LLM知道“之前做了什么”。
- 工具返回的错误信息：这是修正的关键线索，必须完整且清晰地提供。
- 工具的Schema或描述（可选但强烈推荐）：如果知道工具的输入规范，将其提供给LLM，能极大地提高修正的准确性。这包括参数名、类型、枚举值、范围、是否必需等。
- 历史尝试记录（可选）：如果已经尝试过几次修正，将之前的尝试和错误也提供给LLM，帮助它避免重复错误。
- 期望的输出格式：明确告诉LLM以结构化的格式（如JSON）返回修正后的参数和解释。
Prompt 示例结构：
**系统提示（System Prompt）**:
你是一个智能体辅助系统，专门负责分析工具调用失败的原因，并根据错误信息智能地修正工具的参数。你的目标是帮助智能体成功执行其任务。
请仔细阅读提供的工具调用信息、原始错误消息以及可选的工具描述。
你的输出必须是JSON格式，包含'corrected_params'（修正后的参数字典）和'explanation'（修正理由）。
如果无法修正，或者认为原始参数已经正确，请返回一个空字典作为'corrected_params'。
**用户提示（User Prompt）**:
当前任务目标: {task_goal}
原始工具调用信息:
工具名称: {tool_name}
原始参数: {original_tool_args}
工具返回的错误信息:
{error_message}
工具描述（Schema，如果可用）:
{tool_schema_or_description}
请分析错误，推断正确的参数，并以JSON格式返回修正后的参数和修正理由。
JSON格式示例:
{{
"corrected_params": {{
"parameter_name_1": "new_value_1",
"parameter_name_2": "new_value_2"
}},
"explanation": "根据错误信息，参数'X'的类型不正确，已将其转换为整数。"
}}示例分析：
假设一个 search_products 工具期望 category 是一个枚举值，max_results 是一个整数，但智能体错误地提供了 category="food" 和 max_results="ten"。
- 任务目标：
帮助用户找到关于电子产品的前5个结果 - 原始调用：
search_products(query="laptop", category="food", max_results="ten") - 错误信息：
{"error": "Invalid value 'food' for parameter 'category'. Expected one of ['electronics', 'books', 'clothing']. Also, 'max_results' must be an integer, but received 'ten'."} - 工具Schema（简化版）：
{ "name": "search_products", "description": "搜索商品", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"}, "category": {"type": "string", "enum": ["electronics", "books", "clothing"], "description": "商品类别"}, "max_results": {"type": "integer", "description": "最大返回结果数量"} }, "required": ["query", "category"] } }
LLM接收到这些信息后，应该能够：
- 识别
category的值food不在enum列表中。 - 识别
max_results的值ten不是整数。 - 根据任务目标和
enum列表，建议将category改为electronics。 - 根据任务目标和类型要求，建议将
max_results改为5。
并输出类似：
{
"corrected_params": {
"category": "electronics",
"max_results": 5
},
"explanation": "错误信息指出'category'参数为无效值'food'，且'max_results'应为整数。根据任务目标和工具Schema，已将'category'修正为'electronics'，'max_results'修正为整数5。"
}4.3 LLM接口模拟
为了在没有真实LLM服务的情况下演示，我们可以创建一个模拟的LLM客户端。
import json
from typing import Dict, Any, Optional
class MockLLMClient:
"""
模拟一个LLM客户端，用于根据错误信息生成参数修正建议。
在实际应用中，这里会集成OpenAI, Google Gemini等LLM API。
"""
def __init__(self):
print("MockLLMClient initialized.")
def _parse_error_for_mock(self, error_message: str) -> Dict[str, Any]:
"""
一个简化的错误解析逻辑，用于模拟LLM的推理。
实际LLM会通过语言理解来完成。
"""
corrected_params = {}
explanation_parts = []
if "Invalid value 'food' for parameter 'category'" in error_message:
corrected_params["category"] = "electronics"
explanation_parts.append("'category'参数值'food'无效，已修正为'electronics'。")
elif "Missing required parameter 'category'" in error_message:
corrected_params["category"] = "default_category" # 假设一个默认值
explanation_parts.append("缺失必填参数'category'，已提供默认值。")
if "'max_results' must be an integer" in error_message:
# 假设LLM能从任务目标或上下文推断出期望的数值
# 这里简化为直接修正为5
corrected_params["max_results"] = 5
explanation_parts.append("'max_results'参数应为整数，已修正为5。")
elif "max_results' must be less than or equal to 10" in error_message:
# 假设LLM能根据上下文或Schema推断出更合理的值
corrected_params["max_results"] = 10
explanation_parts.append("'max_results'超出最大限制，已修正为10。")
if "Product with ID '12345' not found" in error_message:
# 语义错误，这里模拟LLM建议重新思考
explanation_parts.append("商品ID '12345'未找到。建议智能体重新确认商品ID或尝试搜索。")
# 此时corrected_params可能为空，或者建议调用其他工具
return {
"corrected_params": corrected_params,
"explanation": " ".join(explanation_parts) if explanation_parts else "未检测到可自动修正的参数错误。"
}
def get_correction_suggestion(self,
task_goal: str,
tool_name: str,
original_tool_args: Dict[str, Any],
error_message: str,
tool_schema: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
根据提供的上下文和错误信息，生成参数修正建议。
"""
# 在真实的LLM调用中，我们会将上述所有信息格式化为Prompt
# 并发送给LLM API，然后解析其JSON响应。
# 这里我们直接调用内部模拟逻辑。
print(f"n--- 调用 MockLLMClient 进行参数修正 ---")
print(f"任务目标: {task_goal}")
print(f"工具: {tool_name}, 原始参数: {original_tool_args}")
print(f"错误信息: {error_message}")
# print(f"工具Schema: {tool_schema}") # 模拟LLM内部使用Schema
# 实际LLM会理解这些信息并生成修正
correction_result = self._parse_error_for_mock(error_message)
print(f"MockLLMClient 建议: {json.dumps(correction_result, indent=2, ensure_ascii=False)}")
return correction_result
这个 MockLLMClient 内部的 _parse_error_for_mock 是一个硬编码的规则集，它模拟了LLM通过模式匹配和上下文理解来生成修正建议的过程。在实际生产环境中，这部分逻辑将完全由真实的LLM来完成，它的智能程度远超这种硬编码规则。
5. 构建自修正智能体架构
为了集成上述的重试和参数修正机制，我们需要一个能够管理任务状态、执行工具、捕获错误并触发自修正流程的智能体架构。
5.1 智能体核心组件
一个自修正智能体可以分解为以下核心组件：
-
Agent Core (智能体核心)：
- 接收用户任务。
- 维护任务状态（当前目标、已完成步骤、失败历史等）。
- 根据任务规划，决定下一步要调用的工具。
- 协调其他组件。
-
Tool Executor (工具执行器)：
- 负责实际调用注册的工具。
- 捕获工具执行过程中抛出的异常。
- 将错误信息标准化，传递给 Error Handler。
-
Error Handler (错误处理器)：
- 接收来自 Tool Executor 的错误信息。
- 根据错误类型决定修正策略（简单重试或LLM参数修正）。
- 管理重试次数和修正尝试次数。
-
LLM Interface (LLM接口)：
- 封装与大语言模型交互的逻辑。
- 负责构建Prompt、发送请求、解析LLM的响应。
-
State Manager (状态管理器)：
- 存储和检索智能体在任务执行过程中的所有相关状态。
- 包括工具调用历史、错误记录、当前重试/修正计数等。
5.2 智能体控制流
自修正智能体的控制流将比传统智能体更加复杂，它引入了错误处理和修正的循环：
- 任务初始化：Agent Core 接收任务，初始化 State Manager。
- 规划与工具选择：Agent Core 根据任务目标，决定调用哪个工具，并生成初始参数。
- 工具执行尝试循环：
- 执行工具：Tool Executor 尝试调用工具。
- 错误捕获：如果工具调用成功，返回结果给 Agent Core。如果失败，Tool Executor 捕获异常并传递给 Error Handler。
- 错误处理与决策：Error Handler 根据错误类型：
- 瞬时性错误：检查是否达到最大重试次数。
- 未达到：应用指数退避策略，短暂等待后，再次回到“执行工具”步骤。
- 已达到：报告最终失败，或通知 Agent Core 尝试其他策略。
- 参数/语义错误：检查是否达到最大参数修正尝试次数。
- 未达到：调用 LLM Interface，将任务目标、原始参数、错误信息、工具 Schema 等传递给 LLM。
- LLM返回修正建议（
corrected_params）。 - Error Handler 将
original_params与corrected_params合并，形成新的参数。 - 再次回到“执行工具”步骤，使用新参数重试。
- 已达到：报告最终失败，或通知 Agent Core 尝试其他策略。
- 瞬时性错误：检查是否达到最大重试次数。
- 成功或最终失败：如果工具调用最终成功，Agent Core 继续下一步规划。如果最终失败，Agent Core 向上报告或寻求人工干预。
简化流程图（文字描述）：
[开始任务]
|
V
[Agent Core: 规划工具调用]
| (tool_name, initial_args)
V
[Tool Executor: 尝试调用工具]
|
+---[成功]---> [Agent Core: 处理结果，继续任务]
|
+---[失败 (抛出异常)]---> [Error Handler: 捕获并解析错误]
|
V
[错误类型判断]
|
+---[瞬时性错误]---> [是否达到最大重试次数?]
| |
| +---[否]---> [延迟重试 (指数退避)]
| | |
| | V
| +---[是]---> [报告任务失败或切换工具]
| ^
V |
[参数/语义错误]---> [是否达到最大参数修正次数?]
|
+---[否]---> [LLM Interface: 获取修正建议]
| | (original_args, error_msg, schema)
| V
| [LLM: 生成 corrected_params]
| |
V [Error Handler: 应用修正，更新参数]
| |
+---[是]---> [报告任务失败或切换工具]6. 实践演练：一个自修正智能体的实现细节
现在，让我们通过具体的Python代码来实现这个自修正智能体。
6.1 模拟工具：ProductSearchTool
我们首先定义一个模拟的工具，它会根据输入的参数模拟不同类型的错误。
# tool_definitions.py
from typing import Dict, Any, List
class ProductSearchTool:
"""
一个模拟的商品搜索工具，用于演示不同类型的错误。
"""
name = "search_products"
description = "根据关键词和类别搜索商品，可以限制返回结果数量。"
schema = {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词", "examples": ["laptop", "smartphone"]},
"category": {"type": "string", "enum": ["electronics", "books", "clothing"], "description": "商品类别", "examples": ["electronics"]},
"max_results": {"type": "integer", "description": "最大返回结果数量，默认为10", "minimum": 1, "maximum": 100}
},
"required": ["query", "category"]
}
_valid_categories = ["electronics", "books", "clothing"]
def __init__(self):
self.transient_fail_count = 0
def execute(self, query: str, category: str, max_results: int = 10) -> Dict[str, Any]:
"""
执行商品搜索。
"""
print(f"--- 尝试调用 ProductSearchTool.search_products(query='{query}', category='{category}', max_results={max_results}) ---")
# 1. 模拟瞬时性错误 (每3次失败一次)
self.transient_fail_count += 1
if self.transient_fail_count % 3 != 0:
import random
if random.random() < 0.6: # 60%概率模拟瞬时错误
print("模拟瞬时性错误: 网络连接中断或服务暂时不可用。")
raise ConnectionError("Service temporarily unavailable or network issue.")
# 2. 模拟参数验证错误
if not isinstance(query, str) or not query:
raise ValueError("Parameter 'query' must be a non-empty string.")
if category not in self._valid_categories:
raise ValueError(f"Invalid value '{category}' for parameter 'category'. Expected one of {self._valid_categories}.")
if not isinstance(max_results, int):
raise TypeError(f"Parameter 'max_results' must be an integer, but received '{type(max_results).__name__}'.")
if not (1 <= max_results <= 100):
raise ValueError(f"Parameter 'max_results' must be between 1 and 100, received {max_results}.")
# 3. 模拟语义错误 (例如，搜索不到结果)
if query == "non_existent_product" and category == "electronics":
print(f"模拟语义错误: 未找到产品 '{query}'。")
return {"results": [], "message": f"No products found for '{query}' in '{category}'."}
# 成功响应
results = [
{"name": f"{query} - Item A", "price": 100.0},
{"name": f"{query} - Item B", "price": 150.0}
]
return {"results": results[:max_results], "message": f"Successfully found {len(results[:max_results])} products for '{query}' in '{category}'."}
6.2 错误解析器
一个简单的错误解析器，从字符串错误消息中提取关键信息。
# error_parser.py
import re
from typing import Dict, Any, Optional
class ErrorParser:
"""
负责解析工具返回的错误信息，提取结构化数据。
"""
@staticmethod
def parse_tool_error(error: Exception) -> Dict[str, Any]:
"""
解析工具执行时抛出的异常。
"""
error_type = type(error).__name__
error_message = str(error)
parsed_info = {
"type": error_type,
"message": error_message,
"is_transient": False,
"problematic_params": {} # 尝试提取有问题的参数
}
# 瞬时性错误判断
if error_type in ["ConnectionError", "TimeoutError"]:
parsed_info["is_transient"] = True
return parsed_info
# 参数验证错误提取（基于我们ProductSearchTool的错误格式）
# 示例: "Invalid value 'food' for parameter 'category'. Expected one of ['electronics', 'books', 'clothing']."
param_value_error_match = re.search(r"Invalid value '([^']+)' for parameter '([^']+)'. Expected one of [(.+?)].", error_message)
if param_value_error_match:
problem_value = param_value_error_match.group(1)
param_name = param_value_error_match.group(2)
expected_values_str = param_value_error_match.group(3)
parsed_info["problematic_params"][param_name] = {
"reported_value": problem_value,
"error_detail": f"Invalid value. Expected one of [{expected_values_str}].",
"param_name": param_name
}
# 示例: "Parameter 'max_results' must be an integer, but received 'str'."
param_type_error_match = re.search(r"Parameter '([^']+)' must be an ([^,]+), but received '([^']+)'.", error_message)
if param_type_error_match:
param_name = param_type_error_match.group(1)
expected_type = param_type_error_match.group(2)
received_type = param_type_error_match.group(3)
parsed_info["problematic_params"][param_name] = {
"reported_value": f"Received type {received_type}",
"error_detail": f"Type mismatch. Expected {expected_type}.",
"param_name": param_name
}
# 示例: "Parameter 'max_results' must be between 1 and 100, received 150."
param_range_error_match = re.search(r"Parameter '([^']+)' must be between (d+) and (d+), received (d+).", error_message)
if param_range_error_match:
param_name = param_range_error_match.group(1)
min_val = param_range_error_match.group(2)
max_val = param_range_error_match.group(3)
received_val = param_range_error_match.group(4)
parsed_info["problematic_params"][param_name] = {
"reported_value": received_val,
"error_detail": f"Value out of range. Expected between {min_val} and {max_val}.",
"param_name": param_name
}
# 示例: "Parameter 'query' must be a non-empty string."
param_empty_error_match = re.search(r"Parameter '([^']+)' must be a non-empty string.", error_message)
if param_empty_error_match:
param_name = param_empty_error_match.group(1)
parsed_info["problematic_params"][param_name] = {
"reported_value": "",
"error_detail": "Parameter cannot be empty.",
"param_name": param_name
}
return parsed_info
代码解析：
ErrorParser包含一个静态方法parse_tool_error。- 它首先识别异常类型 (
ConnectionError,TimeoutError被标记为瞬时性)。 - 然后使用正则表达式尝试从错误消息中提取更具体的参数验证错误信息，例如哪个参数有问题，期望什么值，实际是什么值。这部分是模拟智能体理解错误的关键。
6.3 智能体实现
将所有组件整合到 SelfCorrectingAgent 类中。
# self_correcting_agent.py
import time
import json
from typing import Dict, Any, List, Callable, Tuple, Type
from tool_definitions import ProductSearchTool
from error_parser import ErrorParser
from mock_llm_client import MockLLMClient # 引入我们之前定义的MockLLMClient
# 定义一个用于装饰器的重试函数（可以从之前的 retry_with_exponential_backoff 复制过来）
def retry_with_exponential_backoff_func(func, max_attempts: int = 3, initial_delay: float = 1.0, max_delay: float = 60.0,
factor: float = 2.0, jitter: bool = True):
delay = initial_delay
for attempt in range(max_attempts):
try:
return func() # func是一个lambda或无参数函数
except Exception as e:
print(f"尝试 {attempt + 1}/{max_attempts} 失败: {e}")
if attempt + 1 == max_attempts:
raise # 最后一次尝试失败，向上抛出异常
sleep_time = delay
if jitter:
import random
sleep_time = delay * (1 + random.uniform(-0.25, 0.25))
print(f"等待 {sleep_time:.2f} 秒后重试...")
time.sleep(min(sleep_time, max_delay))
delay *= factor
# 理论上不会到达这里
raise Exception("Max retry attempts reached without success.")
class SelfCorrectingAgent:
def __init__(self, llm_client: MockLLMClient):
self.llm_client = llm_client
self.registered_tools: Dict[str, Any] = {
ProductSearchTool.name: ProductSearchTool()
}
self.tool_schemas: Dict[str, Dict[str, Any]] = {
ProductSearchTool.name: ProductSearchTool.schema
}
self.max_retries_transient = 3 # 瞬时性错误最大重试次数
self.max_correction_attempts = 2 # 参数修正最大尝试次数
self.task_goal = "" # 当前任务目标
def _execute_tool(self, tool_name: str, args: Dict[str, Any]) -> Any:
"""
内部方法，用于执行具体的工具。
"""
tool_instance = self.registered_tools.get(tool_name)
if not tool_instance:
raise ValueError(f"Tool '{tool_name}' not registered.")
# 将字典参数解包传递给工具的execute方法
return tool_instance.execute(**args)
def run_task(self, task_goal: str, initial_tool_call: Dict[str, Any]) -> Any:
"""
智能体执行一个包含自修正机制的任务。
"""
self.task_goal = task_goal
tool_name = initial_tool_call["tool_name"]
current_args = initial_tool_call["args"]
transient_retry_count = 0
correction_attempt_count = 0
print(f"n--- 智能体开始执行任务: {task_goal} ---")
print(f"初始工具调用: {tool_name} with {current_args}")
while True:
try:
# 尝试执行工具
result = retry_with_exponential_backoff_func(
lambda: self._execute_tool(tool_name, current_args),
max_attempts=self.max_retries_transient,
initial_delay=0.5
)
print(f"n--- 工具调用成功！ ---")
print(f"结果: {result}")
return result
except Exception as e:
parsed_error = ErrorParser.parse_tool_error(e)
print(f"n--- 工具调用失败！错误类型: {parsed_error['type']}, 消息: {parsed_error['message']} ---")
if parsed_error["is_transient"]:
# 瞬时性错误，由 retry_with_exponential_backoff_func 内部处理重试
# 如果能到这里，说明 retry_with_exponential_backoff_func 已经用尽了重试次数
print(f"瞬时性错误，已达最大重试次数 ({self.max_retries_transient})。任务失败。")
raise e # 向上抛出最终失败
else: # 参数或语义错误
if correction_attempt_count >= self.max_correction_attempts:
print(f"已达到最大参数修正尝试次数 ({self.max_correction_attempts})。任务失败。")
raise e # 向上抛出最终失败
print(f"尝试进行参数修正 (第 {correction_attempt_count + 1}/{self.max_correction_attempts} 次修正尝试)...")
correction_attempt_count += 1
# 调用LLM获取修正建议
llm_suggestion = self.llm_client.get_correction_suggestion(
task_goal=self.task_goal,
tool_name=tool_name,
original_tool_args=current_args,
error_message=parsed_error["message"],
tool_schema=self.tool_schemas.get(tool_name)
)
corrected_params = llm_suggestion.get("corrected_params", {})
explanation = llm_suggestion.get("explanation", "无详细解释。")
if corrected_params:
# 应用LLM的修正建议
new_args = current_args.copy()
new_args.update(corrected_params)
print(f"LLM修正建议: {explanation}")
print(f"应用修正: 从 {current_args} 修正为 {new_args}")
current_args = new_args
# 重置瞬时性重试计数，因为参数已更改，可以重新尝试
transient_retry_count = 0
# 继续循环，使用新的参数再次尝试
else:
print(f"LLM未能提供有效修正建议。原因: {explanation} 任务失败。")
raise ValueError(f"LLM failed to provide valid correction for {tool_name} with error: {parsed_error['message']}")
# 运行示例
if __name__ == "__main__":
llm_client = MockLLMClient()
agent = SelfCorrectingAgent(llm_client)
print("n======== 场景 1: 参数类型错误和值错误 ========")
# 模拟智能体第一次调用时，参数类型和值都错误
task_1_goal = "搜索关于电子产品的5个结果"
initial_call_1 = {
"tool_name": "search_products",
"args": {
"query": "smartwatch",
"category": "gadgets", # 错误类别
"max_results": "five" # 错误类型
}
}
try:
agent.run_task(task_1_goal, initial_call_1)
except Exception as e:
print(f"n任务 1 最终失败: {e}")
print("nn======== 场景 2: 仅参数值错误，同时有瞬时性错误 ========")
# 模拟智能体第一次调用时，参数值错误，并且可能遇到瞬时性错误
agent = SelfCorrectingAgent(llm_client) # 重置agent状态
task_2_goal = "寻找关于书籍的前10个结果"
initial_call_2 = {
"tool_name": "search_products",
"args": {
"query": "fantasy novel",
"category": "story", # 错误类别
"max_results": 150 # 超出范围
}
}
try:
agent.run_task(task_2_goal, initial_call_2)
except Exception as e:
print(f"n任务 2 最终失败: {e}")
print("nn======== 场景 3: 语义错误 (MockLLM目前无法修正) ========")
agent = SelfCorrectingAgent(llm_client) # 重置agent状态
task_3_goal = "搜索一个不存在的电子产品"
initial_call_3 = {
"tool_name": "search_products",
"args": {
"query": "non_existent_product",
"category": "electronics",
"max_results": 5
}
}
try:
agent.run_task(task_3_goal, initial_call_3)
except Exception as e:
print(f"n任务 3 最终失败: {e}")
print("nn======== 场景 4: 瞬时性错误，最终成功 ========")
agent = SelfCorrectingAgent(llm_client) # 重置agent状态
task_4_goal = "搜索关于书籍的前5个结果"
initial_call_4 = {
"tool_name": "search_products",
"args": {
"query": "history book",
"category": "books",
"max_results": 5
}
}
try:
agent.run_task(task_4_goal, initial_call_4)
except Exception as e:
print(f"n任务 4 最终失败: {e}")代码解析：
SelfCorrectingAgent类封装了智能体的核心逻辑。registered_tools存储可用的工具实例及其schema。run_task方法是任务执行的主循环。- 它包含一个
while True循环，用于不断尝试执行工具，直到成功或达到最大重试/修正次数。 _execute_tool方法负责实际调用工具，并处理参数解包。- 当
_execute_tool抛出异常时，ErrorParser.parse_tool_error被调用来解析错误。 - 根据
parsed_error["is_transient"]决定是进行瞬时性重试（由retry_with_exponential_backoff_func处理）还是参数修正。 - 对于参数错误，智能体检查
correction_attempt_count，然后调用llm_client.get_correction_suggestion获取修正建议。 - 如果LLM提供了修正，智能体更新
current_args并继续循环，重新尝试调用工具。 - 如果达到最大修正次数或LLM无法提供有效修正，则任务最终失败。
7. 进阶考量与最佳实践
实现自修正机制并非一劳永逸，还需要考虑一些进阶问题和最佳实践：
-
Schema 驱动的修正：
- 将工具的完整 OpenAPI/JSON Schema 传递给LLM是提高修正准确性的关键。LLM可以利用Schema中的类型、枚举、范围、正则等信息进行更精准的推理。
- 在Prompt中以结构化的方式（例如，JSON或YAML）呈现Schema，而不是长篇大论的自然语言描述。
-
避免无限循环：
- 必须设置严格的
max_retries_transient和max_correction_attempts。 - 如果LLM反复提供相同的错误修正或无效修正，智能体应该能够识别并终止循环。可以跟踪
current_args是否在连续修正中发生变化，或者变化是否有效。
- 必须设置严格的
-
Human-in-the-Loop (HIL)：
- 当智能体在多次尝试和修正后仍无法解决问题时，应将其升级为人工干预。将失败的上下文、错误信息和智能体的修正历史呈现给用户或开发者，以便他们介入解决。
- 这可以在
max_correction_attempts达到后触发。
-
学习与优化：
- 记录智能体成功修正的案例。这些数据可以用于：
- 微调（Fine-tuning）LLM：使用实际的错误-修正对来进一步训练或微调LLM，使其在特定领域的修正能力更强。
- 优化Prompt：分析失败的修正案例，改进LLM的Prompt，使其更清晰、更全面。
- 构建规则库：对于某些频繁出现的、模式化的参数错误，可以将其转换为硬编码的规则，绕过LLM调用，节省成本并提高效率。
- 记录智能体成功修正的案例。这些数据可以用于：
-
成本管理：
- LLM调用是消耗计算资源和费用的。应优化何时调用LLM：
- 优先使用基于规则的错误解析和修正（如果可行）。
- 仅在无法通过简单重试或硬编码规则解决的参数/语义错误时才调用LLM。
- 考虑使用更小、更快的模型进行初步修正，只有在初步修正失败时才升级到更强大的模型。
- LLM调用是消耗计算资源和费用的。应优化何时调用LLM：
-
安全性与验证：
- LLM生成的参数修正，在应用之前必须进行严格的验证。防止LLM生成恶意或意外的参数，导致工具执行不安全操作。
- 这包括类型检查、范围检查、枚举值检查，甚至安全沙箱执行。
-
上下文的丰富性：
- 除了错误消息和Schema，智能体还可以向LLM提供更丰富的上下文，例如：
- 用户原始的自然语言请求。
- 智能体当前的思维链（Chain of Thought）。
- 之前成功执行的工具步骤。
- 这些信息可以帮助LLM更好地理解任务意图，从而提出更符合逻辑的修正。
- 除了错误消息和Schema，智能体还可以向LLM提供更丰富的上下文，例如：
自修正机制的未来展望与持续演进
自修正机制是智能体迈向真正自主和鲁棒的关键一步。随着LLM能力的不断提升，我们可以预见更加智能和复杂的自修正能力：
- 更深层次的语义理解：LLM将能够更准确地理解工具错误中的业务逻辑含义，而不仅仅是语法或类型错误。
- 多工具协作修正：当一个工具失败时，LLM可能不仅仅修正参数，还能建议调用其他辅助工具来获取缺失的信息，或者切换到完全不同的工具链来完成任务。
- 主动预防错误：智能体在生成工具调用参数之前，就能利用LLM的能力，根据工具Schema和历史成功/失败案例，预测潜在的参数错误并提前修正。
- 持续学习与适应：智能体将能够通过每次成功的自修正，不断优化其内部的知识表示和推理能力，使其在面对新工具或新错误类型时也能快速适应。
这不仅仅是技术上的优化，更是智能体从被动反应到主动适应、从简单工具执行者到复杂问题解决者的演进。一个能够从错误中学习并自我修复的智能体，将无疑在各种实际应用场景中展现出更强大的价值。