Create smolagents/tools.py

smolagents/tools.py

@@ -0,0 +1,212 @@
+"""CSV Tool Support for Vendor-Neutral Smolagents Fork
+
+This module provides CSV-based tool handlers for the agent system.
+Designed to be vendor-neutral and work with various LLM backends.
+"""
+
+import csv
+import os
+from typing import List, Dict, Any, Optional
+import json
+
+
+class CSVToolHandler:
+    """Handler for CSV-based tool definitions and operations."""
+    
+    def __init__(self, csv_path: Optional[str] = None):
+        """Initialize the CSV tool handler.
+        
+        Args:
+            csv_path: Path to CSV file containing tool definitions.
+                     Defaults to './tools.csv' if not specified.
+        """
+        self.csv_path = csv_path or os.getenv('CSV_TOOLS_PATH', './tools.csv')
+        self.tools = []
+        if os.path.exists(self.csv_path):
+            self.load_tools()
+    
+    def load_tools(self) -> None:
+        """Load tool definitions from CSV file."""
+        try:
+            with open(self.csv_path, 'r', encoding='utf-8') as f:
+                reader = csv.DictReader(f)
+                self.tools = list(reader)
+                print(f"Loaded {len(self.tools)} tools from {self.csv_path}")
+        except Exception as e:
+            print(f"Error loading tools from CSV: {e}")
+            self.tools = []
+    
+    def get_tools(self) -> List[Dict[str, Any]]:
+        """Return list of tool definitions."""
+        return self.tools
+    
+    def get_tool_by_name(self, name: str) -> Optional[Dict[str, Any]]:
+        """Get a specific tool by name."""
+        for tool in self.tools:
+            if tool.get('name') == name:
+                return tool
+        return None
+    
+    def add_tool(self, tool: Dict[str, Any]) -> None:
+        """Add a new tool definition."""
+        self.tools.append(tool)
+    
+    def save_tools(self) -> None:
+        """Save tool definitions back to CSV file."""
+        if not self.tools:
+            print("No tools to save")
+            return
+        
+        try:
+            fieldnames = list(self.tools[0].keys())
+            with open(self.csv_path, 'w', encoding='utf-8', newline='') as f:
+                writer = csv.DictWriter(f, fieldnames=fieldnames)
+                writer.writeheader()
+                writer.writerows(self.tools)
+                print(f"Saved {len(self.tools)} tools to {self.csv_path}")
+        except Exception as e:
+            print(f"Error saving tools to CSV: {e}")
+
+
+class CSVDataHandler:
+    """Handler for CSV data operations within agents."""
+    
+    @staticmethod
+    def read_csv(file_path: str) -> List[Dict[str, Any]]:
+        """Read data from CSV file.
+        
+        Args:
+            file_path: Path to the CSV file to read.
+            
+        Returns:
+            List of dictionaries representing CSV rows.
+        """
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                reader = csv.DictReader(f)
+                return list(reader)
+        except Exception as e:
+            print(f"Error reading CSV file {file_path}: {e}")
+            return []
+    
+    @staticmethod
+    def write_csv(file_path: str, data: List[Dict[str, Any]], 
+                  fieldnames: Optional[List[str]] = None) -> bool:
+        """Write data to CSV file.
+        
+        Args:
+            file_path: Path to the CSV file to write.
+            data: List of dictionaries to write as CSV rows.
+            fieldnames: Optional list of field names. If not provided,
+                       uses keys from first data item.
+                       
+        Returns:
+            True if successful, False otherwise.
+        """
+        if not data:
+            print("No data to write")
+            return False
+        
+        try:
+            if fieldnames is None:
+                fieldnames = list(data[0].keys())
+            
+            with open(file_path, 'w', encoding='utf-8', newline='') as f:
+                writer = csv.DictWriter(f, fieldnames=fieldnames)
+                writer.writeheader()
+                writer.writerows(data)
+            return True
+        except Exception as e:
+            print(f"Error writing CSV file {file_path}: {e}")
+            return False
+    
+    @staticmethod
+    def query_csv(file_path: str, filter_func) -> List[Dict[str, Any]]:
+        """Query CSV data with a filter function.
+        
+        Args:
+            file_path: Path to the CSV file.
+            filter_func: Function that takes a row dict and returns bool.
+            
+        Returns:
+            Filtered list of row dictionaries.
+        """
+        data = CSVDataHandler.read_csv(file_path)
+        return [row for row in data if filter_func(row)]
+
+
+def register_csv_tools(tool_handler: CSVToolHandler) -> Dict[str, callable]:
+    """Register CSV tools as callable functions.
+    
+    Args:
+        tool_handler: CSVToolHandler instance with loaded tools.
+        
+    Returns:
+        Dictionary mapping tool names to callable functions.
+    """
+    tool_functions = {}
+    
+    for tool in tool_handler.get_tools():
+        tool_name = tool.get('name', '')
+        tool_description = tool.get('description', '')
+        tool_function = tool.get('function', '')
+        
+        # Create a lambda that captures the tool configuration
+        def create_tool_func(name, desc, func_str):
+            def tool_func(*args, **kwargs):
+                """Dynamically created tool function."""
+                print(f"Executing tool: {name}")
+                print(f"Description: {desc}")
+                print(f"Args: {args}, Kwargs: {kwargs}")
+                # In a real implementation, this would execute the function
+                return {"status": "executed", "tool": name}
+            
+            tool_func.__name__ = name
+            tool_func.__doc__ = desc
+            return tool_func
+        
+        if tool_name:
+            tool_functions[tool_name] = create_tool_func(
+                tool_name, tool_description, tool_function
+            )
+    
+    return tool_functions
+
+
+# Example usage and testing
+if __name__ == "__main__":
+    # Create sample tools CSV for testing
+    sample_tools = [
+        {
+            "name": "read_file",
+            "description": "Read contents of a file",
+            "function": "file_reader",
+            "parameters": "file_path:str"
+        },
+        {
+            "name": "write_file",
+            "description": "Write contents to a file",
+            "function": "file_writer",
+            "parameters": "file_path:str,content:str"
+        },
+        {
+            "name": "search_web",
+            "description": "Search the web for information",
+            "function": "web_search",
+            "parameters": "query:str"
+        }
+    ]
+    
+    # Test CSV operations
+    print("Testing CSV Tool Handler...")
+    handler = CSVToolHandler('./test_tools.csv')
+    
+    for tool in sample_tools:
+        handler.add_tool(tool)
+    
+    handler.save_tools()
+    print(f"Tools available: {[t['name'] for t in handler.get_tools()]}")
+    
+    # Register tools
+    tool_funcs = register_csv_tools(handler)
+    print(f"Registered functions: {list(tool_funcs.keys())}")