Complete cache removal from ConfigLoader system

Implemented comprehensive cache removal plan to simplify ConfigLoader architecture:

Core Implementation Changes:
- Removed cache dictionaries from all ConfigLoaders (AgentConfigLoader, GraphConfigLoader, SwarmConfigLoader, ToolConfigLoader)
- Removed cache_key parameters from all load_* methods
- Removed cache check and storage logic throughout system
- Removed clear_cache() and get_available_* methods from all loaders
- Simplified module loading by removing _loaded_modules cache

Method Signature Updates:
- load_agent(config) - removed cache_key parameter
- load_graph(config) - removed cache_key parameter
- load_swarm(config) - removed cache_key parameter
- Updated all cross-loader dependencies to use new signatures

Test Updates:
- Replaced cache-specific tests with object independence tests
- Added test_multiple_loads_create_independent_objects across all loaders
- Removed 7 cache-related test methods
- All 113 tests passing with new architecture

Benefits Achieved:
- Cleaner code: Removed ~200+ lines of cache-related complexity
- Predictable behavior: Each load creates fresh objects
- Easier testing: No cache state management needed
- Reduced memory usage: No indefinite object retention
- Simpler API: No cache management methods or parameters
- Immediate configuration change reflection

Breaking Changes:
- cache_key parameter removed from all load_* methods
- clear_cache() methods removed from all loaders
- get_available_* methods removed from all loaders

Migration: Users can implement application-level caching if needed.
Performance: Slight increase in load time, significant reduction in memory usage and complexity.

Ready for production deployment with simplified, maintainable architecture.

Author: vawsgit

src/strands/experimental/config_loader/agent/agent_config_loader.py

@@ -52,7 +52,6 @@ def __init__(self, tool_config_loader: Optional["ToolConfigLoader"] = None):
                                If not provided, will be imported and created when needed.
         """
         self._tool_config_loader = tool_config_loader
-        self._agent_cache: Dict[str, Agent] = {}
         self.schema_registry = SchemaRegistry()
         self._global_schemas_loaded = False
         self._structured_output_defaults: Dict[str, Any] = {}
@@ -72,12 +71,11 @@ def _get_tool_config_loader(self) -> "ToolConfigLoader":
             self._tool_config_loader = ToolConfigLoader()
         return self._tool_config_loader
 
-    def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) -> Agent:
+    def load_agent(self, config: Dict[str, Any]) -> Agent:
         """Load an Agent from a dictionary configuration.
 
         Args:
             config: Dictionary containing agent configuration with top-level 'agent' key.
-            cache_key: Optional key for caching the loaded agent.
 
         Returns:
             Agent instance configured according to the provided dictionary.
@@ -94,11 +92,6 @@ def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         if not isinstance(agent_config, dict):
             raise ValueError("The 'agent' configuration must be a dictionary")
 
-        # Check cache first
-        if cache_key and cache_key in self._agent_cache:
-            logger.debug("agent_cache_key=<%s> | found in cache", cache_key)
-            return self._agent_cache[cache_key]
-
         # Load global schemas if present and not already loaded
         if not self._global_schemas_loaded and "schemas" in config:
             self._load_global_schemas(config["schemas"])
@@ -180,11 +173,6 @@ def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         if "structured_output" in agent_config:
             self._configure_agent_structured_output(agent, agent_config["structured_output"])
 
-        # Cache the agent if cache key provided
-        if cache_key:
-            self._agent_cache[cache_key] = agent
-            logger.debug("agent_cache_key=<%s> | cached agent", cache_key)
-
         return agent
 
     def serialize_agent(self, agent: Agent) -> Dict[str, Any]:
@@ -435,19 +423,6 @@ def _load_session_manager(self, sm_config: Optional[Dict[str, Any]]) -> Optional
         logger.warning("Session manager loading from configuration is not yet implemented")
         return None
 
