From 8c63c694cddf161b039de61cc8b4044502585015 Mon Sep 17 00:00:00 2001 From: Bentlybro Date: Thu, 5 Jun 2025 10:22:54 +0100 Subject: [PATCH] Fix and improve dynamic tool loading --- SimpleAgent/core/execution.py | 103 ++++++++++++++++++++++++++++++- SimpleAgent/core/tool_manager.py | 16 ++++- 2 files changed, 115 insertions(+), 4 deletions(-) diff --git a/SimpleAgent/core/execution.py b/SimpleAgent/core/execution.py index 919a937..69fa272 100644 --- a/SimpleAgent/core/execution.py +++ b/SimpleAgent/core/execution.py @@ -7,6 +7,7 @@ import os import json import time +import inspect from typing import Dict, Any, List, Optional, Tuple, Callable from core.tool_manager import REGISTERED_COMMANDS, COMMAND_SCHEMAS, load_tool @@ -188,7 +189,9 @@ def execute_function(self, function_name: str, function_args: Dict) -> Tuple[Any # Execute the function with sanitized arguments try: - function_response = function_to_call(**function_args) + # Apply dynamic parameter mapping to handle parameter name mismatches + mapped_args = self._map_function_parameters(function_to_call, function_args) + function_response = function_to_call(**mapped_args) print(f"📊 Function result: {function_response}") except UnboundLocalError as e: if "stop_words" in str(e) and function_name == "text_analysis": @@ -205,7 +208,9 @@ def execute_function(self, function_name: str, function_args: Dict) -> Tuple[Any # Retry the function call try: - function_response = function_to_call(**function_args) + # Re-apply parameter mapping after modifying analysis_types + mapped_args = self._map_function_parameters(function_to_call, function_args) + function_response = function_to_call(**mapped_args) print(f"📊 Function result (retry): {function_response}") except Exception as retry_e: print(f"❌ Retry failed: {str(retry_e)}") @@ -270,4 +275,96 @@ def get_next_action(self, conversation_history: List[Dict[str, Any]]) -> Optiona return None except Exception as e: print(f"Error getting next action: {str(e)}") - return None \ No newline at end of file + return None + + def _map_function_parameters(self, function: Callable, function_args: Dict[str, Any]) -> Dict[str, Any]: + """ + Map function arguments using the tool's schema. + + Args: + function: The function to call + function_args: The arguments from the LLM + + Returns: + Mapped arguments that match the function signature + """ + try: + # Get the function signature + sig = inspect.signature(function) + expected_params = set(sig.parameters.keys()) + provided_params = set(function_args.keys()) + + # If all parameters match, no mapping needed + if provided_params.issubset(expected_params): + return function_args + + # Find the schema for this function + function_name = function.__name__ + function_schema = None + + for schema in COMMAND_SCHEMAS: + schema_name = schema.get("function", {}).get("name") + if schema_name == function_name: + function_schema = schema + break + + if not function_schema: + print(f"⚠️ No schema found for {function_name}") + return function_args + + # Get expected parameter names from schema + schema_params = function_schema.get("function", {}).get("parameters", {}).get("properties", {}) + schema_param_names = set(schema_params.keys()) + + mapped_args = {} + + # Map parameters based on schema + for provided_key, value in function_args.items(): + # Direct match with schema + if provided_key in schema_param_names: + mapped_args[provided_key] = value + continue + + # Case-insensitive match + matched = False + for schema_param in schema_param_names: + if provided_key.lower() == schema_param.lower(): + mapped_args[schema_param] = value + print(f"🔧 Parameter mapping: '{provided_key}' -> '{schema_param}' (case insensitive)") + matched = True + break + + # If no match, try partial matching (e.g., 'path' -> 'file_path') + if not matched: + for schema_param in schema_param_names: + # Enhanced partial matching for common variations + provided_lower = provided_key.lower() + schema_lower = schema_param.lower() + + # Check if they share common roots + match_found = False + + # Common file/path parameter variations + if (('file' in provided_lower or 'path' in provided_lower or 'name' in provided_lower) and + ('file' in schema_lower or 'path' in schema_lower)): + match_found = True + + # Standard partial matching (substring check) + elif (provided_lower in schema_lower or schema_lower in provided_lower): + match_found = True + + if match_found and schema_param not in mapped_args: + mapped_args[schema_param] = value + print(f"🔧 Parameter mapping: '{provided_key}' -> '{schema_param}' (partial match)") + matched = True + break + + # If still no match, keep the original parameter name + if not matched: + mapped_args[provided_key] = value + + return mapped_args + + except Exception as e: + print(f"⚠️ Parameter mapping failed: {e}") + return function_args \ No newline at end of file diff --git a/SimpleAgent/core/tool_manager.py b/SimpleAgent/core/tool_manager.py index 4cb8a9d..9f0dd54 100644 --- a/SimpleAgent/core/tool_manager.py +++ b/SimpleAgent/core/tool_manager.py @@ -74,7 +74,21 @@ def register_command(self, name: str, func: Callable, schema: Dict[str, Any]) -> schema: The OpenAI function schema for the command """ REGISTERED_COMMANDS[name] = func - COMMAND_SCHEMAS.append(schema) + + # Remove any existing placeholder schema for this command + existing_schema_index = None + for i, existing_schema in enumerate(COMMAND_SCHEMAS): + if existing_schema.get("function", {}).get("name") == name: + existing_schema_index = i + break + + if existing_schema_index is not None: + # Replace the placeholder schema with the real one + COMMAND_SCHEMAS[existing_schema_index] = schema + self.logger.debug(f'Replaced placeholder schema for {name} with real schema') + else: + # No existing schema, append the new one + COMMAND_SCHEMAS.append(schema) # Determine category from the module path module = func.__module__