Skip to content

Factories

The factories module implements the Factory pattern for creating and managing various components of the Agency system. This includes factories for agents, tools, guardrails, and agency instances, ensuring consistent initialization and configuration.

Overview

The Agency framework provides four main factory types plus a powerful Loader pattern:

  1. AgentFactory - Creates and manages agent instances with different specializations
  2. ToolFactory - Registers and builds tools organized by category
  3. GuardrailFactory - Creates and configures content validation guardrails
  4. AgencyFactory - Creates properly configured Agency instances with memory and observability
  5. Loader - Automatically discovers and loads constructors from Python modules (new pattern)

AgentFactory

The AgentFactory provides a centralized system for creating and managing agents using a category-based builder pattern.

Quick Start

from agency.factories.agents import get_agent_factory

# Get the global factory instance (singleton)
factory = get_agent_factory()

# Build a research agent
researcher = factory.build(category="research", name="analyst")

# Build a coding agent
coder = factory.build(category="coding", name="python_expert")

# List available agents
agents = factory.list_agents()
print(factory.describe())

Agent Categories

Agents are organized into specialized categories:

Category Purpose Examples
research Research and analysis tasks Analyst, Data Researcher
coding Software development Python Expert, Code Reviewer
content Content creation Writer, Editor
support Customer support Support Agent, FAQ Bot
data Data processing Data Engineer, Analyst

Registering Custom Agents

Add your agent to the appropriate category registration function:

# In agency/factories/agents/agent_registry.py

def register_research_agents(factory: AgentFactory) -> None:
    """Register research agents."""
    from agency.agents.research.analyst import AnalystAgent

    factory.register_builder(
        category="research",
        name="analyst",
        builder=lambda: AnalystAgent(),
    )

    # Add your custom agent
    from agency.agents.research.my_researcher import MyResearcher

    factory.register_builder(
        category="research",
        name="my_researcher",
        builder=lambda: MyResearcher(config={"depth": "deep"}),
    )

Advanced Patterns

Wildcard Category

Use None as the category for agents that work across categories:

factory.register_builder(
    category=None,  # Wildcard - matches any category
    name="universal_assistant",
    builder=lambda: UniversalAgent(),
)

# Can be accessed with any category
agent1 = factory.build(category="research", name="universal_assistant")
agent2 = factory.build(category="coding", name="universal_assistant")

Builder with Configuration

Pass configuration to your agent builder:

def create_analyst(specialization: str = "general"):
    """Factory function that accepts configuration."""
    return AnalystAgent(specialization=specialization)

factory.register_builder(
    category="research",
    name="financial_analyst",
    builder=lambda: create_analyst(specialization="finance"),
)

ToolFactory

The ToolFactory provides a centralized system for registering and managing tools using a category-based organization pattern.

Quick Start

from agency.factories.tools import get_tool_factory

# Get the global factory instance (singleton)
factory = get_tool_factory()

# Build a calculator tool
calculator = factory.build(category="computation", name="calculator")
result = await calculator.execute(expression="2 + 2")

# Build a search tool
search = factory.build(category="search", name="duckduckgo")
results = await search.execute(query="Python async patterns", max_results=5)

# Build a drawing tool
drawing = factory.build(category="visualization", name="drawing")
await drawing.execute(task="plot_surface")

# List available tools
tools = factory.list_tools()
print(factory.describe())

Tool Categories

Tools are organized into functional categories:

Category Purpose Examples
search Web search and information retrieval DuckDuckGo, Google, Brave
computation Mathematical and computational operations Calculator, Analyzer
visualization Data visualization and plotting Drawing, Charting
data Data processing and transformation Parser, Transformer, Validator
integration External service integrations API clients, Database connectors
communication Communication and notifications Email, Slack, SMS

Currently Registered Tools

  • SearchTool (search/duckduckgo) - Web search using DuckDuckGo
  • CalculatorTool (computation/calculator) - Mathematical expression evaluation
  • DrawingTool (visualization/drawing) - Data visualization and plotting

Registering Custom Tools

Step 1: Ensure Tool Implements ToolProtocol

from agency.core.types import ToolProtocol
from typing import Any

class MyCustomTool(ToolProtocol):
    @property
    def name(self) -> str:
        return "my_tool"

    @property
    def description(self) -> str:
        return "Description of what my tool does"

    async def execute(self, **kwargs) -> Any:
        # Your tool's implementation
        return result

Step 2: Register in tool_registry.py

# In agency/factories/tools/tool_registry.py

def register_computation_tools(factory: ToolFactory) -> None:
    """Register computation tools in the factory."""
    from agency.tools.computation.calculator import CalculatorTool

    factory.register_builder(
        category="computation",
        name="calculator",
        builder=lambda: CalculatorTool(),
    )

    # Add your new tool here
    from agency.tools.computation.my_analyzer import MyAnalyzer

    factory.register_builder(
        category="computation",
        name="analyzer",
        builder=lambda: MyAnalyzer(),
    )

Tool Factory Phase 2 (Coming Soon)

Future decorators for enhanced tool functionality: - with_cache - Caching decorator for tool results - with_timeout - Timeout handling for long-running tools - with_retries - Automatic retry logic - with_rate_limit - Rate limiting for API-based tools - with_logging - Comprehensive logging - with_metrics - Performance metrics collection

GuardrailFactory