-    def clear_cache(self) -> None:
-        """Clear the internal agent cache."""
-        self._agent_cache.clear()
-        logger.debug("agent cache cleared")
-
-    def get_available_agents(self) -> List[str]:
-        """Get list of cached agent keys.
-
-        Returns:
-            List of cached agent keys.
-        """
-        return list(self._agent_cache.keys())
-
     def _load_global_schemas(self, schemas_config: List[Dict[str, Any]]) -> None:
         """Load global schema registry from configuration.
 

src/strands/experimental/config_loader/graph/graph_config_loader.py

@@ -52,7 +52,6 @@ def __init__(
         self._agent_loader = agent_loader
         self._swarm_loader = swarm_loader
         self._condition_registry = ConditionRegistry()
-        self._graph_cache: Dict[str, Graph] = {}
 
     def _get_agent_config_loader(self) -> "AgentConfigLoader":
         """Get or create an AgentConfigLoader instance.
@@ -84,12 +83,11 @@ def _get_swarm_config_loader(self) -> "SwarmConfigLoader":
             self._swarm_loader = SwarmConfigLoader()
         return self._swarm_loader
 
-    def load_graph(self, config: Dict[str, Any], cache_key: Optional[str] = None) -> Graph:
-        """Load a Graph from YAML configuration (loaded as dictionary).
+    def load_graph(self, config: Dict[str, Any]) -> Graph:
+        """Load a Graph from configuration dictionary.
 
         Args:
             config: Dictionary containing graph configuration with top-level 'graph' key.
-            cache_key: Optional key for caching the loaded graph.
 
         Returns:
             Graph instance configured according to the provided dictionary.
