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