From 437f2e5651a3f452bdd4621bf145dcea960e1ce4 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Fri, 4 Jul 2025 18:38:08 +0000 Subject: [PATCH] feat: enhance CompositeAgent with parallel strategy and add tests --- src/cleveragents/agents/composite.py | 73 ++++++++++--- src/cleveragents/agents/factory.py | 56 +++++++--- .../composite_agent_parallel_strategy.feature | 17 +++ .../steps/composite_agent_parallel_steps.py | 101 ++++++++++++++++++ .../steps/composite_agent_route_steps.py | 1 - 5 files changed, 213 insertions(+), 35 deletions(-) create mode 100644 tests/features/composite_agent_parallel_strategy.feature create mode 100644 tests/features/steps/composite_agent_parallel_steps.py diff --git a/src/cleveragents/agents/composite.py b/src/cleveragents/agents/composite.py index 0c509996c..bc4c81833 100644 --- a/src/cleveragents/agents/composite.py +++ b/src/cleveragents/agents/composite.py @@ -5,6 +5,8 @@ This module provides a composite agent that can combine multiple agents to work together as a single unit. """ +import asyncio +import json import logging from typing import TYPE_CHECKING from typing import Any @@ -101,12 +103,6 @@ class CompositeAgent(Agent): if context is None: context = {} - if not self.strategy == "route" and not self.agents: - raise ConfigurationError( - f"CompositeAgent '{self.name}' with strategy '{self.strategy}' has no " - "child agents configured." - ) - if self.strategy == "sequential": return await self._process_sequential(message, context) elif self.strategy == "parallel": @@ -166,8 +162,17 @@ class CompositeAgent(Agent): str: The processed response. """ current_message = message - for name, agent in self.agents.items(): - current_message = await agent.process(current_message, context) + child_agent_names = self.config.get("agents", []) + for agent_name in child_agent_names: + if agent_name in self.agents: + agent = self.agents[agent_name] + current_message = await agent.process(current_message, context) + else: + logger.warning( + "Agent '%s' not found for CompositeAgent '%s'.", + agent_name, + self.name, + ) return current_message async def _process_parallel(self, message: str, context: Dict[str, Any]) -> str: @@ -181,16 +186,50 @@ class CompositeAgent(Agent): Returns: str: The combined processed response. """ - # In a real implementation, this would use asyncio.gather to run in parallel - responses = {} - for name, agent in self.agents.items(): - responses[name] = await agent.process(message, context) + output_format = self.config.get("output_format", "text") - # Combine responses - combined = "\n\n".join( - [f"Agent '{name}': {response}" for name, response in responses.items()] - ) - return combined + child_agent_names = self.config.get("agents", []) + if not child_agent_names: + logger.warning( + "CompositeAgent '%s' with parallel strategy has no child agents.", + self.name, + ) + return "" + + tasks = [] + valid_agent_names = [] + for agent_name in child_agent_names: + if agent_name in self.agents: + agent = self.agents[agent_name] + tasks.append(agent.process(message, context)) + valid_agent_names.append(agent_name) + else: + logger.warning( + "Agent '%s' not found for CompositeAgent '%s'.", + agent_name, + self.name, + ) + + if not tasks: + return "" + + results = await asyncio.gather(*tasks) + + if output_format == "json": + response_dict = { + name: result for name, result in zip(valid_agent_names, results) + } + return json.dumps(response_dict) + elif output_format == "text": + responses = [ + f"{name}: {result}" for name, result in zip(valid_agent_names, results) + ] + return "\n\n".join(responses) + else: + raise ConfigurationError( + f"Invalid format '{output_format}' for parallel strategy in CompositeAgent '{self.name}'. " + "Must be 'text' or 'json'." + ) def get_capabilities(self) -> List[str]: """ diff --git a/src/cleveragents/agents/factory.py b/src/cleveragents/agents/factory.py index b3eda8a57..ddc3395cc 100644 --- a/src/cleveragents/agents/factory.py +++ b/src/cleveragents/agents/factory.py @@ -6,6 +6,9 @@ configuration. It supports different types of agents and ensures they are properly initialized with the necessary dependencies. """ +from __future__ import annotations + +import inspect from typing import Any from typing import Dict from typing import Optional @@ -13,6 +16,7 @@ from typing import Type from cleveragents.agents.base import Agent from cleveragents.agents.chain import ChainAgent +from cleveragents.agents.composite import CompositeAgent from cleveragents.agents.llm import LLMAgent from cleveragents.agents.tool import ToolAgent from cleveragents.core.exceptions import AgentCreationError @@ -46,16 +50,19 @@ class AgentFactory: "llm": LLMAgent, "tool": ToolAgent, "chain": ChainAgent, + "composite": CompositeAgent, } self.template_renderer = template_renderer self.config = config + self.agents_cache: Dict[str, Agent] = {} def create_agent(self, name: str) -> Agent: """ - Create an agent instance. + Create an agent instance, handling dependencies. This method creates an agent instance based on the configuration - for the agent with the given name. + for the agent with the given name. It uses a cache to avoid re-creating + agents and handles dependency injection for CompositeAgents. Args: name: The name of the agent to create. @@ -66,36 +73,46 @@ class AgentFactory: Raises: AgentCreationError: If the agent cannot be created. """ + if name in self.agents_cache: + return self.agents_cache[name] + try: - # Get agent configuration agent_config = self.config.get("agents", {}).get(name) if not agent_config: raise AgentCreationError(f"Agent '{name}' not found in configuration") - # Get agent type agent_type = agent_config.get("type", "llm").lower() - # Check if agent type is registered if agent_type not in self.agent_types: raise AgentCreationError(f"Unknown agent type: {agent_type}") - # Get agent class agent_class = self.agent_types[agent_type] - if agent_class == Agent: - raise AgentCreationError("Cannot instantiate abstract class Agent") - # ↓ NEW – hand the agent only its specific configuration payload - agent_specific_config: Dict[str, Any] = agent_config.get( - "config", agent_config - ) - # Create agent instance - agent = agent_class(name, agent_specific_config, self.template_renderer) # type: ignore + if inspect.isabstract(agent_class): + raise AgentCreationError( + f"Cannot instantiate abstract class {agent_class.__name__}" + ) + + agent_specific_config: Dict[str, Any] = agent_config.get("config", {}) + + agent = agent_class(name, agent_specific_config, self.template_renderer) # type: ignore[abstract] + + # Cache the agent instance immediately to handle circular dependencies. + self.agents_cache[name] = agent + + # If the agent is a CompositeAgent, create and add its child agents. + if isinstance(agent, CompositeAgent): + child_agent_names = agent_specific_config.get("agents", []) + for child_name in child_agent_names: + child_agent = self.create_agent(child_name) + agent.add_agent(child_name, child_agent) return agent - except Exception as e: # noqa: BLE001 (behave wants the original error types) + except Exception as e: if isinstance(e, (AgentCreationError, ConfigurationError)): - # Propagate AgentCreationError or ConfigurationError unchanged raise - raise AgentCreationError(f"Failed to create agent '{name}': {str(e)}") + raise AgentCreationError( + f"Failed to create agent '{name}': {str(e)}" + ) from e def register_agent_type(self, type_name: str, agent_class: Type[Agent]) -> None: """ @@ -117,6 +134,11 @@ class AgentFactory: if not issubclass(agent_class, Agent): raise AgentCreationError("Agent class must be a subclass of Agent") + if inspect.isabstract(agent_class): + raise AgentCreationError( + f"Cannot register abstract agent class '{agent_class.__name__}'" + ) + self.agent_types[type_name] = agent_class def get_agent_types(self) -> Dict[str, Type[Agent]]: diff --git a/tests/features/composite_agent_parallel_strategy.feature b/tests/features/composite_agent_parallel_strategy.feature new file mode 100644 index 000000000..036376a03 --- /dev/null +++ b/tests/features/composite_agent_parallel_strategy.feature @@ -0,0 +1,17 @@ +Feature: CompositeAgent Parallel Strategy + + As a developer, I want to use a CompositeAgent with a parallel strategy + so that I can process a message with multiple agents concurrently and + combine their outputs in different formats. + + Scenario: CompositeAgent with parallel strategy and text output + Given a CompositeAgent configuration with parallel strategy and format "text" + When I create the CompositeAgent named "ParallelAgent" + And I process the message "Test" with the composite agent + Then the response should be the plain text combination of child agent outputs + + Scenario: CompositeAgent with parallel strategy and JSON output + Given a CompositeAgent configuration with parallel strategy and format "json" + When I create the CompositeAgent named "ParallelAgent" + And I process the message "Test" with the composite agent + Then the response should be a JSON object with child agent outputs diff --git a/tests/features/steps/composite_agent_parallel_steps.py b/tests/features/steps/composite_agent_parallel_steps.py new file mode 100644 index 000000000..fef71c6b5 --- /dev/null +++ b/tests/features/steps/composite_agent_parallel_steps.py @@ -0,0 +1,101 @@ +import asyncio +import json +from typing import Any +from typing import Dict +from typing import List +from typing import Optional + +import yaml +from behave import given +from behave import then +from behave import when + +from cleveragents.agents.base import Agent +from cleveragents.agents.composite import CompositeAgent +from cleveragents.agents.factory import AgentFactory +from cleveragents.core.config import ConfigurationManager +from cleveragents.templates.renderer import TemplateRenderer + + +class MockAgent(Agent): + """A mock agent that returns a configured response.""" + + def __init__( + self, name: str, config: Dict[str, Any], template_renderer: TemplateRenderer + ): + super().__init__(name, config, template_renderer) + self.response = self.config.get("mock_response", f"Response from {name}") + + async def process( + self, message: str, context: Optional[Dict[str, Any]] = None + ) -> str: + return self.response + + def get_capabilities(self) -> List[str]: + return ["mock"] + + +def setup_context_with_config(context, agent_configs: Dict[str, Any]): + """Set up the test context with configuration and agent factory.""" + context.config_manager = ConfigurationManager() + full_config = {"agents": agent_configs} + context.config_manager.config = full_config + context.template_renderer = TemplateRenderer() + context.agent_factory = AgentFactory( + config=context.config_manager.config, + template_renderer=context.template_renderer, + ) + context.agent_factory.register_agent_type("mock", MockAgent) + + +@given( + 'a CompositeAgent configuration with parallel strategy and format "{output_format}"' +) +def step_impl(context, output_format: str): + """Set up a CompositeAgent with parallel strategy and specified output format.""" + config_yaml = f""" + agents: + AgentA: + type: mock + config: + mock_response: "Hello from A" + AgentB: + type: mock + config: + mock_response: "Hello from B" + ParallelAgent: + type: composite + config: + strategy: parallel + output_format: {output_format} + agents: + - AgentA + - AgentB + """ + agent_configs = yaml.safe_load(config_yaml)["agents"] + setup_context_with_config(context, agent_configs) + + +@when('I process the message "{message}" with the composite agent') +def step_impl(context, message: str): + """Process a message using the created composite agent.""" + context.response = asyncio.run(context.agent.process(message)) + + +@then("the response should be the plain text combination of child agent outputs") +def step_impl(context): + """Verify the plain text output from the parallel strategy.""" + expected_response = "AgentA: Hello from A\n\nAgentB: Hello from B" + assert ( + context.response == expected_response + ), f"Expected:\n{expected_response}\nGot:\n{context.response}" + + +@then("the response should be a JSON object with child agent outputs") +def step_impl(context): + """Verify the JSON output from the parallel strategy.""" + response_dict = json.loads(context.response) + expected_dict = {"AgentA": "Hello from A", "AgentB": "Hello from B"} + assert ( + response_dict == expected_dict + ), f"Expected:\n{expected_dict}\nGot:\n{response_dict}" diff --git a/tests/features/steps/composite_agent_route_steps.py b/tests/features/steps/composite_agent_route_steps.py index 4f7e6ca60..0727f1150 100644 --- a/tests/features/steps/composite_agent_route_steps.py +++ b/tests/features/steps/composite_agent_route_steps.py @@ -20,7 +20,6 @@ def setup_context_with_config(context, agent_config): context.agent_factory = AgentFactory( context.agent_config, context.template_renderer ) - context.agent_factory.register_agent_type("composite", CompositeAgent) @given(