The GuardrailFactory provides a centralized system for creating and managing content validation guardrails using a category-based organization pattern.

Guardrails validate agent inputs and outputs to ensure safety, compliance, quality, and appropriateness. They follow the same factory pattern as agents and tools for consistency.

Quick Start

from agency.factories.guardrails import get_guardrail_factory

# Get the global factory instance (singleton)
factory = get_guardrail_factory()

# Build a content safety guardrail
safety = factory.build(category="safety", name="content_safety")
result = await safety.check("Some text to validate")

# Build a PII detection guardrail
pii = factory.build(category="safety", name="pii_detection")
result = await pii.check("Contact me at john@example.com")

# Build a tone validation guardrail
tone = factory.build(category="quality", name="tone_sentiment", expected_tone="professional")
result = await tone.check("Thank you for your inquiry.")

# List available guardrails
guardrails = factory.list_guardrails()
print(factory.describe())

Guardrail Categories

Guardrails are organized into functional categories:

Category Purpose Examples
safety Safety and security validation Content Safety, PII Detection
quality Quality and format validation Tone/Sentiment, Format Validation, JSON Validation
compliance Regulatory and policy compliance (Future: GDPR, HIPAA, Industry-specific)

Currently Registered Guardrails

Safety Guardrails: - ContentSafetyGuardrail (safety/content_safety) - Detects harmful, offensive, or inappropriate content - PIIDetectionGuardrail (safety/pii_detection) - Detects personally identifiable information (emails, phones, SSNs, etc.)

Quality Guardrails: - ToneSentimentGuardrail (quality/tone_sentiment) - Validates tone appropriateness (professional, friendly, neutral, empathetic) - FormatValidationGuardrail (quality/format_validation) - Validates length, structure, and required patterns - JSONFormatGuardrail (quality/json_format) - Validates JSON structure

Using Guardrails

Basic Validation

from agency.factories.guardrails import get_guardrail_factory

async def validate_content(text: str) -> bool:
    """Validate content with multiple guardrails."""
    factory = get_guardrail_factory()

    # Check content safety
    safety = factory.build("safety", "content_safety")
    result = await safety.check(text)
    if not result.passed:
        print(f"Safety violation: {result.message}")
        return False

    # Check for PII
    pii = factory.build("safety", "pii_detection")
    result = await pii.check(text)
    if not result.passed:
        print(f"PII detected: {result.violations}")
        return False

    return True

# Usage
is_valid = await validate_content("Some user-generated content")

With Configuration

# Build with custom threshold
guardrail = factory.build(
    category="safety",
    name="content_safety",
    with_threshold=0.95  # Very strict (default is 0.8)
)

# Build with non-blocking mode (warnings only)
guardrail = factory.build(
    category="safety",
    name="pii_detection",
    with_blocking=False
)

# Build with custom rules
guardrail = factory.build(
    category="quality",
    name="tone_sentiment",
    expected_tone="friendly",
    with_threshold=0.6
)

From Configuration Dictionary

# Define guardrail configurations
configs = [
    {
        "category": "safety",
        "name": "content_safety",
        "with_threshold": 0.9
    },
    {
        "category": "safety",
        "name": "pii_detection",
        "with_blocking": True
    },
    {
        "category": "quality",
        "name": "format_validation",
        "min_length": 50,
        "max_length": 1000,
        "required_patterns": [r"\[Source \d+\]"]
    }
]

# Build guardrails from configs
factory = get_guardrail_factory()
guardrails = [factory.build_from_config(config) for config in configs]

# Validate with all guardrails
async def validate_all(text: str):
    for guardrail in guardrails:
        result = await guardrail.check(text)
        if not result.passed:
            print(f"Failed: {guardrail.name} - {result.message}")
            return False
    return True

Registering Custom Guardrails

Step 1: Create Your Guardrail Class

from agency.guardrails.base import GuardrailBase, GuardrailResult, GuardrailSeverity
from typing import Any

class CustomComplianceGuardrail(GuardrailBase):
    """Custom guardrail for compliance validation."""

    def __init__(self, **kwargs):
        super().__init__(
            name="Custom Compliance",
            description="Validates content against custom compliance rules",
            **kwargs
        )

    async def check(self, text: str, **kwargs: Any) -> GuardrailResult:
        """Check content against compliance rules."""
        # Your validation logic here
        violations = []

        # Example: Check for required disclaimers
        if "Terms and Conditions" not in text:
            violations.append("Missing required terms disclaimer")

        if violations:
            return GuardrailResult(
                passed=False,
                severity=GuardrailSeverity.ERROR,
                message="Compliance validation failed",
                violations=violations,
                confidence=0.95
            )

        return GuardrailResult(
            passed=True,
            message="Compliance validation passed",
            confidence=0.95
        )

# Create builder function
def create_compliance_guardrail(**kwargs) -> CustomComplianceGuardrail:
    return CustomComplianceGuardrail(**kwargs)

Step 2: Register in guardrail_registry.py

# In agency/factories/guardrails/guardrail_registry.py

def register_compliance_guardrails(factory: GuardrailFactory) -> None:
    """Register compliance-related guardrails with the factory."""
    from agency.guardrails.custom.compliance import create_compliance_guardrail

    factory.register_builder(
        category="compliance",
        name="custom_compliance",
        builder=create_compliance_guardrail
    )

    LOGGER.info("Registered 1 compliance guardrail")

