Support list of YAML files for agents_config and tasks_config in CrewBase

src/crewai/project/crew_base.py

@@ -1,6 +1,7 @@
 import inspect
+import logging
 from pathlib import Path
-from typing import Any, Callable, Dict, TypeVar, cast
+from typing import Any, Callable, Dict, List, TypeVar, Union, cast
 
 import yaml
 from dotenv import load_dotenv
@@ -25,11 +26,12 @@ class WrappedClass(cls):  # type: ignore
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
 
-            agents_config_path = self.base_directory / self.original_agents_config_path
-            tasks_config_path = self.base_directory / self.original_tasks_config_path
+            agents_config_paths = self._normalize_to_path_list(self.original_agents_config_path)
+            tasks_config_paths = self._normalize_to_path_list(self.original_tasks_config_path)
 
-            self.agents_config = self.load_yaml(agents_config_path)
-            self.tasks_config = self.load_yaml(tasks_config_path)
+            # Load and merge configurations
+            self.agents_config = self.load_and_merge_yaml_configs(agents_config_paths)
+            self.tasks_config = self.load_and_merge_yaml_configs(tasks_config_paths)
 
             self.map_all_agent_variables()
             self.map_all_task_variables()
@@ -67,14 +69,75 @@ def __init__(self, *args, **kwargs):
                 self._original_functions, "is_kickoff"
             )
 
+        def _normalize_to_path_list(self, paths) -> List[Path]:
+            """
+            Normalize input paths to always be a list of Path objects.
+
+            Args:
+                paths: A string path, Path object, or list of paths
+
+            Returns:
+                A list of Path objects
+            """
+            if isinstance(paths, (list, tuple)):
+                return [self.base_directory / p for p in paths]
+            else:
+                return [self.base_directory / paths]
+
         @staticmethod
         def load_yaml(config_path: Path):
             try:
                 with open(config_path, "r", encoding="utf-8") as file:
                     return yaml.safe_load(file)
             except FileNotFoundError:
-                print(f"File not found: {config_path}")
+                logging.error(f"Configuration YAML file not found: {config_path}")
                 raise
+
+        def deep_merge(self, dict1: dict, dict2: dict) -> dict:
+            """
+            Recursively merge two dictionaries, with values from dict2 taking precedence.
+
+            Args:
+                dict1: First dictionary
+                dict2: Second dictionary with values that will override dict1 for duplicate keys
+
+            Returns:
+                A new dictionary with merged values
+            """
+            result = dict1.copy()
+            for key, value in dict2.items():
+                if key in result and isinstance(result[key], dict) and isinstance(value, dict):
+                    result[key] = self.deep_merge(result[key], value)
+                else:
+                    result[key] = value
+            return result
+
+        def load_and_merge_yaml_configs(self, config_paths: List[Path]) -> dict:
+            """
+            Load and merge configurations from multiple YAML files.
+
+            This function loads each YAML file in the provided list and merges their
+            configurations. For duplicate keys, later files in the list will override
+            earlier ones. For nested dictionaries, a deep merge is performed, meaning
+            that nested keys are preserved unless explicitly overridden.
+
+            Example:
+                If file1.yaml contains: {"agent1": {"role": "researcher", "goal": "find info"}}
+                And file2.yaml contains: {"agent1": {"role": "analyst"}}
+                The result will be: {"agent1": {"role": "analyst", "goal": "find info"}}
+
+            Args:
+                config_paths: A list of Path objects pointing to YAML files
+
+            Returns:
+                A dictionary with merged configurations
+            """
+            result = {}
+            for path in config_paths:
+                config = self.load_yaml(path)
+                if config:
+                    result = self.deep_merge(result, config)
+            return result
 
         def _get_all_functions(self):
             return {

tests/config/multi/agents1.yaml

@@ -0,0 +1,5 @@
+test_agent1:
+  role: Test Agent 1
+  goal: Test Goal 1
+  backstory: Test Backstory 1
+  verbose: true

tests/config/multi/agents2.yaml

@@ -0,0 +1,8 @@
+test_agent1:
+  role: Updated Test Agent 1
+  goal: Updated Test Goal 1
+test_agent2:
+  role: Test Agent 2
+  goal: Test Goal 2
+  backstory: Test Backstory 2
+  verbose: true

tests/config/multi/tasks1.yaml

@@ -0,0 +1,4 @@
+test_task1:
+  description: Test Description 1
+  expected_output: Test Output 1
+  agent: test_agent1

tests/config/multi/tasks2.yaml

@@ -0,0 +1,6 @@
+test_task1:
+  description: Updated Test Description 1
+test_task2:
+  description: Test Description 2
+  expected_output: Test Output 2
+  agent: test_agent2

tests/project_test.py

@@ -184,3 +184,68 @@ def crew(self):
     assert "plants" in result.raw, "First before_kickoff not executed"
     assert "processed first" in result.raw, "First after_kickoff not executed"
     assert "processed second" in result.raw, "Second after_kickoff not executed"
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_multiple_yaml_configs():
+    @CrewBase
+    class MultiConfigCrew:
+        agents_config = ["config/multi/agents1.yaml", "config/multi/agents2.yaml"]
+        tasks_config = ["config/multi/tasks1.yaml", "config/multi/tasks2.yaml"]
+
+        @agent
+        def test_agent1(self):
+            return Agent(config=self.agents_config["test_agent1"])
+
+        @agent
+        def test_agent2(self):
+            return Agent(config=self.agents_config["test_agent2"])
+
+        @task
+        def test_task1(self):
+            task_config = self.tasks_config["test_task1"].copy()
+            if isinstance(task_config.get("agent"), str):
+                agent_name = task_config.pop("agent")
+                if hasattr(self, agent_name):
+                    task_config["agent"] = getattr(self, agent_name)()
+            return Task(config=task_config)
+
+        @task
+        def test_task2(self):
+            task_config = self.tasks_config["test_task2"].copy()
+            if isinstance(task_config.get("agent"), str):
+                agent_name = task_config.pop("agent")
+                if hasattr(self, agent_name):
+                    task_config["agent"] = getattr(self, agent_name)()
+            return Task(config=task_config)
+
+        @crew
+        def crew(self):
+            return Crew(agents=self.agents, tasks=self.tasks, verbose=True)
+
+    crew = MultiConfigCrew()
+
+    assert "test_agent1" in crew.agents_config
+    assert "test_agent2" in crew.agents_config
+
+    assert crew.agents_config["test_agent1"]["role"] == "Updated Test Agent 1"
+    assert crew.agents_config["test_agent1"]["goal"] == "Updated Test Goal 1"
+    assert crew.agents_config["test_agent1"]["backstory"] == "Test Backstory 1"
+    assert crew.agents_config["test_agent1"]["verbose"] is True
+
+    assert "test_task1" in crew.tasks_config
+    assert "test_task2" in crew.tasks_config
+
+    assert crew.tasks_config["test_task1"]["description"] == "Updated Test Description 1"
+    assert crew.tasks_config["test_task1"]["expected_output"] == "Test Output 1"
+    assert crew.tasks_config["test_task1"]["agent"].role == "Updated Test Agent 1"
+
+    agent1 = crew.test_agent1()
+    agent2 = crew.test_agent2()
+    task1 = crew.test_task1()
+    task2 = crew.test_task2()
+
+    assert agent1.role == "Updated Test Agent 1"
+    assert agent2.role == "Test Agent 2"
+    assert task1.description == "Updated Test Description 1"
+    assert task2.description == "Test Description 2"