@@ -106,11 +104,6 @@ def load_graph(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         if not isinstance(graph_config, dict):
             raise ValueError("The 'graph' configuration must be a dictionary")
 
-        # Check cache first
-        if cache_key and cache_key in self._graph_cache:
-            logger.debug("graph_cache_key=<%s> | found in cache", cache_key)
-            return self._graph_cache[cache_key]
-
         # Validate configuration structure
         self._validate_config(graph_config)
 
@@ -129,11 +122,6 @@ def load_graph(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         # Create graph
         graph = Graph(nodes=nodes, edges=edges, entry_points=entry_points, **graph_params)
 
-        # Cache the graph if cache key provided
-        if cache_key:
-            self._graph_cache[cache_key] = graph
-            logger.debug("graph_cache_key=<%s> | cached graph", cache_key)
-
         return graph
 
     def serialize_graph(self, graph: Graph) -> Dict[str, Any]:
@@ -183,19 +171,6 @@ def serialize_graph(self, graph: Graph) -> Dict[str, Any]:
 
         return {"graph": graph_config}
 
-    def clear_cache(self) -> None:
-        """Clear the internal graph cache."""
-        self._graph_cache.clear()
-        logger.debug("graph cache cleared")
-
-    def get_available_graphs(self) -> List[str]:
-        """Get list of cached graph keys.
-
-        Returns:
-            List of cached graph keys.
-        """
-        return list(self._graph_cache.keys())
-
     def _load_nodes(self, nodes_config: List[Dict[str, Any]]) -> Dict[str, GraphNode]:
         """Load graph nodes from configuration.
 

src/strands/experimental/config_loader/swarm/swarm_config_loader.py

@@ -40,7 +40,6 @@ def __init__(self, agent_config_loader: Optional["AgentConfigLoader"] = None):
                                 If not provided, will be imported and created when needed.
         """
         self._agent_config_loader = agent_config_loader
-        self._swarm_cache: Dict[str, Swarm] = {}
 
     def _get_agent_config_loader(self) -> "AgentConfigLoader":
         """Get or create an AgentConfigLoader instance.
@@ -57,12 +56,11 @@ def _get_agent_config_loader(self) -> "AgentConfigLoader":
             self._agent_config_loader = AgentConfigLoader()
         return self._agent_config_loader
 
-    def load_swarm(self, config: Dict[str, Any], cache_key: Optional[str] = None) -> Swarm:
-        """Load a Swarm from YAML configuration (loaded as dictionary).
+    def load_swarm(self, config: Dict[str, Any]) -> Swarm:
+        """Load a Swarm from configuration dictionary.
 
         Args:
             config: Dictionary containing swarm configuration with top-level 'swarm' key.
-            cache_key: Optional key for caching the loaded swarm.
 
         Returns:
             Swarm instance configured according to the provided dictionary.
@@ -79,11 +77,6 @@ def load_swarm(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         if not isinstance(swarm_config, dict):
             raise ValueError("The 'swarm' configuration must be a dictionary")
 
-        # Check cache first
-        if cache_key and cache_key in self._swarm_cache:
-            logger.debug("swarm_cache_key=<%s> | found in cache", cache_key)
-            return self._swarm_cache[cache_key]
-
         # Validate configuration structure
         self._validate_config(swarm_config)
 
@@ -101,11 +94,6 @@ def load_swarm(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         # Create swarm
         swarm = Swarm(nodes=agents, **swarm_params)
 
-        # Cache the swarm if cache key provided
-        if cache_key:
-            self._swarm_cache[cache_key] = swarm
-            logger.debug("swarm_cache_key=<%s> | cached swarm", cache_key)
-
         return swarm
 
     def serialize_swarm(self, swarm: Swarm) -> Dict[str, Any]:
@@ -223,14 +211,12 @@ def load_agents(self, agents_config: List[Dict[str, Any]]) -> List[Agent]:
             if "model" not in agent_config:
                 raise ValueError(f"Agent configuration at index {i} must include 'model' field")
 
-            # Use agent name as cache key for individual agent caching
             agent_name = agent_config["name"]
-            cache_key = f"agent:{agent_name}"
 
             try:
                 # Wrap the agent config in the required top-level 'agent' key
                 wrapped_agent_config = {"agent": agent_config}
-                agent = agent_loader.load_agent(wrapped_agent_config, cache_key=cache_key)
+                agent = agent_loader.load_agent(wrapped_agent_config)
                 agents.append(agent)
                 logger.debug("agent_name=<%s> | loaded agent for swarm", agent_name)
             except Exception as e:
@@ -329,16 +315,3 @@ def _validate_config(self, config: Dict[str, Any]) -> None:
                 value = config[param_name]
                 if not isinstance(value, (int, float)):
                     raise ValueError(f"{param_name} must be a number, got: {type(value)}")
-
-    def clear_cache(self) -> None:
-        """Clear the internal swarm cache."""
-        self._swarm_cache.clear()
-        logger.debug("swarm cache cleared")
-
-    def get_available_swarms(self) -> List[str]:
-        """Get list of cached swarm keys.
-
-        Returns:
-            List of cached swarm keys.
-        """
-        return list(self._swarm_cache.keys())

src/strands/experimental/config_loader/tools/tool_config_loader.py

@@ -680,8 +680,6 @@ def __init__(self, registry: Optional[ToolRegistry] = None):
                      If not provided, a new registry will be created.
         """
         self._registry = registry or ToolRegistry()
-        self._loaded_modules: Dict[str, Any] = {}
-        self._tool_cache: Dict[str, AgentTool] = {}
         self._agent_config_loader: Optional["AgentConfigLoader"] = None
         self._swarm_config_loader: Optional["SwarmConfigLoader"] = None
         self._graph_config_loader: Optional["GraphConfigLoader"] = None
@@ -856,11 +854,6 @@ def _load_swarm_as_tool(self, tool_config: Dict[str, Any]) -> AgentTool:
         if not swarm_config:
             raise ValueError("Swarm tool configuration must include 'swarm' field")
 
-        # Check cache first
-        cache_key = f"swarm:{tool_name}:{hash(str(sorted(tool_config.items())))}"
-        if cache_key in self._tool_cache:
-            return self._tool_cache[cache_key]
-
         try:
             # Load the swarm using SwarmConfigLoader
             # Wrap the swarm config in the required top-level 'swarm' key
@@ -878,10 +871,6 @@ def _load_swarm_as_tool(self, tool_config: Dict[str, Any]) -> AgentTool:
                 entry_agent=entry_agent,
             )
 
-            # Cache the result
-            self._tool_cache[cache_key] = swarm_tool
-            logger.debug("swarm_tool=<%s> | loaded and cached", tool_name)
-
             return swarm_tool
 
         except Exception as e:
@@ -1036,11 +1025,6 @@ def _load_graph_as_tool(self, tool_config: Dict[str, Any]) -> AgentTool:
         if not graph_config:
             raise ValueError("Graph tool configuration must include 'graph' field")
 
-        # Check cache first
-        cache_key = f"graph:{tool_name}:{hash(str(sorted(tool_config.items())))}"
-        if cache_key in self._tool_cache:
-            return self._tool_cache[cache_key]
-
         try:
             # Load the graph using GraphConfigLoader
             graph_loader = self._get_graph_config_loader()
@@ -1062,10 +1046,6 @@ def _load_graph_as_tool(self, tool_config: Dict[str, Any]) -> AgentTool:
                 entry_point=entry_point,
             )
 