# Add to register_all_guardrails function
def register_all_guardrails(factory: GuardrailFactory) -> None:
    """Register all guardrails with the factory."""
    LOGGER.info("Starting full guardrail registration...")

    register_safety_guardrails(factory)
    register_quality_guardrails(factory)
    register_compliance_guardrails(factory)  # Add this line

    LOGGER.info("Full guardrail registration complete")

Advanced Patterns

Composite Guardrail

Combine multiple guardrails into one:

from agency.guardrails.base import CompositeGuardrail

async def create_comprehensive_validator():
    """Create a composite guardrail that runs multiple checks."""
    factory = get_guardrail_factory()

    # Build individual guardrails
    safety = factory.build("safety", "content_safety")
    pii = factory.build("safety", "pii_detection")
    tone = factory.build("quality", "tone_sentiment", expected_tone="professional")

    # Combine into composite
    composite = CompositeGuardrail(
        name="Comprehensive Validator",
        description="Validates safety, PII, and tone",
        guardrails=[safety, pii, tone],
        mode="all"  # All must pass
    )

    # Use composite
    result = await composite.check("Some text to validate")
    if not result.passed:
        print(f"Validation failed: {result.message}")
        print(f"Violations: {result.violations}")

# Usage
await create_comprehensive_validator()

Agent Output Validation

Integrate guardrails with agent workflows:

from agency.factories.agents import get_agent_factory
from agency.factories.guardrails import get_guardrail_factory

async def validated_agent_response(agent, user_input: str) -> str:
    """Get agent response with guardrail validation."""

    # Get response from agent
    response = await agent.run(user_input)

    # Validate response
    factory = get_guardrail_factory()

    # Check for PII in output
    pii_guard = factory.build("safety", "pii_detection")
    pii_result = await pii_guard.check(response.content)

    if not pii_result.passed:
        # Sanitize or reject response
        print(f"PII detected in response: {pii_result.violations}")
        return "Response contained sensitive information and was blocked."

    # Check tone
    tone_guard = factory.build("quality", "tone_sentiment", expected_tone="professional")
    tone_result = await tone_guard.check(response.content)

    if not tone_result.passed:
        print(f"Warning: Tone issue - {tone_result.message}")

    return response.content

Conditional Guardrails

Apply guardrails based on context:

async def context_aware_validation(text: str, context: dict) -> bool:
    """Apply guardrails based on context."""
    factory = get_guardrail_factory()

    # Always check safety
    safety = factory.build("safety", "content_safety")
    if not (await safety.check(text)).passed:
        return False

    # Check PII only for public-facing content
    if context.get("audience") == "public":
        pii = factory.build("safety", "pii_detection")
        if not (await pii.check(text)).passed:
            return False

    # Check tone for customer communications
    if context.get("type") == "customer_communication":
        tone = factory.build("quality", "tone_sentiment", expected_tone="professional")
        if not (await tone.check(text)).passed:
            return False

    return True

# Usage
is_valid = await context_aware_validation(
    text="Some content",
    context={"audience": "public", "type": "customer_communication"}
)

Guardrail Results

All guardrails return a GuardrailResult object:

from agency.guardrails.base import GuardrailResult, GuardrailSeverity

result = await guardrail.check("Some text")

# Check if passed
if result.passed:
    print("Content is valid")
else:
    print(f"Validation failed: {result.message}")

# Access details
print(f"Severity: {result.severity}")  # INFO, WARNING, ERROR, CRITICAL
print(f"Confidence: {result.confidence}")  # 0.0 - 1.0
print(f"Violations: {result.violations}")  # List of specific violations
print(f"Metadata: {result.metadata}")  # Additional context

Performance Considerations

  • Async Design: All guardrails use async check() for non-blocking validation
  • Confidence Thresholds: Tune thresholds to balance false positives/negatives
  • Caching: Consider caching guardrail results for identical content
  • Parallel Execution: Run independent guardrails in parallel for better performance
import asyncio

async def parallel_validation(text: str) -> list[GuardrailResult]:
    """Run multiple guardrails in parallel."""
    factory = get_guardrail_factory()

    # Create guardrails
    guardrails = [
        factory.build("safety", "content_safety"),
        factory.build("safety", "pii_detection"),
        factory.build("quality", "tone_sentiment", expected_tone="professional")
    ]

    # Run all guardrails in parallel
    results = await asyncio.gather(
        *[g.check(text) for g in guardrails]
    )

    return results

# Usage
results = await parallel_validation("Some text to validate")
all_passed = all(r.passed for r in results)

Testing with GuardrailFactory

import pytest
from agency.factories.guardrails import GuardrailFactory, get_guardrail_factory

@pytest.mark.asyncio
async def test_guardrail_registration():
    """Test guardrail registration."""
    factory = GuardrailFactory()

    # Verify registration
    guardrails = factory.list_guardrails(category="safety")
    assert ("safety", "content_safety") in guardrails
    assert ("safety", "pii_detection") in guardrails

@pytest.mark.asyncio
async def test_guardrail_validation():
    """Test guardrail validation."""
    factory = get_guardrail_factory()

    # Test content safety
    safety = factory.build("safety", "content_safety")
    result = await safety.check("Hello, world!")
    assert result.passed

    # Test PII detection
    pii = factory.build("safety", "pii_detection")
    result = await pii.check("Contact: john@example.com")
    assert not result.passed
    assert "email" in str(result.violations).lower()

