`agentic_fabric.core.manager`¶

Hierarchical manager agent for fabric agent orchestration.

This module provides a manager agent that can orchestrate multiple fabric agents, enabling complex workflows with smart delegation, sequential and parallel execution, and human-in-the-loop checkpoints.

Example: ```python from agentic_fabric.core.manager import ManagerAgent

class GameDevManager(ManagerAgent):
    '''Manager that orchestrates game development fabric agents.'''

    def __init__(self):
        super().__init__(
            fabric_agents={
                "design": "gameplay_design",
                "implementation": "ecs_implementation",
                "assets": "asset_pipeline",
                "qa": "qa_validation",
            }
        )

    async def execute_workflow(self, task: str):
        # Sequential execution
        design_result = await self.delegate_async("design", task)

        # Parallel execution
        impl_result, asset_result = await asyncio.gather(
            self.delegate_async("implementation", design_result),
            self.delegate_async("assets", design_result)
        )

        # Final QA
        return await self.delegate_async("qa", {
            "implementation": impl_result,
            "assets": asset_result,
        })
```

Module Contents¶

Classes¶

ManagerAgent

Base class for hierarchical manager agents.

Data¶

logger

API¶

agentic_fabric.core.manager.logger = 'getLogger(...)'¶

class agentic_fabric.core.manager.ManagerAgent(fabric_agents: dict[str, str], package_name: str | None = None, workspace_root: pathlib.Path | None = None, reviewer: collections.abc.Callable[[str, Any], tuple[bool, Any]] | None = None)¶

Bases: abc.ABC

Base class for hierarchical manager agents.

A manager agent orchestrates multiple specialized fabric agents to accomplish complex tasks that require coordination between different domains or phases.

Attributes: fabric_agents: Dict mapping fabric agent role names to fabric agent names in packages. package_name: Optional package name if all fabric agents are in one package. workspace_root: Optional workspace root for fabric agent discovery.

Initialization

Initialize the manager agent.

Args: fabric_agents: Dict mapping role names to fabric agent names (e.g., {“design”: “game_design”}). package_name: Optional package name if all fabric agents are in the same package. workspace_root: Optional workspace root for discovering packages. reviewer: Optional callback for human-in-the-loop checkpoints. Called as reviewer(message, result) and should return (approved, result). If None, checkpoints auto-approve.

delegate(fabric_agent_role: str, inputs: dict[str, Any] | str, framework: str | None = None) → str¶

Delegate a task to a specific fabric agent synchronously.

Args: fabric_agent_role: Role name from the fabric_agents dict (e.g., “design”). inputs: Input dict or string to pass to the fabric agent. framework: Optional framework override.

Returns: Fabric agent output as a string.

Raises: ValueError: If fabric_agent_role not found in fabric_agents mapping.

async delegate_async(fabric_agent_role: str, inputs: dict[str, Any] | str, framework: str | None = None) → str¶

Delegate a task to a specific fabric agent asynchronously.

This runs the fabric agent in a thread pool to avoid blocking the event loop.

Args: fabric_agent_role: Role name from the fabric_agents dict (e.g., “design”). inputs: Input dict or string to pass to the fabric agent. framework: Optional framework override.

Returns: Fabric agent output as a string.

async delegate_parallel(delegations: list[tuple[str, dict[str, Any] | str]], framework: str | None = None) → list[str]¶

Delegate tasks to multiple fabric agents in parallel.

Args: delegations: List of (fabric_agent_role, inputs) tuples. framework: Optional framework override for all fabric agents.

Returns: List of fabric_agent outputs in the same order as delegations.

Example: python results = await manager.delegate_parallel([ ("design", "Create game concept"), ("assets", "Generate placeholder assets"), ]) design_result, assets_result = results

delegate_sequential(delegations: list[tuple[str, dict[str, Any] | str]], framework: str | None = None) → list[str]¶

Delegate tasks to multiple fabric agents sequentially.

Args: delegations: List of (fabric_agent_role, inputs) tuples. framework: Optional framework override for all fabric agents.

Returns: List of fabric_agent outputs in the same order as delegations.

Example: python results = manager.delegate_sequential([ ("design", "Create game concept"), ("implementation", "Implement the design"), ("qa", "Test the implementation"), ])

checkpoint(message: str, result: Any, auto_approve: bool = False) → tuple[bool, Any]¶

Create a human-in-the-loop checkpoint.

Args: message: Message to display to the human reviewer. result: Current result to review. auto_approve: If True, automatically approve without waiting.

Returns: Tuple of (approved, result). If not approved, result may be modified.

Note: When a reviewer callback is set on the manager and auto_approve is False, the reviewer is called with (message, result) and its return value is used. Without a reviewer, checkpoints are logged and auto-approved.

abstractmethod async execute_workflow(task: str, **kwargs: Any) → str¶

Execute the manager’s workflow.

This is the main entry point for the manager agent. Subclasses must override this method to implement their specific orchestration logic.

Args: task: The main task to accomplish. **kwargs: Additional keyword arguments.

Returns: Final result as a string.

agentic_fabric.core.manager¶

Module Contents¶

Classes¶

Data¶

API¶

`agentic_fabric.core.manager`¶