Clean up tool config and old code cleanup

Author: marktayl1

src/backend/kernel_agents/agent_base.py

@@ -8,7 +8,6 @@
 from semantic_kernel.functions import KernelFunction
 from semantic_kernel.functions.kernel_arguments import KernelArguments
 from semantic_kernel.functions.kernel_function_decorator import kernel_function
-from semantic_kernel.agents import AzureAIAgentThread
 
 # Updated imports for compatibility
 try:
@@ -78,18 +77,7 @@ def __init__(
             definition: The definition required by AzureAIAgent
         """
         # Add plugins if not already set
-        # if not self.plugins:
-        # If agent_type is provided, load tools from config automatically
-        if agent_type and not tools:
-            tools = self.get_tools_from_config(kernel, agent_type)
-            # If system_message isn't provided, try to get it from config
-            if not system_message:
-                config = self.load_tools_config(agent_type)
-                system_message = config.get(
-                    "system_message", self._default_system_message(agent_name)
-                )
-        else:
-            tools = tools or []
+        tools = tools or []
         system_message = system_message or self._default_system_message(agent_name)
 
         # Call AzureAIAgent constructor with required client and definition
@@ -120,15 +108,6 @@ def __init__(
         # Required properties for AgentGroupChat compatibility
         self.name = agent_name  # This is crucial for AgentGroupChat to identify agents
 
-    # @property
-    # def plugins(self) -> Optional[dict[str, Callable]]:
-    #     """Get the plugins for this agent.
-
-    #     Returns:
-    #         A list of plugins, or None if not applicable.
-    #     """
-    #     return None
-
     def _default_system_message(self, agent_name=None) -> str:
         name = agent_name or getattr(self, "_agent_name", "Agent")
         return f"You are an AI assistant named {name}. Help the user by providing accurate and helpful information."
@@ -186,14 +165,10 @@ async def handle_action_request(self, action_request: ActionRequest) -> str:
         )
 
         try:
-            # Use the agent to process the action
-            # chat_history = self._chat_history.copy()
 
             # Call the agent to handle the action
             thread = None