@pytest.mark.asyncio
async def test_guardrail_configuration():
    """Test guardrail configuration."""
    factory = get_guardrail_factory()

    # Test with custom threshold
    guardrail = factory.build(
        "safety",
        "content_safety",
        with_threshold=0.95
    )
    assert guardrail.config.threshold == 0.95

AgencyFactory

The main factory class for creating Agency instances with different configurations:

from agency.factories.agency_factory import AgencyFactory

Factory Methods

Basic Agency Creation

from agency.agency.config import AgencyConfig

# Create basic agency
config = AgencyConfig(
    name="CustomerSupport",
    strategy_type="router"
)

agency = AgencyFactory.create_agency(config)

Agency with Memory

from agency.sessions.session_factory import SessionFactory

# Create session for memory
session = SessionFactory.create_file_session(
    session_id="user_123",
    db_path="data/sessions/user_123.db"
)

# Create agency with session
agency = AgencyFactory.create_agency(
    config=config,
    session=session,
    session_id="user_123",
    user_id="user_456"
)

Agency with Custom Strategy

from agency.strategies.router import RouterStrategy

# Create custom strategy
strategy = RouterStrategy(
    overrides={
        "billing_agent": custom_billing_agent,
        "technical_agent": custom_technical_agent
    }
)

# Create agency with strategy
agency = AgencyFactory.create_agency_with_strategy(
    name="CustomSupport",
    strategy=strategy,
    session=session
)

Configuration Patterns

Environment-Based Configuration

import os

def create_production_agency(user_id: str):
    """Create production agency with proper configuration."""

    # Create session
    session = SessionFactory.create_file_session(
        session_id=f"prod_{user_id}",
        db_path=f"data/production/sessions/{user_id}.db"
    )

    # Production config
    config = AgencyConfig(
        name="ProductionSupport",
        strategy_type="router",
        settings={
            "max_tokens": 4000,
            "temperature": 0.1,  # More deterministic for production
            "timeout": 60
        }
    )

    return AgencyFactory.create_agency(
        config=config,
        session=session,
        session_id=f"prod_{user_id}",
        user_id=user_id,
        auto_flush=True  # Enable Langfuse auto-flush
    )

Development Configuration

def create_development_agency():
    """Create development agency with debug settings."""

    # In-memory session for development
    session = SessionFactory.create_memory_session("dev_session")

    config = AgencyConfig(
        name="DevelopmentSupport",
        strategy_type="router",
        settings={
            "max_tokens": 2000,
            "temperature": 0.7,
            "debug": True
        }
    )

    return AgencyFactory.create_agency(
        config=config,
        session=session,
        langfuse_mode=LangfuseMode.LOCAL  # Use local Langfuse
    )

Testing Configuration

def create_test_agency():
    """Create agency configured for testing."""

    session = SessionFactory.create_memory_session("test_session")

    config = AgencyConfig(
        name="TestSupport",
        strategy_type="simple",  # Use simple strategy for tests
        settings={
            "max_tokens": 1000,
            "temperature": 0.0,  # Deterministic for testing
        }
    )

    return AgencyFactory.create_agency(
        config=config,
        session=session,
        langfuse_mode=LangfuseMode.DISABLED  # No observability in tests
    )

Advanced Factory Patterns

Builder Pattern Integration

class AgencyBuilder:
    """Builder pattern for complex agency configuration."""

    def __init__(self):
        self.config = AgencyConfig(name="DefaultAgency")
        self.session = None
        self.session_id = None
        self.user_id = None
        self.langfuse_mode = LangfuseMode.AUTO

    def with_name(self, name: str):
        self.config.name = name
        return self

    def with_strategy(self, strategy_type: str):
        self.config.strategy_type = strategy_type
        return self

    def with_memory(self, session_id: str, db_path: str):
        self.session = SessionFactory.create_file_session(session_id, db_path)
        self.session_id = session_id
        return self

    def with_user(self, user_id: str):
        self.user_id = user_id
        return self

    def with_observability(self, mode: LangfuseMode):
        self.langfuse_mode = mode
        return self

    def build(self):
        return AgencyFactory.create_agency(
            config=self.config,
            session=self.session,
            session_id=self.session_id,
            user_id=self.user_id,
            langfuse_mode=self.langfuse_mode
        )

# Usage
agency = (AgencyBuilder()
    .with_name("CustomerSupport")
    .with_strategy("router")
    .with_memory("user_123", "data/sessions/user_123.db")
    .with_user("user_123")
    .with_observability(LangfuseMode.LOCAL)
    .build())

Factory with Dependency Injection

class AgencyFactoryWithDI:
    """Factory with dependency injection support."""

    def __init__(self,
                 session_factory=SessionFactory,
                 default_config=None):
        self.session_factory = session_factory
        self.default_config = default_config or AgencyConfig()

    def create_agency(self,
                     name: str,
                     user_id: str,
                     strategy_type: str = "router",
                     use_memory: bool = True):

        # Create session if needed
        session = None
        if use_memory:
            session = self.session_factory.create_file_session(
                session_id=f"{name}_{user_id}",
                db_path=f"data/sessions/{user_id}.db"
            )

        # Create config
        config = AgencyConfig(
            name=name,
            strategy_type=strategy_type
        )

        return AgencyFactory.create_agency(
            config=config,
            session=session,
            user_id=user_id
        )

