Spaces:
Sleeping
Sleeping
| """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.""" | |
| 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 [] | |
| 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 | |
| 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())}") | |