-            # thread = self.client.agents.get_thread(
-            #     thread=step.session_id
-            # )  # AzureAIAgentThread(thread_id=step.session_id)
+
             async_generator = self._agent.invoke(
                 messages=f"{action_request.action}\n\nPlease perform this action",
                 thread=thread,
@@ -293,179 +268,6 @@ async def handle_action_request(self, action_request: ActionRequest) -> str:
 
         return response.json()
 
-    @staticmethod
-    def create_dynamic_function(
-        name: str,
-        response_template: str,
-        description: Optional[str] = None,
-        formatting_instr: str = DEFAULT_FORMATTING_INSTRUCTIONS,
-    ) -> Callable[..., Awaitable[str]]:
-        """Create a dynamic function for agent tools based on the name and template.
-
-        Args:
-            name: The name of the function to create
-            response_template: The template string to use for the response
-            formatting_instr: Optional formatting instructions to append to the response
-
-        Returns:
-            A dynamic async function that can be registered with the semantic kernel
-        """
-
-        # Truncate function name to 64 characters if it exceeds the limit
-        if len(name) > 64:
-            logging.warning(
-                f"Function name '{name}' exceeds 64 characters (length: {len(name)}). Truncating to 64 characters."
-            )
-            name = name[:64]
-
-        async def dynamic_function(**kwargs) -> str:
-            try:
-                # Format the template with the provided kwargs
-                formatted_response = response_template.format(**kwargs)
-                # Append formatting instructions if not already included in the template
-                if formatting_instr and formatting_instr not in formatted_response:
-                    formatted_response = f"{formatted_response}\n{formatting_instr}"
-                return formatted_response
-            except KeyError as e:
-                return f"Error: Missing parameter {e} for {name}"
-            except Exception as e:
-                return f"Error processing {name}: {str(e)}"
-
-        # Name the function properly for better debugging
-        dynamic_function.__name__ = name
-
-        # Create a wrapped kernel function that matches the expected signature
-        logging.info(f"Creating dynamic function: {name} {len(name)}")
-
-        @kernel_function(description=f"Dynamic function {name}", name=name)
-        async def kernel_wrapper(
-            kernel_arguments: KernelArguments = None, **kwargs
-        ) -> str:
-            # Combine all arguments into one dictionary
-            all_args = {}
-            if kernel_arguments:
-                for key, value in kernel_arguments.items():
-                    all_args[key] = value
-            all_args.update(kwargs)
-            return await dynamic_function(**all_args)
-
-        return kernel_wrapper
-
-    @staticmethod
-    def load_tools_config(
-        filename: str, config_path: Optional[str] = None
-    ) -> Dict[str, Any]:
-        """Load tools configuration from a JSON file.
-
-        Args:
-            filename: The filename without extension (e.g., "hr", "marketing")
-            config_path: Optional explicit path to the configuration file
-
-        Returns:
-            A dictionary containing the configuration
-        """
-        if config_path is None:
-            # Default path relative to the tools directory
-            current_dir = os.path.dirname(os.path.abspath(__file__))
-            backend_dir = os.path.dirname(
-                current_dir
-            )  # Just one level up to get to backend dir
-
-            # Normalize filename to avoid issues with spaces and capitalization
-            # Convert "Hr Agent" to "hr" and "TechSupport Agent" to "tech_support"
-            logging.info(f"Normalizing filename: {filename}")
-            normalized_filename = filename.replace(" ", "_").replace("-", "_").lower()
-            # If it ends with "_agent", remove it
-            if normalized_filename.endswith("_agent"):
-                normalized_filename = normalized_filename[:-6]
-
-            config_path = os.path.join(
-                backend_dir, "tools", f"{normalized_filename}_tools.json"
-            )
-            logging.info(f"Looking for tools config at: {config_path}")
-
-        try:
-            with open(config_path, "r") as f:
-                return json.load(f)
-        except Exception as e:
-            logging.error(f"Error loading {filename} tools configuration: {e}")
-            # Return empty default configuration
-            return {
-                "agent_name": f"{filename.capitalize()}Agent",
-                "system_message": "You are an AI assistant",
-                "tools": [],
-            }
-
-    @classmethod
-    def get_tools_from_config(
-        cls, kernel: sk.Kernel, agent_type: str, config_path: Optional[str] = None
-    ) -> List[KernelFunction]:
-        """Get the list of tools for an agent from configuration.
-
-        Args:
-            kernel: The semantic kernel instance
-            agent_type: The type of agent (e.g., "marketing", "hr")
-            config_path: Optional explicit path to the configuration file
-
-        Returns:
-            A list of KernelFunction objects representing the tools
-        """
-        # Load configuration
-        config = cls.load_tools_config(agent_type, config_path)
-
-        # Convert the configured tools to kernel functions
-        kernel_functions = []
-        plugin_name = f"{agent_type}_plugin"
-
-        # Early return if no tools defined - prevent empty iteration
-        if not config.get("tools"):  # or agent_type == "Product_Agent":
-            logging.info(
-                f"No tools defined for agent type '{agent_type}'. Returning empty list."
-            )
-            return kernel_functions
-
-        for tool in config.get("tools", []):
-            try:
-                function_name = tool["name"]
-                description = tool.get("description", "")
-                # Create a dynamic function using the JSON response_template
-                response_template = (
-                    tool.get("response_template") or tool.get("prompt_template") or ""
-                )
-
-                # Generate a dynamic function using our improved approach
-                dynamic_fn = cls.create_dynamic_function(
-                    function_name, response_template
-                )
-
-                # Create kernel function from the decorated function
-                kernel_func = KernelFunction.from_method(dynamic_fn)
-
-                # Add parameter metadata from JSON to the kernel function
-                for param in tool.get("parameters", []):
-                    param_name = param.get("name", "")
-                    param_desc = param.get("description", "")
-                    param_type = param.get("type", "string")
-
-                    # Set this parameter in the function's metadata
-                    if param_name:
-                        logging.info(
-                            f"Adding parameter '{param_name}' to function '{function_name}'"
-                        )
-
-                # Register the function with the kernel
-
-                kernel_functions.append(kernel_func)
-                logging.info(
-                    f"Successfully created dynamic tool '{function_name}' for {agent_type}"
-                )
-            except Exception as e:
-                logging.error(
-                    f"Failed to create tool '{tool.get('name', 'unknown')}': {str(e)}"
-                )
-
-        return kernel_functions
-
     def save_state(self) -> Mapping[str, Any]:
         """Save the state of this agent."""
         return {"memory": self._memory_store.save_state()}

src/backend/kernel_agents/agent_factory.py

@@ -1,36 +1,34 @@
 """Factory for creating agents in the Multi-Agent Custom Automation Engine."""
 
+import inspect
 import logging
-from typing import Dict, List, Callable, Any, Optional, Type
 from types import SimpleNamespace
-from semantic_kernel import Kernel
-from semantic_kernel.functions import KernelFunction
-from semantic_kernel.agents.azure_ai.azure_ai_agent import AzureAIAgent
-import inspect
-
-from kernel_agents.agent_base import BaseAgent
+from typing import Any, Callable, Dict, List, Optional, Type
 
 # Import the new AppConfig instance
 from app_config import config
+from azure.ai.projects.models import (
+    ResponseFormatJsonSchema,
+    ResponseFormatJsonSchemaType,
+)
+from context.cosmos_memory_kernel import CosmosMemoryContext
+from kernel_agents.agent_base import BaseAgent
+from kernel_agents.generic_agent import GenericAgent
+from kernel_agents.group_chat_manager import GroupChatManager
 
 # Import all specialized agent implementations
 from kernel_agents.hr_agent import HrAgent
 from kernel_agents.human_agent import HumanAgent
 from kernel_agents.marketing_agent import MarketingAgent
-from kernel_agents.generic_agent import GenericAgent
-from kernel_agents.tech_support_agent import TechSupportAgent
+from kernel_agents.planner_agent import PlannerAgent  # Add PlannerAgent import
 from kernel_agents.procurement_agent import ProcurementAgent
 from kernel_agents.product_agent import ProductAgent
-from kernel_agents.planner_agent import PlannerAgent  # Add PlannerAgent import
-from kernel_agents.group_chat_manager import GroupChatManager
+from kernel_agents.tech_support_agent import TechSupportAgent
+from models.messages_kernel import AgentType, PlannerResponsePlan
+from semantic_kernel import Kernel
+from semantic_kernel.agents.azure_ai.azure_ai_agent import AzureAIAgent
+from semantic_kernel.functions import KernelFunction
 from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
-from context.cosmos_memory_kernel import CosmosMemoryContext
-from models.messages_kernel import PlannerResponsePlan, AgentType
-
-from azure.ai.projects.models import (
-    ResponseFormatJsonSchema,
-    ResponseFormatJsonSchemaType,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -176,7 +174,7 @@ async def create_agent(
         agent_type_str = cls._agent_type_strings.get(
             agent_type, agent_type.value.lower()
         )
-        tools = await cls._load_tools_for_agent(kernel, agent_type_str)
+        # tools = await cls._load_tools_for_agent(kernel, agent_type_str)
 
         # Build the agent definition (functions schema)
         definition = None
@@ -223,7 +221,7 @@ async def create_agent(
                     "session_id": session_id,
                     "user_id": user_id,
                     "memory_store": memory_store,
-                    "tools": tools,
+                    "tools": [],
                     "system_message": system_message,
                     "client": client,
                     "definition": definition,
@@ -253,37 +251,6 @@ async def create_agent(
 
         return agent
 
-    @classmethod
-    async def _load_tools_for_agent(
-        cls, kernel: Kernel, agent_type: str
-    ) -> List[KernelFunction]:
-        """Load tools for an agent from the tools directory.
-
-        This tries to load tool configurations from JSON files. If that fails,
-        it returns an empty list for agents that don't need tools.
-
-        Args:
-            kernel: The semantic kernel instance
-            agent_type: The agent type string identifier
-
-        Returns:
-            A list of kernel functions for the agent
-        """
-        try:
-            # Try to use the BaseAgent's tool loading mechanism
-            tools = BaseAgent.get_tools_from_config(kernel, agent_type)
-            logger.info(f"Successfully loaded {len(tools)} tools for {agent_type}")
-            return tools
-        except FileNotFoundError:
-            # No tool configuration file found - this is expected for some agents
-            logger.info(
-                f"No tools defined for agent type '{agent_type}'. Returning empty list."
-            )
-            return []
-        except Exception as e:
-            logger.warning(f"Error loading tools for {agent_type}: {e}")
-            return []
-
     @classmethod
     async def create_all_agents(
         cls, session_id: str, user_id: str, temperature: float = 0.0

src/backend/kernel_agents/generic_agent.py

@@ -2,12 +2,11 @@
 from typing import List, Optional
 
 import semantic_kernel as sk
-from semantic_kernel.functions import KernelFunction
-
-from kernel_agents.agent_base import BaseAgent
 from context.cosmos_memory_kernel import CosmosMemoryContext
+from kernel_agents.agent_base import BaseAgent
+from kernel_tools.generic_tools import GenericTools
 from models.messages_kernel import AgentType
-from src.backend.kernel_tools.generic_tools import GenericTools
+from semantic_kernel.functions import KernelFunction
 
 
 class GenericAgent(BaseAgent):
@@ -42,27 +41,15 @@ def __init__(
         """
         # Load configuration if tools not provided
         if tools is None:
-            # Get tools directly from GenericTools class
-            tools_dict = GenericTools.get_all_kernel_functions()
-            logging.info(
-                f"GenericAgent: Got tools_dict with {len(tools_dict)} functions: {list(tools_dict.keys())}"
-            )
-
-            tools = [KernelFunction.from_method(func) for func in tools_dict.values()]
-            logging.info(
-                f"GenericAgent: Created {len(tools)} KernelFunctions from tools_dict"
-            )
-
-            # Load the generic tools configuration for system message
-            config = self.load_tools_config("generic", config_path)
+            tools = [GenericTools]
 
             # Use system message from config if not explicitly provided
             if not system_message:
-                system_message = config.get(
-                    "system_message",
-                    "You are a generic agent. You are used to handle generic tasks that a general Large Language Model can assist with. "
-                    "You are being called as a fallback, when no other agents are able to use their specialised functions in order to solve "
-                    "the user's task. Summarize back to the user what was done.",
+                system_message = (
+                    "system_message"
+                    + "You are a generic agent. You are used to handle generic tasks that a general Large Language Model can assist with. "
+                    + "You are being called as a fallback, when no other agents are able to use their specialized functions in order to solve "
+                    + "the user's task. Summarize back to the user what was done.",
                 )
 
             # Use agent name from config if available