File size: 2,638 Bytes
225a75e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
#!/usr/bin/env python3
"""
Tool System for GAIA Agent Framework
Provides base classes and interfaces for all agent tools
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from dataclasses import dataclass
import time
import logging
# Use existing ToolResult from agents.state
from agents.state import ToolResult
logger = logging.getLogger(__name__)
class BaseTool(ABC):
"""
Base class for all agent tools
Provides consistent interface and error handling
"""
def __init__(self, name: str):
self.name = name
self.usage_count = 0
self.total_execution_time = 0.0
@abstractmethod
def _execute_impl(self, input_data: Any, **kwargs) -> Any:
"""
Implementation-specific execution logic
Override this method in subclasses
"""
pass
def execute(self, input_data: Any, **kwargs) -> ToolResult:
"""
Execute the tool with error handling and metrics tracking
"""
start_time = time.time()
try:
logger.info(f"Executing tool: {self.name}")
result = self._execute_impl(input_data, **kwargs)
execution_time = time.time() - start_time
self.usage_count += 1
self.total_execution_time += execution_time
logger.info(f"✅ Tool {self.name} completed in {execution_time:.2f}s")
return ToolResult(
tool_name=self.name,
success=True,
result=result,
execution_time=execution_time,
metadata={
"input_type": type(input_data).__name__,
"usage_count": self.usage_count
}
)
except Exception as e:
execution_time = time.time() - start_time
error_msg = f"Tool {self.name} failed: {str(e)}"
logger.error(f"❌ {error_msg}")
return ToolResult(
tool_name=self.name,
success=False,
result=None,
error=error_msg,
execution_time=execution_time
)
def get_stats(self) -> Dict[str, Any]:
"""Get usage statistics for this tool"""
return {
"name": self.name,
"usage_count": self.usage_count,
"total_execution_time": self.total_execution_time,
"average_execution_time": self.total_execution_time / max(self.usage_count, 1)
}
__all__ = ['BaseTool', 'ToolResult'] |