| |
| """ |
| 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 |
|
|
| |
| 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'] |
