Spaces:
Sleeping
Sleeping
File size: 7,102 Bytes
2456578 | 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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | """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())}")
|