自我修正循环：驱动Agent自动修复工具参数的图路径设计
在构建复杂的AI Agent系统时，我们常常面临一个核心挑战：Agent如何可靠地与外部世界交互？这些交互通常通过“工具”来实现，例如调用API、执行数据库查询或与外部服务通信。然而，外部世界并非完美无瑕，工具调用也常常会因为各种原因返回错误码。一个健壮的Agent不仅需要识别这些错误，更要具备自我修正的能力，尤其是在参数错误导致工具调用失败时，能够自动诊断并修复这些参数，从而提高其自主性和可靠性。
今天，我们将深入探讨“自我修正循环”（Self-Correction Loop）这一关键概念，并详细阐述如何通过设计精巧的图路径，引导Agent在工具调用返回错误码时，实现参数的自动化修复。
一、理解自我修正循环：韧性与自主的基石
什么是自我修正循环？
自我修正循环是一种使AI系统能够检测、诊断并解决自身操作中出现的问题的机制。它不仅仅是简单的重试，而是一个包含观察、分析、规划、执行和评估的完整闭环过程，旨在提高系统的性能、鲁棒性和自主性。在Agent的语境中，这意味着当Agent的某个行为（例如工具调用）未能达到预期结果时，它能够识别失败，理解失败的原因，然后调整其策略或行动，再次尝试，直至成功或确定无法解决。
为何自我修正对Agent至关重要？
- 提高鲁棒性与可靠性： 外部系统的不稳定性、API的瞬时故障、数据格式的微小偏差，都可能导致工具调用失败。自我修正机制使得Agent能够优雅地从这些错误中恢复，而不是简单地崩溃或停止。
- 增强自主性： Agent无需人工干预即可处理常见错误，减少了操作员的负担，并使其能够在更复杂的、动态的环境中独立运行。
- 优化资源利用： 通过智能地修正参数而非盲目重试，Agent可以避免重复无效的请求，从而节省计算资源和API调用成本。
- 学习与适应： 随着时间的推移，Agent可以通过积累错误处理经验，学习更有效的修正策略，甚至适应新的错误模式。
自我修正循环的核心组件：
一个典型的自我修正循环包含以下阶段：
- 观察/监控 (Observation/Monitoring)： Agent执行任务并监控结果。当工具调用返回错误码时，即触发此阶段。
- 诊断/分析 (Diagnosis/Analysis)： Agent解析错误信息和错误码，尝试理解失败的根本原因。这是从“发生了什么”到“为什么发生”的关键一步。
- 规划/修正 (Planning/Correction)： 基于诊断结果，Agent规划一个修正策略。这可能涉及修改参数、选择不同的工具、或调整执行流程。
- 执行/行动 (Execution/Action)： Agent根据修正策略执行新的操作，例如使用修复后的参数重新调用工具。
- 评估/反馈 (Evaluation/Feedback)： Agent再次评估新操作的结果。如果成功，则任务继续；如果再次失败，则可能进入下一轮修正循环，或向上级系统报告。
在本文的场景中，我们的重点是当工具调用返回参数相关的错误码时，Agent如何通过智能诊断和修正参数来完成这个循环。
二、工具调用错误：参数修复的特定挑战
Agent通过工具与外部世界交互，其核心在于将Agent的意图和上下文转化为工具可理解的输入参数。然而，这个转化过程以及外部工具的验证机制，是错误的常见来源。
Agent中常见的工具错误场景：
- 参数类型不匹配 (Type Mismatch)： 工具期望一个整数，但Agent提供了字符串；期望日期对象，但提供了错误格式的字符串。
- 示例： 调用
create_event(date="2023/10/26", time="10:00AM")，但工具期望date是YYYY-MM-DD格式。
- 示例： 调用
- 参数值超出范围 (Out-of-Range Value)： 参数值不符合工具定义的业务逻辑范围。
- 示例： 调用
set_temperature(temp=-100)，但工具只接受0到100的温度。
- 示例： 调用
- 缺少必要参数 (Missing Required Parameter)： Agent未能提供工具必需的某个参数。
- 示例： 调用
book_flight(origin="NYC", destination="LAX")，但忘记提供departure_date。
- 示例： 调用
- 参数格式错误 (Format Error)： 参数值虽然类型正确，但格式不符合工具的特定要求（如正则表达式）。
- 示例： 调用
set_phone_number(number="123-456-7890")，但工具期望+1 (XXX) XXX-XXXX格式。
- 示例： 调用
- 业务逻辑错误 (Business Logic Error)： 参数值在语法和类型上都正确，但与业务规则冲突（例如，预订的会议室已被占用，或用户没有权限）。
- 示例： 调用
update_user_profile(user_id="invalid_id")，工具返回用户不存在。
- 示例： 调用
- Schema验证失败 (Schema Validation Failure)： 工具通常有明确的API输入Schema（如OpenAPI/Swagger），如果Agent生成的参数不符合此Schema，则会失败。
我们的目标是设计一个图路径，让Agent能够智能地识别这些参数相关的错误，并引导它自动地尝试修复这些参数。
三、设计图路径引导Agent自动修复参数
要实现Agent的自动参数修复，我们需要一个结构化的决策流程。图（Graph）是描述这种流程的理想模型，其中节点代表Agent的决策点或操作状态，边代表基于特定条件（如错误类型）的状态转换。
A. 核心数据结构与概念
在构建图路径之前，我们需要定义一些关键的数据结构来支持Agent的决策和参数修复过程。
-
Tool Definition (工具定义)：
Agent需要了解其可用的工具及其期望的参数。这通常通过JSON Schema或其他描述语言来完成。from typing import Dict, Any, Optional from pydantic import BaseModel, Field import json class ParameterSchema(BaseModel): type: str description: Optional[str] = None enum: Optional[list] = None format: Optional[str] = None # e.g., "date", "date-time", "email", "uuid" pattern: Optional[str] = None # Regex pattern minimum: Optional[float] = None maximum: Optional[float] = None # Add more JSON Schema keywords as needed class ToolParameters(BaseModel): type: str = "object" properties: Dict[str, ParameterSchema] = Field(default_factory=dict) required: list[str] = Field(default_factory=list) class ToolSpec(BaseModel): name: str description: str parameters: ToolParameters # JSON Schema for input parameters def validate_params(self, params: Dict[str, Any]) -> bool: # A more robust validation would use a proper JSON Schema validator like jsonschema # For simplicity, let's implement basic checks here. for required_param in self.parameters.required: if required_param not in params: return False, f"Missing required parameter: {required_param}" for param_name, schema in self.parameters.properties.items(): if param_name in params: value = params[param_name] # Type check if schema.type == "string" and not isinstance(value, str): return False, f"Parameter '{param_name}' must be a string." if schema.type == "integer" and not isinstance(value, int): return False, f"Parameter '{param_name}' must be an integer." if schema.type == "number" and not isinstance(value, (int, float)): return False, f"Parameter '{param_name}' must be a number." if schema.type == "boolean" and not isinstance(value, bool): return False, f"Parameter '{param_name}' must be a boolean." # Format/Pattern check (simplified) if schema.format == "date" and schema.type == "string": import re if not re.match(r"^d{4}-d{2}-d{2}$", value): # YYYY-MM-DD return False, f"Parameter '{param_name}' has invalid date format. Expected YYYY-MM-DD." if schema.pattern and schema.type == "string": if not re.match(schema.pattern, value): return False, f"Parameter '{param_name}' does not match pattern: {schema.pattern}" # Enum check if schema.enum and value not in schema.enum: return False, f"Parameter '{param_name}' must be one of {schema.enum}." # Range check if schema.minimum is not None and isinstance(value, (int, float)) and value < schema.minimum: return False, f"Parameter '{param_name}' must be >= {schema.minimum}." if schema.maximum is not None and isinstance(value, (int, float)) and value > schema.maximum: return False, f"Parameter '{param_name}' must be <= {schema.maximum}." return True, "Parameters are valid." # Example Tool Definition book_flight_tool = ToolSpec( name="book_flight", description="Book a flight for a user.", parameters=ToolParameters( properties={ "origin": ParameterSchema(type="string", description="Departure city IATA code"), "destination": ParameterSchema(type="string", description="Arrival city IATA code"), "departure_date": ParameterSchema(type="string", format="date", description="Departure date in YYYY-MM-DD format"), "return_date": ParameterSchema(type="string", format="date", description="Return date in YYYY-MM-DD format", required=False), "passengers": ParameterSchema(type="integer", minimum=1, maximum=9, description="Number of passengers", required=False), "class": ParameterSchema(type="string", enum=["economy", "business", "first"], description="Travel class", required=False) }, required=["origin", "destination", "departure_date"] ) ) -
Error Code Mapping与修复策略 (Error Strategy Map)：
Agent需要一个机制来将外部工具返回的错误码或错误消息映射到具体的修复策略。Error Code/Pattern Error Description Suggested Repair Strategy Relevant Parameters Priority 400 Bad RequestInvalid date format for 'departure_date'DATE_FORMAT_CORRECTIONdeparture_dateHigh 400 Bad RequestMissing required parameter 'passengers'LLM_INFER_MISSING_PARAM/DEFAULT_VALUE_ASSIGNMENTpassengersHigh 400 Bad RequestValue for 'passengers' must be >= 1CLAMP_VALUE_TO_MINpassengersMedium 400 Bad RequestValue for 'class' must be one of [...]ENUM_VALUE_CORRECTION/LLM_SUGGEST_ENUMclassHigh 404 Not FoundFlight for given route not foundALTERNATIVE_ROUTE_SUGGESTION/DATE_ADJUSTMENTorigin,dest,dateLow 401 UnauthorizedInvalid API KeyREPORT_UNRECOVERABLEN/A Critical 5xx Server ErrorInternal Server ErrorRETRY_WITH_BACKOFF(often not parameter related, but system)N/A Medium VALIDATION_ERRORParameter X has invalid type YSCHEMA_TYPE_CONVERSIONXHigh VALIDATION_ERRORParameter X does not match pattern PLLM_BASED_FORMAT_CORRECTIONXHigh from enum import Enum class RepairStrategyType(Enum): NONE = "No specific repair strategy" LLM_INFER_MISSING_PARAM = "Infer missing parameter using LLM based on context" DATE_FORMAT_CORRECTION = "Attempt to parse and reformat date string" ENUM_VALUE_CORRECTION = "Suggest correct enum value or infer from context" SCHEMA_TYPE_CONVERSION = "Attempt to convert parameter type based on schema" CLAMP_VALUE_TO_MIN = "Adjust numeric value to minimum allowed by schema" LLM_BASED_FORMAT_CORRECTION = "Use LLM to reformat parameter based on pattern/format" REPORT_UNRECOVERABLE = "Report as unrecoverable error" RETRY_WITH_BACKOFF = "Retry after a delay (not parameter specific)" # ... more strategies class ErrorStrategy: def __init__(self, error_pattern: str, strategy_type: RepairStrategyType, target_params: Optional[list[str]] = None, description: str = ""): self.error_pattern = error_pattern # Regex pattern to match error message self.strategy_type = strategy_type self.target_params = target_params if target_params is not None else [] self.description = description # A simplified error strategy map ERROR_STRATEGY_MAP: list[ErrorStrategy] = [ ErrorStrategy(r"Invalid date format for '(w+)'", RepairStrategyType.DATE_FORMAT_CORRECTION, description="Date format error"), ErrorStrategy(r"Missing required parameter '(w+)'", RepairStrategyType.LLM_INFER_MISSING_PARAM, description="Missing required parameter"), ErrorStrategy(r"Value for '(w+)' must be >= (d+)", RepairStrategyType.CLAMP_VALUE_TO_MIN, description="Value below minimum"), ErrorStrategy(r"Value for '(w+)' must be one of [([^]]+)]", RepairStrategyType.ENUM_VALUE_CORRECTION, description="Invalid enum value"), ErrorStrategy(r"Parameter '(w+)' has invalid type", RepairStrategyType.SCHEMA_TYPE_CONVERSION, description="Parameter type mismatch"), ErrorStrategy(r"Parameter '(w+)' does not match pattern", RepairStrategyType.LLM_BASED_FORMAT_CORRECTION, description="Parameter format pattern mismatch"), ErrorStrategy(r"Unauthorized|Invalid API Key", RepairStrategyType.REPORT_UNRECOVERABLE, description="Authentication error"), # Default catch-all for unknown 400s that might still be parameter related ErrorStrategy(r"400 Bad Request", RepairStrategyType.LLM_INFER_MISSING_PARAM, description="Generic bad request, try LLM inference"), # Lower priority, broader match ] -
Agent Context / State (Agent的上下文/状态)：
Agent在执行过程中需要维护其当前状态，包括原始请求、当前尝试的参数、错误历史等。from datetime import datetime class AgentState(BaseModel): task_id: str original_prompt: str current_tool_name: Optional[str] = None current_params: Dict[str, Any] = Field(default_factory=dict) attempt_count: int = 0 max_attempts: int = 3 error_history: list[str] = Field(default_factory=list) last_error_message: Optional[str] = None last_error_code: Optional[int] = None current_repair_strategy: Optional[RepairStrategyType] = None # ... other state variables like LLM chat history for context
B. 图路径设计：节点与转换
我们将Agent的参数修复流程建模为一个有向图。
图节点 (Nodes)：
Start(开始)： Agent接收到任务，开始处理。PlanToolCall(规划工具调用)： Agent根据任务选择合适的工具并生成初始参数。ExecuteTool(执行工具)： Agent调用选定的工具，传入参数。ToolCallSuccess(工具调用成功)： 工具返回成功结果。ErrorDetected(检测到错误)： 工具调用失败，返回错误码和错误信息。DiagnoseError(诊断错误)： Agent分析错误信息，识别错误类型和可能受影响的参数。SelectRepairStrategy(选择修复策略)： 根据诊断结果和ERROR_STRATEGY_MAP选择合适的修复策略。ApplyRepairStrategy(应用修复策略)： Agent根据选定的策略修改参数。ValidateParameters(验证参数)： 重新验证修复后的参数是否符合工具Schema。RetryToolCall(重试工具调用)： 使用修复后的参数再次尝试调用工具。NoFurtherRepairPossible(无法进一步修复)： 达到最大尝试次数或没有合适的修复策略。ReportFailure(报告失败)： 任务无法完成，向用户或上级系统报告。End(结束)： 任务完成或彻底失败。
图边 (Edges) 与转换逻辑：
Start->PlanToolCall: 初始任务启动。PlanToolCall->ExecuteTool: 参数生成完毕，准备执行。ExecuteTool->ToolCallSuccess: (条件：工具调用返回成功状态码2xx)ExecuteTool->ErrorDetected: (条件：工具调用返回错误状态码4xx/5xx)ErrorDetected->DiagnoseError: 捕获到错误后，立即进行诊断。DiagnoseError->SelectRepairStrategy: 诊断完成后，选择策略。SelectRepairStrategy->ApplyRepairStrategy: (条件：找到匹配的修复策略)SelectRepairStrategy->NoFurtherRepairPossible: (条件：未找到匹配策略，或已达最大尝试次数)ApplyRepairStrategy->ValidateParameters: 修复参数后，进行客户端侧的Schema验证。ValidateParameters->RetryToolCall: (条件：参数通过客户端验证)ValidateParameters->NoFurtherRepairPossible: (条件：修复后的参数依然不符合Schema，或修复过程本身出错)RetryToolCall->ExecuteTool: 重新执行工具调用。ToolCallSuccess->End: 任务成功完成。NoFurtherRepairPossible->ReportFailure: 无法解决的错误。ReportFailure->End: 任务以失败告终。
图路径的可视化表示 (概念性)：
Start
|
V
PlanToolCall
|
V
ExecuteTool ---------------------------------------------> ToolCallSuccess
| (Error) |
V V
ErrorDetected -----------------------------------------> End
| ^
V |
DiagnoseError -----------------------------------------> ReportFailure
| ^
V |
SelectRepairStrategy ---------------------------------> NoFurtherRepairPossible
| (Strategy Found)
V
ApplyRepairStrategy
|
V
ValidateParameters
| (Valid)
V
RetryToolCall -------------------------------------------> ExecuteTool (Loop back)
| (Invalid/Repair Failed)
V
NoFurtherRepairPossibleC. 实现细节：Agent与修正策略
我们将Agent定义为一个能够执行工具和管理状态的类，并实现各种修复策略作为独立的方法或类。
-
Agent核心类结构：
import time import re from typing import Callable, Tuple, Any from datetime import datetime # Mock LLM Client for demonstration class MockLLMClient: def chat_completion(self, messages: list[dict], response_format: str = "text") -> str: # Simulate LLM's parameter inference capability print(f"LLM called with messages: {messages}") # In a real scenario, this would involve a complex prompt and parsing # For demonstration, let's assume it can infer "passengers" from context. if "missing required parameter 'passengers'" in messages[-1]['content']: if "for myself" in messages[0]['content'] or "for me" in messages[0]['content']: if response_format == "json": return json.dumps({"passengers": 1}) return "1" if "Invalid date format" in messages[-1]['content']: # Simplified: assume LLM can fix a common format match = re.search(r"(d{4}[-/]d{2}[-/]d{2})", messages[0]['content']) if match: try: dt_obj = datetime.strptime(match.group(1).replace('/', '-'), "%Y-%m-%d") if response_format == "json": return json.dumps({"departure_date": dt_obj.strftime("%Y-%m-%d")}) return dt_obj.strftime("%Y-%m-%d") except ValueError: pass if "enum" in messages[-1]['content'] and "class" in messages[-1]['content']: if "business class" in messages[0]['content']: if response_format == "json": return json.dumps({"class": "business"}) return "business" return "" # LLM couldn't infer or fix # Mock Tool Executor class MockToolExecutor: def __init__(self, tool_specs: Dict[str, ToolSpec]): self.tool_specs = tool_specs def execute(self, tool_name: str, params: Dict[str, Any]) -> Tuple[bool, Any, Optional[str], Optional[int]]: print(f"Executing tool '{tool_name}' with params: {params}") if tool_name not in self.tool_specs: return False, None, f"Tool '{tool_name}' not found", 404 tool_spec = self.tool_specs[tool_name] is_valid, validation_msg = tool_spec.validate_params(params) if not is_valid: print(f"Tool execution failed due to schema validation: {validation_msg}") # Simulate a 400 Bad Request with a specific error message return False, None, validation_msg, 400 # Simulate successful execution print(f"Tool '{tool_name}' executed successfully.") return True, {"status": "success", "message": f"Operation successful for {tool_name}"}, None, 200 class Agent: def __init__(self, tools: Dict[str, ToolSpec], llm_client: MockLLMClient): self.tools = tools self.llm_client = llm_client self.tool_executor = MockToolExecutor(tools) self.error_strategy_map = ERROR_STRATEGY_MAP def _diagnose_error(self, error_message: str, error_code: int) -> Optional[ErrorStrategy]: print(f"Diagnosing error: [Code: {error_code}] {error_message}") for strategy in self.error_strategy_map: if re.search(strategy.error_pattern, error_message): print(f"Matched error pattern '{strategy.error_pattern}', suggesting strategy: {strategy.strategy_type.name}") return strategy print("No specific repair strategy found for this error.") return None def _apply_repair_strategy(self, state: AgentState, strategy: ErrorStrategy) -> Tuple[bool, str]: print(f"Applying repair strategy: {strategy.strategy_type.name} for params: {strategy.target_params}") repaired_params = state.current_params.copy() tool_spec = self.tools[state.current_tool_name] # type: ignore if strategy.strategy_type == RepairStrategyType.LLM_INFER_MISSING_PARAM: # Try to extract the missing parameter name from the error message match = re.search(r"Missing required parameter '(w+)'", state.last_error_message or "") missing_param = match.group(1) if match else (strategy.target_params[0] if strategy.target_params else None) if missing_param: print(f"Attempting to infer missing parameter '{missing_param}' using LLM.") # Provide LLM with original prompt, current tool, and error messages = [ {"role": "system", "content": "You are a helpful assistant that infers missing parameters for tool calls."}, {"role": "user", "content": f"The original request was: '{state.original_prompt}'. I'm trying to call the tool '{state.current_tool_name}' with parameters {state.current_params}. It failed because: '{state.last_error_message}'. Please infer a value for '{missing_param}'. Return only the inferred value or a JSON object like {{'{missing_param}': 'value'}} if complex."}, ] llm_response = self.llm_client.chat_completion(messages, response_format="json") try: inferred_data = json.loads(llm_response) if missing_param in inferred_data: repaired_params[missing_param] = inferred_data[missing_param] print(f"LLM inferred '{missing_param}': {inferred_data[missing_param]}") return True, "Parameter inferred by LLM." except json.JSONDecodeError: # Fallback for non-JSON response if llm_response: repaired_params[missing_param] = llm_response print(f"LLM inferred '{missing_param}': {llm_response}") return True, "Parameter inferred by LLM." return False, "Failed to infer missing parameter." elif strategy.strategy_type == RepairStrategyType.DATE_FORMAT_CORRECTION: # Try to extract the parameter name from the error message match = re.search(r"Invalid date format for '(w+)'", state.last_error_message or "") date_param = match.group(1) if match else (strategy.target_params[0] if strategy.target_params else None) if date_param and date_param in repaired_params: original_date_str = repaired_params[date_param] print(f"Attempting to reformat date parameter '{date_param}': '{original_date_str}'") # Use LLM for more flexible parsing, or a rule-based system messages = [ {"role": "system", "content": "You are a date formatter. Convert the given date string to YYYY-MM-DD format. If unable, respond with an empty string."}, {"role": "user", "content": f"Convert '{original_date_str}' to YYYY-MM-DD. Example: '2023/10/26' -> '2023-10-26'."} ] llm_response = self.llm_client.chat_completion(messages) if re.match(r"^d{4}-d{2}-d{2}$", llm_response): # Simple validation for LLM output repaired_params[date_param] = llm_response print(f"Successfully reformatted '{date_param}' to '{llm_response}'") return True, "Date format corrected." else: print(f"LLM failed to reformat date: {llm_response}") return False, "Failed to correct date format." elif strategy.strategy_type == RepairStrategyType.CLAMP_VALUE_TO_MIN: match = re.search(r"Value for '(w+)' must be >= (d+)", state.last_error_message or "") param_name = match.group(1) if match else (strategy.target_params[0] if strategy.target_params else None) min_value = int(match.group(2)) if match else None if param_name and min_value is not None and param_name in repaired_params: if isinstance(repaired_params[param_name], (int, float)) and repaired_params[param_name] < min_value: repaired_params[param_name] = min_value print(f"Clamped '{param_name}' to minimum value: {min_value}") return True, "Value clamped to minimum." return False, "Failed to clamp value." elif strategy.strategy_type == RepairStrategyType.ENUM_VALUE_CORRECTION: match = re.search(r"Value for '(w+)' must be one of [([^]]+)]", state.last_error_message or "") param_name = match.group(1) if match else (strategy.target_params[0] if strategy.target_params else None) allowed_values_str = match.group(2) if match else "" allowed_values = [v.strip().strip("'"") for v in allowed_values_str.split(',')] if param_name and param_name in repaired_params and allowed_values: original_value = repaired_params[param_name] # Try to infer correct enum value from original prompt or LLM messages = [ {"role": "system", "content": f"You are an assistant to correct enum values. Given the original request, the invalid value '{original_value}', and allowed values {allowed_values}, suggest the most appropriate allowed value. If no suitable one, respond with an empty string."}, {"role": "user", "content": f"Original request: '{state.original_prompt}'. Invalid value for '{param_name}': '{original_value}'. Allowed values: {allowed_values_str}. Suggest correct value."} ] llm_response = self.llm_client.chat_completion(messages) if llm_response in allowed_values: repaired_params[param_name] = llm_response print(f"Corrected enum for '{param_name}' to '{llm_response}'") return True, "Enum value corrected." return False, "Failed to correct enum value." elif strategy.strategy_type == RepairStrategyType.SCHEMA_TYPE_CONVERSION: match = re.search(r"Parameter '(w+)' must be a (w+)", state.last_error_message or "") param_name = match.group(1) if match else (strategy.target_params[0] if strategy.target_params else None) expected_type_str = match.group(2) if match else None if param_name and expected_type_str and param_name in repaired_params: original_value = repaired_params[param_name] print(f"Attempting type conversion for '{param_name}' from '{type(original_value).__name__}' to '{expected_type_str}'") try: if expected_type_str == "integer": repaired_params[param_name] = int(original_value) elif expected_type_str == "number": repaired_params[param_name] = float(original_value) elif expected_type_str == "string": repaired_params[param_name] = str(original_value) elif expected_type_str == "boolean": # Simple boolean conversion if isinstance(original_value, str): if original_value.lower() in ["true", "1", "yes"]: repaired_params[param_name] = True elif original_value.lower() in ["false", "0", "no"]: repaired_params[param_name] = False else: raise ValueError("Cannot convert string to boolean") else: repaired_params[param_name] = bool(original_value) print(f"Successfully converted '{param_name}' to type '{expected_type_str}'. New value: {repaired_params[param_name]}") return True, "Parameter type converted." except (ValueError, TypeError) as e: print(f"Type conversion failed for '{param_name}': {e}") return False, "Failed to convert parameter type." # Update the state with potentially repaired parameters state.current_params = repaired_params return False, "No specific repair logic implemented for this strategy yet." # Default fallback def run_task(self, original_prompt: str, tool_name: str, initial_params: Dict[str, Any]) -> Any: state = AgentState(task_id=f"task_{int(time.time())}", original_prompt=original_prompt, current_tool_name=tool_name, current_params=initial_params.copy()) current_node = "PlanToolCall" print(f"[{current_node}] Initializing task for: '{original_prompt}'") while True: print(f"n--- Current Node: {current_node} (Attempt {state.attempt_count + 1}/{state.max_attempts}) ---") if current_node == "PlanToolCall": state.current_params = initial_params.copy() # Reset params for a fresh start or retry path current_node = "ExecuteTool" elif current_node == "ExecuteTool": state.attempt_count += 1 if state.attempt_count > state.max_attempts: current_node = "NoFurtherRepairPossible" continue success, result, error_msg, error_code = self.tool_executor.execute( state.current_tool_name, state.current_params # type: ignore ) if success: state.current_node = "ToolCallSuccess" current_node = "ToolCallSuccess" else: state.last_error_message = error_msg state.last_error_code = error_code state.error_history.append(f"Attempt {state.attempt_count}: {error_msg} (Code: {error_code})") current_node = "ErrorDetected" elif current_node == "ToolCallSuccess": print(f"[{current_node}] Task completed successfully. Result: {result}") return result elif current_node == "ErrorDetected": print(f"[{current_node}] Error detected: {state.last_error_message}") current_node = "DiagnoseError" elif current_node == "DiagnoseError": strategy = self._diagnose_error(state.last_error_message or "", state.last_error_code or 0) state.current_repair_strategy = strategy.strategy_type if strategy else RepairStrategyType.NONE if strategy and strategy.strategy_type != RepairStrategyType.REPORT_UNRECOVERABLE: current_node = "ApplyRepairStrategy" else: current_node = "NoFurtherRepairPossible" # Either no strategy or unrecoverable elif current_node == "ApplyRepairStrategy": if state.current_repair_strategy and state.current_tool_name: strategy_obj = next((s for s in self.error_strategy_map if s.strategy_type == state.current_repair_strategy), None) if strategy_obj: repair_successful, repair_message = self._apply_repair_strategy(state, strategy_obj) if repair_successful: print(f"[{current_node}] Parameter repair successful: {repair_message}") current_node = "ValidateParameters" else: print(f"[{current_node}] Parameter repair failed: {repair_message}") current_node = "NoFurtherRepairPossible" else: print(f"[{current_node}] No concrete strategy object found for {state.current_repair_strategy.name}") current_node = "NoFurtherRepairPossible" else: current_node = "NoFurtherRepairPossible" elif current_node == "ValidateParameters": tool_spec = self.tools[state.current_tool_name] # type: ignore is_valid, validation_msg = tool_spec.validate_params(state.current_params) if is_valid: print(f"[{current_node}] Repaired parameters passed client-side validation.") current_node = "RetryToolCall" else: print(f"[{current_node}] Repaired parameters failed client-side validation: {validation_msg}") # If repair failed internal validation, it's a critical issue, probably can't fix further current_node = "NoFurtherRepairPossible" elif current_node == "RetryToolCall": print(f"[{current_node}] Retrying tool call with modified parameters: {state.current_params}") current_node = "ExecuteTool" # Loop back to execute elif current_node == "NoFurtherRepairPossible": print(f"[{current_node}] No further automated repair possible after {state.attempt_count} attempts.") current_node = "ReportFailure" elif current_node == "ReportFailure": print(f"[{current_node}] Task failed after multiple attempts and repairs. Original prompt: '{state.original_prompt}'. Last error: {state.last_error_message}") return {"status": "failed", "reason": state.last_error_message, "history": state.error_history} elif current_node == "End": return None # Should be handled by ToolCallSuccess or ReportFailure time.sleep(0.1) # Simulate some processing time
D. 运行示例：机票预订场景
让我们通过一个机票预订的例子来演示这个图路径的工作机制。
场景1：缺少必要参数
Agent尝试预订机票，但遗漏了乘客数量（虽然在ToolSpec中是可选的，但在某些业务逻辑中可能被强制要求或有默认值，我们这里模拟工具返回“missing required parameter ‘passengers’”）。
# Initialize Agent with tools and LLM client
llm_client = MockLLMClient()
agent = Agent(tools={"book_flight": book_flight_tool}, llm_client=llm_client)
print("n--- Scenario 1: Missing Required Parameter 'passengers' ---")
initial_prompt_1 = "Book a flight for me from New York to Los Angeles on 2024-01-15."
initial_params_1 = {
"origin": "NYC",
"destination": "LAX",
"departure_date": "2024-01-15",
# "passengers" is missing here
}
# Simulate tool returning an error for missing 'passengers' despite schema marking it optional
# This simulates a business logic validation that's stricter than schema.
# For demo, we'll modify the tool_executor to explicitly check for 'passengers'
class StricterMockToolExecutor(MockToolExecutor):
def execute(self, tool_name: str, params: Dict[str, Any]) -> Tuple[bool, Any, Optional[str], Optional[int]]:
if tool_name == "book_flight":
if "passengers" not in params:
return False, None, "Missing required parameter 'passengers'", 400
return super().execute(tool_name, params)
agent.tool_executor = StricterMockToolExecutor(agent.tools)
result_1 = agent.run_task(initial_prompt_1, "book_flight", initial_params_1)
print("nFinal Result 1:", result_1)
# Expected Output Flow:
# 1. PlanToolCall -> ExecuteTool
# 2. ExecuteTool fails: "Missing required parameter 'passengers'" (400)
# 3. ErrorDetected -> DiagnoseError -> SelectRepairStrategy (LLM_INFER_MISSING_PARAM)
# 4. ApplyRepairStrategy: LLM infers 'passengers': 1 (from "for me")
# 5. ValidateParameters (success) -> RetryToolCall -> ExecuteTool
# 6. ExecuteTool (with passengers=1) succeeds
# 7. ToolCallSuccess -> End场景2：日期格式错误
Agent提供了不符合YYYY-MM-DD格式的日期。
print("n--- Scenario 2: Invalid Date Format ---")
initial_prompt_2 = "I need to fly from London to Paris on October 26, 2024 for business class."
initial_params_2 = {
"origin": "LHR",
"destination": "CDG",
"departure_date": "Oct 26, 2024", # Incorrect format
"passengers": 1,
"class": "business class" # Also incorrect, should be "business"
}
# Reset tool_executor to original (less strict for this demo, focusing on schema validation)
agent.tool_executor = MockToolExecutor(agent.tools)
result_2 = agent.run_task(initial_prompt_2, "book_flight", initial_params_2)
print("nFinal Result 2:", result_2)
# Expected Output Flow:
# 1. PlanToolCall -> ExecuteTool
# 2. ExecuteTool fails: "Parameter 'departure_date' has invalid date format. Expected YYYY-MM-DD." (400)
# 3. ErrorDetected -> DiagnoseError -> SelectRepairStrategy (DATE_FORMAT_CORRECTION)
# 4. ApplyRepairStrategy: LLM reformats "Oct 26, 2024" to "2024-10-26"
# 5. ValidateParameters (success) -> RetryToolCall -> ExecuteTool
# 6. ExecuteTool fails again (due to "business class" enum error, as it's the next error found)
# Error: "Value for 'class' must be one of ['economy', 'business', 'first']" (400)
# 7. ErrorDetected -> DiagnoseError -> SelectRepairStrategy (ENUM_VALUE_CORRECTION)
# 8. ApplyRepairStrategy: LLM infers "business class" -> "business"
# 9. ValidateParameters (success) -> RetryToolCall -> ExecuteTool
# 10. ExecuteTool succeeds (all errors fixed)
# 11. ToolCallSuccess -> End场景3：不可修复的错误
Agent尝试调用一个不存在的工具，或者遇到无法通过参数修复的错误。
print("n--- Scenario 3: Unrecoverable Error (Tool Not Found) ---")
initial_prompt_3 = "Find me a hotel in New York."
initial_params_3 = {"city": "New York"}
# "find_hotel" tool is not registered in our agent's tools
result_3 = agent.run_task(initial_prompt_3, "find_hotel", initial_params_3)
print("nFinal Result 3:", result_3)
# Expected Output Flow:
# 1. PlanToolCall -> ExecuteTool
# 2. ExecuteTool fails: "Tool 'find_hotel' not found" (404)
# 3. ErrorDetected -> DiagnoseError -> SelectRepairStrategy (NONE, or specific REPORT_UNRECOVERABLE if mapped)
# 4. NoFurtherRepairPossible -> ReportFailure -> End四、高级考量与最佳实践
- 容错与幂等性： 确保修复操作是幂等的，即多次执行相同修复操作不会产生额外的副作用。对于外部API调用，要考虑API本身的幂等性。
- 循环检测与终止： 设定最大重试次数或最大修复尝试次数，防止Agent陷入无限循环。在图路径中，
attempt_count就是实现这一点的关键。 - 人机协作（Human-in-the-Loop）： 当Agent的自动化修复能力达到极限时，应设计机制将问题上报给人类操作员，提供详细的错误上下文和Agent的尝试历史，以便人类介入解决。
- 详细日志与监控： 记录Agent每次决策、工具调用、错误诊断和参数修复的详细信息。这对于调试、性能分析和改进修复策略至关重要。
- 动态策略生成： 对于更复杂的错误，LLM不仅可以用于参数推断，甚至可以尝试根据错误描述和工具Schema动态生成新的修复步骤或代码片段，这代表了更高级的自我修正。
- 工具特定与通用策略： 区分通用的参数修复策略（如类型转换、缺失参数推断）和工具特有的业务逻辑修复策略（如预订冲突时查找替代方案）。
- 成本管理： 每次调用LLM进行诊断或参数推断都会产生费用。应权衡自动化修复的收益与LLM调用的成本，在某些情况下，简单的规则匹配可能比复杂的LLM推理更具成本效益。
- 上下文保持： 确保Agent在修正循环中始终保持对原始任务意图和历史对话的理解，这对于LLM进行准确的参数推断至关重要。
五、构建更强大的自主智能系统
自我修正循环是构建真正自主和弹性AI Agent的关键。通过将Agent的决策过程建模为清晰的图路径，并结合智能的错误诊断和参数修复策略（特别是利用大型语言模型的能力），我们可以极大地提升Agent在复杂、不完美环境下的可靠性和效率。这不仅减少了人工干预的需求，也使得Agent能够更智能地从失败中学习和恢复，逐步迈向更强大的自主智能系统。