Spaces:
Running
Running
""" | |
AGNO-compatible wrapper for Phase 3 mathematical code execution tools. | |
This module provides AGNO-compatible tool classes that wrap the existing | |
Phase 3 mathematical code execution functionality to resolve Pydantic validation errors. | |
""" | |
import logging | |
from typing import Any, Dict, List, Optional | |
from tools.code_execution_tool import get_code_execution_tools | |
from tools.mathematical_engine import get_mathematical_engine_tools | |
from tools.code_analyzer import get_code_analyzer_tools | |
logger = logging.getLogger(__name__) | |
class AGNOCodeExecutionTool: | |
"""AGNO-compatible wrapper for Phase 3 code execution tools.""" | |
def __init__(self): | |
"""Initialize the AGNO-compatible code execution tool.""" | |
self.name = "agno_code_execution" | |
self.description = "Execute Python code securely with comprehensive error handling and result formatting" | |
self.available = False | |
self._phase3_tools = {} | |
try: | |
# Get Phase 3 code execution tools | |
phase3_tools = get_code_execution_tools() | |
if phase3_tools: | |
# Store the Phase 3 tools by name for easy access | |
for tool in phase3_tools: | |
self._phase3_tools[tool['name']] = tool['function'] | |
self.available = True | |
logger.info(f"β AGNO Code Execution Tool initialized with {len(phase3_tools)} Phase 3 tools") | |
else: | |
logger.warning("β οΈ No Phase 3 code execution tools available") | |
except Exception as e: | |
logger.error(f"β Failed to initialize AGNO Code Execution Tool: {e}") | |
def execute_python_code(self, code: str, timeout: int = 30) -> Dict[str, Any]: | |
"""Execute Python code using Phase 3 code execution functionality.""" | |
if not self.available or 'execute_python_code' not in self._phase3_tools: | |
return { | |
'success': False, | |
'error': 'Code execution tool not available', | |
'result': None | |
} | |
try: | |
# Use the Phase 3 code execution function | |
phase3_func = self._phase3_tools['execute_python_code'] | |
result = phase3_func(code, timeout) | |
return result | |
except Exception as e: | |
logger.error(f"β Code execution failed: {e}") | |
return { | |
'success': False, | |
'error': str(e), | |
'result': None | |
} | |
def analyze_code_structure(self, code: str) -> Dict[str, Any]: | |
"""Analyze code structure using Phase 3 functionality.""" | |
if not self.available or 'analyze_code_structure' not in self._phase3_tools: | |
return { | |
'success': False, | |
'error': 'Code analysis tool not available', | |
'result': None | |
} | |
try: | |
# Use the Phase 3 code analysis function | |
phase3_func = self._phase3_tools['analyze_code_structure'] | |
result = phase3_func(code) | |
return result | |
except Exception as e: | |
logger.error(f"β Code analysis failed: {e}") | |
return { | |
'success': False, | |
'error': str(e), | |
'result': None | |
} | |
class AGNOMathematicalEngineTool: | |
"""AGNO-compatible wrapper for Phase 3 mathematical engine tools.""" | |
def __init__(self): | |
"""Initialize the AGNO-compatible mathematical engine tool.""" | |
self.name = "agno_mathematical_engine" | |
self.description = "Advanced mathematical computations using SymPy, NumPy, and SciPy" | |
self.available = False | |
self._phase3_tools = {} | |
try: | |
# Get Phase 3 mathematical engine tools | |
phase3_tools = get_mathematical_engine_tools() | |
if phase3_tools: | |
# Store the Phase 3 tools by name for easy access | |
for tool in phase3_tools: | |
self._phase3_tools[tool['name']] = tool['function'] | |
self.available = True | |
logger.info(f"β AGNO Mathematical Engine Tool initialized with {len(phase3_tools)} Phase 3 tools") | |
else: | |
logger.warning("β οΈ No Phase 3 mathematical engine tools available") | |
except Exception as e: | |
logger.error(f"β Failed to initialize AGNO Mathematical Engine Tool: {e}") | |
def solve_mathematical_expression(self, expression: str, variables: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: | |
"""Solve mathematical expressions using Phase 3 mathematical engine.""" | |
if not self.available or 'solve_mathematical_expression' not in self._phase3_tools: | |
return { | |
'success': False, | |
'error': 'Mathematical engine tool not available', | |
'result': None | |
} | |
try: | |
# Use the Phase 3 mathematical engine function | |
phase3_func = self._phase3_tools['solve_mathematical_expression'] | |
result = phase3_func(expression, variables) | |
return result | |
except Exception as e: | |
logger.error(f"β Mathematical computation failed: {e}") | |
return { | |
'success': False, | |
'error': str(e), | |
'result': None | |
} | |
def compute_numerical_analysis(self, operation: str, data: List[float], **kwargs) -> Dict[str, Any]: | |
"""Perform numerical analysis using Phase 3 functionality.""" | |
if not self.available or 'compute_numerical_analysis' not in self._phase3_tools: | |
return { | |
'success': False, | |
'error': 'Numerical analysis tool not available', | |
'result': None | |
} | |
try: | |
# Use the Phase 3 numerical analysis function | |
phase3_func = self._phase3_tools['compute_numerical_analysis'] | |
result = phase3_func(operation, data, **kwargs) | |
return result | |
except Exception as e: | |
logger.error(f"β Numerical analysis failed: {e}") | |
return { | |
'success': False, | |
'error': str(e), | |
'result': None | |
} | |
class AGNOCodeAnalyzerTool: | |
"""AGNO-compatible wrapper for Phase 3 code analyzer tools.""" | |
def __init__(self): | |
"""Initialize the AGNO-compatible code analyzer tool.""" | |
self.name = "agno_code_analyzer" | |
self.description = "Comprehensive code analysis including complexity, security, and quality metrics" | |
self.available = False | |
self._phase3_tools = {} | |
try: | |
# Get Phase 3 code analyzer tools | |
phase3_tools = get_code_analyzer_tools() | |
if phase3_tools: | |
# Store the Phase 3 tools by name for easy access | |
for tool in phase3_tools: | |
self._phase3_tools[tool['name']] = tool['function'] | |
self.available = True | |
logger.info(f"β AGNO Code Analyzer Tool initialized with {len(phase3_tools)} Phase 3 tools") | |
else: | |
logger.warning("β οΈ No Phase 3 code analyzer tools available") | |
except Exception as e: | |
logger.error(f"β Failed to initialize AGNO Code Analyzer Tool: {e}") | |
def analyze_code_quality(self, code: str) -> Dict[str, Any]: | |
"""Analyze code quality using Phase 3 functionality.""" | |
if not self.available or 'analyze_code_quality' not in self._phase3_tools: | |
return { | |
'success': False, | |
'error': 'Code quality analyzer not available', | |
'result': None | |
} | |
try: | |
# Use the Phase 3 code quality analysis function | |
phase3_func = self._phase3_tools['analyze_code_quality'] | |
result = phase3_func(code) | |
return result | |
except Exception as e: | |
logger.error(f"β Code quality analysis failed: {e}") | |
return { | |
'success': False, | |
'error': str(e), | |
'result': None | |
} | |
def check_code_security(self, code: str) -> Dict[str, Any]: | |
"""Check code security using Phase 3 functionality.""" | |
if not self.available or 'check_code_security' not in self._phase3_tools: | |
return { | |
'success': False, | |
'error': 'Code security checker not available', | |
'result': None | |
} | |
try: | |
# Use the Phase 3 code security check function | |
phase3_func = self._phase3_tools['check_code_security'] | |
result = phase3_func(code) | |
return result | |
except Exception as e: | |
logger.error(f"β Code security check failed: {e}") | |
return { | |
'success': False, | |
'error': str(e), | |
'result': None | |
} | |
def create_agno_compatible_math_tools() -> List[Any]: | |
"""Create AGNO-compatible mathematical code execution tools.""" | |
tools = [] | |
# Create AGNO-compatible code execution tool | |
code_exec_tool = AGNOCodeExecutionTool() | |
if code_exec_tool.available: | |
tools.append(code_exec_tool) | |
# Create AGNO-compatible mathematical engine tool | |
math_engine_tool = AGNOMathematicalEngineTool() | |
if math_engine_tool.available: | |
tools.append(math_engine_tool) | |
# Create AGNO-compatible code analyzer tool | |
code_analyzer_tool = AGNOCodeAnalyzerTool() | |
if code_analyzer_tool.available: | |
tools.append(code_analyzer_tool) | |
logger.info(f"β Created {len(tools)} AGNO-compatible mathematical code execution tools") | |
return tools |