Configuration Validation

def validate_agency_config(config: AgencyConfig) -> bool:
    """Validate agency configuration before creation."""

    required_fields = ['name', 'strategy_type']
    for field in required_fields:
        if not getattr(config, field):
            raise ValueError(f"Missing required field: {field}")

    # Validate strategy type
    valid_strategies = ['router', 'simple', 'custom']
    if config.strategy_type not in valid_strategies:
        raise ValueError(f"Invalid strategy type: {config.strategy_type}")

    return True

# Usage with validation
def create_validated_agency(config: AgencyConfig, **kwargs):
    validate_agency_config(config)
    return AgencyFactory.create_agency(config, **kwargs)

Error Handling Patterns

def create_resilient_agency(config: AgencyConfig, **kwargs):
    """Create agency with comprehensive error handling."""

    try:
        return AgencyFactory.create_agency(config, **kwargs)

    except ValueError as e:
        print(f"Configuration error: {e}")
        # Return agency with default config
        return AgencyFactory.create_agency(AgencyConfig())

    except Exception as e:
        print(f"Agency creation failed: {e}")
        # Return minimal agency
        return AgencyFactory.create_agency(
            config=AgencyConfig(name="FallbackAgency"),
            session=SessionFactory.create_memory_session("fallback"),
            langfuse_mode=LangfuseMode.DISABLED
        )

Testing Factory Methods

import pytest
from agency.factories.agency_factory import AgencyFactory

def test_basic_agency_creation():
    """Test basic agency creation."""
    config = AgencyConfig(name="TestAgency")
    agency = AgencyFactory.create_agency(config)

    assert agency.name == "TestAgency"
    assert agency.processor is not None

def test_agency_with_memory():
    """Test agency creation with memory session."""
    session = SessionFactory.create_memory_session("test")
    config = AgencyConfig(name="MemoryAgency")

    agency = AgencyFactory.create_agency(
        config=config,
        session=session
    )

    assert agency.session == session

@pytest.fixture
def mock_session():
    """Mock session for testing."""
    return SessionFactory.create_memory_session("mock")

def test_agency_with_fixture(mock_session):
    """Test using pytest fixture."""
    config = AgencyConfig(name="FixtureAgency")
    agency = AgencyFactory.create_agency(config, session=mock_session)

    assert agency.session == mock_session

Integration Examples

With Streamlit

import streamlit as st
from agency.factories.agency_factory import AgencyFactory

@st.cache_resource
def get_agency(user_id: str):
    """Cached agency creation for Streamlit."""

    config = AgencyConfig(
        name="StreamlitSupport",
        strategy_type="router"
    )

    session = SessionFactory.create_file_session(
        session_id=f"streamlit_{user_id}",
        db_path=f"data/streamlit/{user_id}.db"
    )

    return AgencyFactory.create_agency(
        config=config,
        session=session,
        user_id=user_id
    )

With FastAPI

from fastapi import Depends, HTTPException
from agency.factories.agency_factory import AgencyFactory