-            # Cache the result
-            self._tool_cache[cache_key] = graph_tool
-            logger.debug("graph_tool=<%s> | loaded and cached", tool_name)
-
             return graph_tool
 
         except Exception as e:
@@ -1140,11 +1120,6 @@ def _load_agent_as_tool(self, tool_config: Dict[str, Any]) -> AgentTool:
         # Extract prompt template from agent config
         prompt = agent_config.get("prompt")
 
-        # Check cache first
-        cache_key = f"agent:{tool_name}:{hash(str(sorted(tool_config.items())))}"
-        if cache_key in self._tool_cache:
-            return self._tool_cache[cache_key]
-
         try:
             # Load the agent using AgentConfigLoader
             # Wrap the agent config in the required top-level 'agent' key
@@ -1157,10 +1132,6 @@ def _load_agent_as_tool(self, tool_config: Dict[str, Any]) -> AgentTool:
                 agent=agent, tool_name=tool_name, description=description, input_schema=input_schema, prompt=prompt
             )
 
-            # Cache the result
-            self._tool_cache[cache_key] = agent_tool
-            logger.debug("agent_tool=<%s> | loaded and cached", tool_name)
-
             return agent_tool
 
         except Exception as e:
@@ -1177,11 +1148,6 @@ def _load_string_tool(self, identifier: str, module_path: Optional[str] = None)
         Returns:
             AgentTool instance for the specified identifier.
         """
-        # Check cache first
-        cache_key = f"{module_path or ''}:{identifier}"
-        if cache_key in self._tool_cache:
-            return self._tool_cache[cache_key]
-
         tool = None
 
         # Strategy 1: Check registry for already loaded tools
@@ -1205,10 +1171,6 @@ def _load_string_tool(self, identifier: str, module_path: Optional[str] = None)
         if tool is None:
             raise ValueError(f"Tool '{identifier}' not found")
 
-        # Cache the result
-        self._tool_cache[cache_key] = tool
-        logger.debug("tool_identifier=<%s> | loaded and cached", identifier)
-
         return tool
 
     def load_tools(self, identifiers: List[Union[str, Dict[str, Any]]]) -> List[AgentTool]:
@@ -1396,10 +1358,6 @@ def _import_module_from_path(self, module_path: str) -> Any:
         if not path.exists():
             raise ImportError(f"Module file not found: {module_path}")
 
-        # Check cache
-        if module_path in self._loaded_modules:
-            return self._loaded_modules[module_path]
-
         # Import the module
         spec = importlib.util.spec_from_file_location(path.stem, module_path)
         if spec is None or spec.loader is None:
@@ -1416,8 +1374,6 @@ def _import_module_from_path(self, module_path: str) -> Any:
             if path.stem in sys.modules and sys.modules[path.stem] is module:
                 del sys.modules[path.stem]
 
-        # Cache the module
-        self._loaded_modules[module_path] = module
         return module
 
     def _extract_tool_from_module(self, identifier: str, module: Any) -> Optional[AgentTool]:
@@ -1563,18 +1519,3 @@ def _scan_module_for_tool_names(self, module_path: str) -> List[str]:
         except Exception as e:
             logger.warning("module_path=<%s> | failed to scan | %s", module_path, e)
             return []
-
-    def clear_cache(self) -> None:
-        """Clear the internal tool and module caches."""
-        self._tool_cache.clear()
-        self._loaded_modules.clear()
-
-        # Clear caches in dependent loaders if they exist
-        if self._agent_config_loader:
-            self._agent_config_loader.clear_cache()
-        if self._swarm_config_loader:
-            self._swarm_config_loader.clear_cache()
-        if self._graph_config_loader:
-            self._graph_config_loader.clear_cache()
-
-        logger.debug("tool and module caches cleared")