概述

工具调用循环是 Agent 的核心执行机制，需要处理各种错误情况：参数解析失败、执行超时、权限不足等。本文介绍健壮的错误处理架构。

完整示例

from typing import Any, Callable
from dataclasses import dataclass
from enum import Enum
import asyncio

class ToolError(Exception):
    """工具执行错误"""
    pass

class ParameterError(ToolError):
    """参数解析错误"""
    pass

class PermissionError(ToolError):
    """权限错误"""
    pass

@dataclass
class ToolResult:
    success: bool
    result: Any = None
    error: str = ""
    retryable: bool = False

class ToolExecutor:
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
    
    async def execute_with_retry(
        self,
        tool_func: Callable,
        parameters: dict
    ) -> ToolResult:
        """带重试的工具执行"""
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                # 参数验证
                validated_params = self._validate_params(parameters)
                
                # 执行工具
                if asyncio.iscoroutinefunction(tool_func):
                    result = await tool_func(**validated_params)
                else:
                    result = tool_func(**validated_params)
                
                # 结果验证
                if self._validate_result(result):
                    return ToolResult(success=True, result=result)
                else:
                    return ToolResult(
                        success=False,
                        error="结果验证失败",
                        retryable=False
                    )
                    
            except ParameterError as e:
                return ToolResult(success=False, error=str(e), retryable=False)
                
            except PermissionError as e:
                return ToolResult(success=False, error=str(e), retryable=False)
                
            except TimeoutError as e:
                last_error = e
                if attempt < self.max_retries - 1:
                    await asyncio.sleep(2 ** attempt)  # 指数退避
                
            except Exception as e:
                last_error = e
                
        return ToolResult(
            success=False,
            error=str(last_error),
            retryable=True
        )
    
    def _validate_params(self, params: dict) -> dict:
        """参数验证"""
        if not isinstance(params, dict):
            raise ParameterError(f"参数必须是字典，实际: {type(params)}")
        return params
    
    def _validate_result(self, result: Any) -> bool:
        """结果验证"""
        return result is not None

# 使用示例
async def main():
    executor = ToolExecutor(max_retries=3)
    
    async def fake_tool(query: str) -> str:
        return f"结果: {query}"
    
    result = await executor.execute_with_retry(
        fake_tool,
        {"query": "测试查询"}
    )
    
    if result.success:
        print(f"执行成功: {result.result}")
    else:
        print(f"执行失败: {result.error}")

asyncio.run(main())

错误处理策略

参考资料

• LangChain Tool Calling