def get_user_agency(user_id: str = Depends(get_current_user)):
    """FastAPI dependency for user agency."""

    try:
        config = get_user_config(user_id)  # Load user-specific config
        session = SessionFactory.create_file_session(
            session_id=f"api_{user_id}",
            db_path=f"data/api/{user_id}.db"
        )

        return AgencyFactory.create_agency(
            config=config,
            session=session,
            user_id=user_id
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agency creation failed: {e}")

Using Factories Together

The four factory types work together to build complete Agency solutions:

Example: Building an Agency with Agents, Tools, and Guardrails

from agency.factories.agents import get_agent_factory
from agency.factories.tools import get_tool_factory
from agency.factories.guardrails import get_guardrail_factory
from agency.factories.agency_factory import AgencyFactory
from agency.agency.config import AgencyConfig
from agency.sessions.session_factory import SessionFactory

async def create_validated_research_agency(user_id: str):
    """Create a research agency with agents, tools, and validation guardrails."""

    # Get factory instances
    agent_factory = get_agent_factory()
    tool_factory = get_tool_factory()
    guardrail_factory = get_guardrail_factory()

    # Build specialized agents
    analyst = agent_factory.build(category="research", name="analyst")
    researcher = agent_factory.build(category="research", name="data_researcher")

    # Build tools for the agents
    search_tool = tool_factory.build(category="search", name="duckduckgo")
    calculator = tool_factory.build(category="computation", name="calculator")
    drawing = tool_factory.build(category="visualization", name="drawing")

    # Build guardrails for output validation
    safety_guard = guardrail_factory.build("safety", "content_safety")
    pii_guard = guardrail_factory.build("safety", "pii_detection")
    tone_guard = guardrail_factory.build("quality", "tone_sentiment", expected_tone="professional")

    # Equip agents with tools
    analyst.tools = [search_tool, calculator]
    researcher.tools = [search_tool, drawing]

    # Create session for memory
    session = SessionFactory.create_file_session(
        session_id=f"research_{user_id}",
        db_path=f"data/research/{user_id}.db"
    )

    # Create agency with configured agents
    config = AgencyConfig(
        name="ResearchAgency",
        strategy_type="router",
        agents=[analyst, researcher]
    )

    agency = AgencyFactory.create_agency(
        config=config,
        session=session,
        user_id=user_id
    )

    # Wrap agency with validation
    async def validated_process(query: str):
        """Process with output validation."""
        result = await agency.process(query)

        # Validate output with guardrails
        for guard in [safety_guard, pii_guard, tone_guard]:
            check_result = await guard.check(result)
            if not check_result.passed:
                print(f"Validation failed: {guard.name} - {check_result.message}")
                return f"Output failed validation: {check_result.message}"

        return result

    agency.validated_process = validated_process
    return agency

# Usage
agency = await create_validated_research_agency("user_123")
result = await agency.validated_process("Research quantum computing trends")

Example: Dynamic Tool Registration and Agent Creation

from agency.factories.agents import AgentFactory, get_agent_factory
from agency.factories.tools import ToolFactory, get_tool_factory

async def setup_custom_environment():
    """Setup a custom environment with dynamically registered agents and tools."""

    # Get or create factories
    agent_factory = get_agent_factory(reload=True)
    tool_factory = get_tool_factory(reload=True)

    # Register a custom tool
    from agency.tools.custom.my_tool import MyCustomTool

    tool_factory.register_builder(
        category="custom",
        name="my_tool",
        builder=lambda: MyCustomTool(),
    )

    # Register a custom agent
    from agency.agents.custom.my_agent import MyCustomAgent

    agent_factory.register_builder(
        category="custom",
        name="my_agent",
        builder=lambda: MyCustomAgent(),
    )

    # Build and use
    my_tool = tool_factory.build(category="custom", name="my_tool")
    my_agent = agent_factory.build(category="custom", name="my_agent")
    my_agent.tools = [my_tool]

    return my_agent

# Usage
agent = await setup_custom_environment()

Example: Factory Inspection for Debugging

from agency.factories.agents import get_agent_factory
from agency.factories.tools import get_tool_factory

def inspect_available_components():
    """Inspect all available agents and tools."""

    agent_factory = get_agent_factory()
    tool_factory = get_tool_factory()

    print("=== Available Agents ===")
    print(agent_factory.describe())

    print("\n=== Available Tools ===")
    print(tool_factory.describe())

    print("\n=== Agents by Category ===")
    for category in ["research", "coding", "content"]:
        agents = agent_factory.list_agents(category=category)
        print(f"{category}: {[name for _, name in agents]}")

    print("\n=== Tools by Category ===")
    for category in ["search", "computation", "visualization"]:
        tools = tool_factory.list_tools(category=category)
        print(f"{category}: {[name for _, name in tools]}")

# Usage
inspect_available_components()

Best Practices

AgentFactory and ToolFactory

  1. Use Singletons: Use get_agent_factory() and get_tool_factory() for global access
  2. Category Organization: Organize agents and tools by category for better discoverability
  3. Builder Pattern: Use lambda builders for lazy instantiation
  4. Wildcard Support: Use category=None for cross-category components
  5. Registry Functions: Group registrations in category-specific functions

GuardrailFactory

  1. Use Singletons: Use get_guardrail_factory() for global access
  2. Category Organization: Organize guardrails by purpose (safety, quality, compliance)
  3. Threshold Tuning: Adjust thresholds based on your risk tolerance
  4. Async Validation: All guardrails are async - use await for checks
  5. Parallel Execution: Run independent guardrails in parallel for performance
  6. Context-Aware: Apply different guardrails based on content context
  7. Composite Pattern: Combine multiple guardrails for comprehensive validation

AgencyFactory

  1. Configuration Validation: Always validate configurations before agency creation
  2. Error Handling: Implement comprehensive error handling with fallbacks
  3. Resource Management: Properly clean up sessions and resources
  4. Testing: Use factories in tests for consistent object creation
  5. Dependency Injection: Use DI for better testability and flexibility
  6. Caching: Cache agencies when appropriate (e.g., Streamlit)
  7. Environment Separation: Use different configurations for dev/staging/prod

Combined Usage

  1. Separation of Concerns: Use AgentFactory for agents, ToolFactory for tools, GuardrailFactory for validation, AgencyFactory for agencies
  2. Lazy Loading: Load heavy resources only when needed
  3. Reload for Testing: Use reload=True when testing to get fresh instances
  4. Documentation: Use describe() methods for debugging and documentation
  5. Validation Pipeline: Integrate guardrails into agent workflows for output validation

Performance Considerations

AgentFactory and ToolFactory

  • Singleton Pattern: Reduces instantiation overhead through global caching
  • Lazy Building: Builders are only executed when build() is called
  • Instance Reuse: Consider caching built instances when appropriate
  • Registration Overhead: Minimal - registration happens once at module import

GuardrailFactory

  • Async Design: Non-blocking validation with async/await
  • Parallel Checks: Run independent guardrails concurrently
  • Threshold Tuning: Balance accuracy vs. performance with confidence thresholds
  • Pattern Compilation: Regex patterns are pre-compiled for efficiency
  • Result Caching: Consider caching results for identical content

AgencyFactory

  • Session Reuse: Reuse sessions when possible to avoid creation overhead
  • Lazy Loading: Load heavy resources only when needed
  • Connection Pooling: Consider connection pooling for database sessions
  • Memory Management: Clean up unused agencies and sessions

Testing with Factories

Testing AgentFactory

import pytest
from agency.factories.agents import AgentFactory, get_agent_factory

def test_agent_registration():
    """Test agent registration."""
    factory = AgentFactory()

    # Register a mock agent
    factory.register_builder(
        category="test",
        name="mock_agent",
        builder=lambda: MockAgent(),
    )

    # Verify registration
    agents = factory.list_agents(category="test")
    assert ("test", "mock_agent") in agents

def test_agent_building():
    """Test building an agent."""
    factory = get_agent_factory()

    agent = factory.build(category="research", name="analyst")
    assert agent is not None
    assert hasattr(agent, "name")

def test_singleton_pattern():
    """Test singleton behavior."""
    factory1 = get_agent_factory()
    factory2 = get_agent_factory()

    assert factory1 is factory2

Testing ToolFactory

import pytest
from agency.factories.tools import ToolFactory, get_tool_factory

@pytest.mark.asyncio
async def test_tool_execution():
    """Test tool building and execution."""
    factory = get_tool_factory()

    calculator = factory.build(category="computation", name="calculator")
    result = await calculator.execute(expression="2 + 2")

    assert result == 4.0

def test_tool_listing():
    """Test listing tools by category."""
    factory = get_tool_factory()

    search_tools = factory.list_tools(category="search")
    assert len(search_tools) > 0
    assert ("search", "duckduckgo") in search_tools

Testing AgencyFactory

import pytest
from agency.factories.agency_factory import AgencyFactory
from agency.agency.config import AgencyConfig
from agency.sessions.session_factory import SessionFactory

def test_basic_agency_creation():
    """Test basic agency creation."""
    config = AgencyConfig(name="TestAgency")
    agency = AgencyFactory.create_agency(config)

    assert agency.name == "TestAgency"
    assert agency.processor is not None

def test_agency_with_memory():
    """Test agency creation with memory session."""
    session = SessionFactory.create_memory_session("test")
    config = AgencyConfig(name="MemoryAgency")

    agency = AgencyFactory.create_agency(
        config=config,
        session=session
    )

    assert agency.session == session

@pytest.fixture
def mock_session():
    """Mock session for testing."""
    return SessionFactory.create_memory_session("mock")

def test_agency_with_fixture(mock_session):
    """Test using pytest fixture."""
    config = AgencyConfig(name="FixtureAgency")
    agency = AgencyFactory.create_agency(config, session=mock_session)

    assert agency.session == mock_session

Loader Pattern

The Loader is a powerful pattern that automatically discovers and loads constructors from Python modules without manual registration. This eliminates boilerplate code and enables plugin-style architectures.

When to Use Loader vs Factory

Pattern Use Case Registration Best For
Factory Manual control over constructors Explicit registration required Small, curated sets of components
Loader Automatic discovery No registration needed Large codebases, plugins, dynamic modules

Key Features

  • Automatic Discovery: Scans modules and finds matching classes/functions
  • Protocol Support: Filters by Protocol (structural typing)
  • Type Filtering: Filters by parent class or interface
  • Pattern Matching: Regex-based name filtering
  • Custom Conditions: Advanced filtering with custom functions
  • Name Templates: Transform discovered names (e.g., create_{name})
  • Lazy Loading: Constructors loaded on first access with caching
  • Graceful Errors: Missing modules are skipped without errors

Basic Usage

from agency.core.types import ToolProtocol
from agency.factories.base import Loader

# Create a loader for search tools
loader = Loader[ToolProtocol](
    template="agency.tools.search.crawl_4_ai_tool",  # Module path
    kind=ToolProtocol,  # Filter by Protocol
    ignore_suffixes=["Tool"],  # Remove "Tool" from names
    strict=True,  # Only include classes that match Protocol
)

# Discover what's available
print(f"Found {len(loader)} tools")
for (task, name), constructor in loader._constructors.items():
    print(f"  - {name}: {constructor.__name__}")

# Build a specific tool
search_tool = loader.build(task=None, name="search")
result = await search_tool.execute(query="Python async patterns")

Template-Based Discovery

Use {task} placeholders to scan multiple modules:

# Scan multiple tool categories
loader = Loader[ToolProtocol](
    template="agency.tools.{task}",  # Scans agency.tools.search, agency.tools.computation, etc.
    tasks=["search", "computation", "visualization"],  # Which tasks to load
    kind=ToolProtocol,
    ignore_suffixes=["Tool"],
)

# Tools are namespaced by task
search_tool = loader.build(task="search", name="crawl")
calculator = loader.build(task="computation", name="calculator")
drawing = loader.build(task="visualization", name="drawing")

Pattern-Based Filtering

Filter discovered constructors by name pattern:

# Only load tools with "search" in the name
loader = Loader[ToolProtocol](
    template="agency.tools.{task}",
    tasks=["search", "computation"],
    kind=ToolProtocol,
    pattern=r".*search.*",  # Regex pattern
    ignore_suffixes=["Tool"],
)

print(f"Found {len(loader)} tools matching '.*search.*'")

Custom Condition Filtering

Use custom functions for advanced filtering:

def has_execute_method(constructor):
    """Only load classes with an execute method."""
    return hasattr(constructor, "execute") or hasattr(constructor, "_execute")

loader = Loader[ToolProtocol](
    template="agency.tools.{task}",
    tasks=["search", "computation"],
    kind=ToolProtocol,
    condition=has_execute_method,  # Custom filter
    ignore_suffixes=["Tool"],
)

Name Template Transformation

Transform discovered names using templates:

# Prefix all names with "create_"
loader = Loader[ToolProtocol](
    template="agency.tools.search.crawl_4_ai_tool",
    kind=ToolProtocol,
    name_template="create_{name}",  # Transform names
    ignore_suffixes=["Tool"],
)

# Now access with prefixed name
tool = loader.build(task=None, name="create_search")

Loader Configuration Options

loader = Loader[T](
    template: str,              # Module path template (required)
    tasks: list[str] = None,    # Tasks for {task} placeholder
    kind: type = None,          # Filter by parent class/Protocol
    pattern: str = None,        # Regex pattern for names
    condition: Callable = None, # Custom filter function
    name_template: str = None,  # Name transformation template
    ignore_suffixes: list = None, # Suffixes to remove from names
    strict: bool = None,        # Strict Protocol checking (auto-detected)
)

Loader Introspection

# Check what was discovered
print(f"Total constructors: {len(loader)}")

# List all keys
for key in loader:
    print(f"  - {key}")

# Search for specific constructors
results = list(loader.search(task="search"))
print(f"Found {len(results)} in 'search' category")

# Access loader configuration
print(f"Template: {loader.template}")
print(f"Tasks: {loader.tasks}")
print(f"Kind: {loader.kind}")
print(f"Strict mode: {loader.strict}")

Combining Loader with Factory

Use Loader to populate a Factory for more control:

from agency.factories.base import Factory, Loader
from agency.core.types import ToolProtocol

# Create a loader to discover tools
loader = Loader[ToolProtocol](
    template="agency.tools.{task}",
    tasks=["search", "computation", "visualization"],
    kind=ToolProtocol,
    ignore_suffixes=["Tool"],
)

# Create a factory
factory = Factory[ToolProtocol]()

# Populate factory from loader
for (task, name), constructor in loader._constructors.items():
    factory.register(task, name, constructor)

# Now use factory with full Builder features
tool = factory.build(task="search", name="search")
print(factory.describe())  # Get formatted description

Inspection Utilities

The Loader uses inspection utilities that can be used independently:

from agency.utils.inspection import (
    get_classes_from_module,
    get_builder_name,
    get_builder_aliases,
    is_skipped,
    skip_builder,
    builder_name,
    builder_aliases,
)

# Get all classes from a module
import agency.tools.search.crawl_4_ai_tool as search_module
classes = get_classes_from_module(search_module, parent=ToolProtocol)
print(f"Found {len(classes)} classes implementing ToolProtocol")

# Get builder name with suffix removal
from agency.tools.search.crawl_4_ai_tool import SearchTool
name = get_builder_name(SearchTool, ignore_suffixes=["Tool"])
print(f"Builder name: {name}")  # "search"

# Use decorators for custom behavior
@skip_builder
class InternalTool(ToolProtocol):
    """This tool will be skipped by Loader."""
    pass

@builder_name("custom_search")
class SearchV2Tool(ToolProtocol):
    """This tool will have a custom name."""
    pass

@builder_aliases("quick_calc", "calc")
class CalculatorTool(ToolProtocol):
    """This tool has aliases."""
    pass

# Check if skipped
print(is_skipped(InternalTool))  # True
print(get_builder_name(SearchV2Tool))  # "custom_search"
print(get_builder_aliases(CalculatorTool))  # ["quick_calc", "calc"]

Testing with Loader

import pytest
from agency.factories.base import Loader
from agency.core.types import ToolProtocol

def test_loader_discovers_tools():
    """Test that Loader discovers tools from a module."""
    loader = Loader[ToolProtocol](
        template="agency.tools.search.crawl_4_ai_tool",
        kind=ToolProtocol,
        ignore_suffixes=["Tool"],
        strict=True,
    )

    assert len(loader) > 0
    assert any(name == "search" for task, name in loader)

def test_loader_builds_tool():
    """Test building a tool from loader."""
    loader = Loader[ToolProtocol](
        template="agency.tools.search.crawl_4_ai_tool",
        kind=ToolProtocol,
        ignore_suffixes=["Tool"],
        strict=True,
    )

    tool = loader.build(task=None, name="search")
    assert tool is not None
    assert hasattr(tool, "execute")

Best Practices

  1. Use for Large Codebases: Loader shines when you have many similar components
  2. Protocol-Based Design: Define clear Protocols for type safety
  3. Convention Over Configuration: Follow naming conventions instead of registration
  4. Combine with Factory: Use Loader for discovery, Factory for control
  5. Test Discovery: Write tests to ensure correct classes are discovered
  6. Use Strict Mode: Enable strict mode when filtering by Protocol
  7. Cache Results: Loader caches discovered constructors automatically
  8. Graceful Degradation: Missing modules won't break your application

Example: Complete Demonstration

See examples/demo_loader_pattern.py for a comprehensive demonstration including:

  • Basic loader usage
  • Pattern filtering
  • Custom condition filtering
  • Name template transformations
  • Loader introspection
  • Graceful error handling

Run the demo: 

poetry run python examples/demo_loader_pattern.py

Further Reading

  • Implementation: See agency/factories/base/loader.py for full implementation
  • Inspection Utilities: See agency/utils/inspection.py for helper functions
  • Tests: See tests/agency/factories/base/test_loader.py for comprehensive test coverage (38 tests, 100% pass rate)
  • Documentation: See docs/LOADER_IMPLEMENTATION.md and docs/LOADER_PROTOCOL_FIX.md for implementation details

Next Steps