DEV Community

Patrick Roebuck
Patrick Roebuck

Posted on

Inside the Agent Factory—where templates become task-specific specialists

#ai #python #machinelearning #agentaichallenge

Part 2 of the Grammar of AI Collaboration series.

In Part 1, we introduced the language metaphor: agents are words, composition patterns are grammar, and solutions are sentences. Today we dive deeper into the Agent Factory—the system that spawns customized agents on demand.

The Problem with Hard-Coded Agents

Traditional multi-agent systems define agents at design time:

# The old way - agents defined once, used everywhere
class SecurityAuditor:
    def __init__(self):
        self.model = "claude-3-opus"  # Fixed tier
        self.prompt = "You are a security auditor..."  # Generic instructions
        self.tools = ["code_reader", "vuln_scanner"]  # Fixed toolset

# Every security audit uses the same configuration
auditor = SecurityAuditor()
auditor.run(code)  # Same agent regardless of task
Enter fullscreen mode Exit fullscreen mode

What's wrong with this?

  1. Over-engineering: Simple security checks use premium models
  2. Under-powered: Complex audits don't get enhanced capabilities
  3. Rigidity: Can't adapt to project-specific requirements
  4. No specialization: Authentication audit same as API audit

The Solution: Agent Templates + Dynamic Spawning

We separate the archetype (what an agent CAN do) from the instance (what it DOES do for this task):

┌─────────────────────────────────────────────────────────────┐
│                    TEMPLATE LIBRARY                          │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  security_auditor                                     │   │
│  │  ├─ capabilities: [vuln_scan, threat_model, comply]   │   │
│  │  ├─ tier_preference: PREMIUM                          │   │
│  │  ├─ tools: [code_reader, vuln_scanner, cve_db]        │   │
│  │  └─ default_instructions: "Audit code for..."         │   │
│  └──────────────────────────────────────────────────────┘   │
│                           │                                  │
│                    SPAWN WITH CONTEXT                        │
│                           ↓                                  │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  security_auditor_a7f2 (INSTANCE)                     │   │
│  │  ├─ tier: CAPABLE (downgraded - simple task)          │   │
│  │  ├─ instructions: "Audit auth module for OWASP..."    │   │
│  │  ├─ focus: authentication                             │   │
│  │  └─ quality_gates: {max_high_vulns: 0}                │   │
│  └──────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

Anatomy of an Agent Template 

from dataclasses import dataclass, field
from typing import Any

@dataclass
class AgentTemplate:
    """Reusable agent archetype that can spawn customized instances."""

    # Identity
    id: str                         # "security_auditor"
    role: str                       # "Security Expert"

    # Capabilities (what it CAN do)
    capabilities: list[str]         # ["vuln_scan", "threat_model", "compliance"]

    # Resource preferences
    tier_preference: str            # "CHEAP", "CAPABLE", "PREMIUM"

    # Tools available
    tools: list[str]                # ["code_reader", "vuln_scanner"]

    # Default behavior
    default_instructions: str       # Base prompt template

    # Quality requirements
    quality_gates: dict[str, Any]   # {"min_score": 0.8, "max_vulns": 0}

    # Optional metadata
    category: str = "security"
    cost_weight: float = 1.0
    timeout_seconds: int = 300
Enter fullscreen mode Exit fullscreen mode

Example Templates 

TEMPLATES = {
    "security_auditor": AgentTemplate(
        id="security_auditor",
        role="Security Expert",
        capabilities=["vulnerability_scan", "threat_modeling", "compliance_check"],
        tier_preference="PREMIUM",
        tools=["code_reader", "vuln_scanner", "cve_database"],
        default_instructions="""
            Analyze code for security vulnerabilities.
            Focus on: {focus_areas}
            Severity threshold: {severity}
            Compliance requirements: {compliance}
        """,
        quality_gates={"max_critical": 0, "max_high": 3},
        category="security",
    ),

    "test_coverage_analyzer": AgentTemplate(
        id="test_coverage_analyzer",
        role="Test Coverage Expert",
        capabilities=["gap_analysis", "coverage_metrics", "test_suggestions"],
        tier_preference="CAPABLE",
        tools=["coverage_analyzer", "ast_parser", "test_framework"],
        default_instructions="""
            Analyze test coverage for the specified code.
            Target coverage: {target_coverage}%
            Focus modules: {modules}
            Identify: untested paths, edge cases, error handling
        """,
        quality_gates={"min_coverage": 80},
        category="testing",
    ),

    "documentation_writer": AgentTemplate(
        id="documentation_writer",
        role="Documentation Specialist",
        capabilities=["api_docs", "readme", "tutorials", "changelogs"],
        tier_preference="CHEAP",
        tools=["code_reader", "markdown_formatter"],
        default_instructions="""
            Generate documentation for the specified code.
            Style: {style}
            Audience: {audience}
            Include: {sections}
        """,
        quality_gates={"completeness": 0.9},
        category="documentation",
    ),
}
Enter fullscreen mode Exit fullscreen mode

The Agent Factory

The factory transforms templates into task-specific agents:

class AgentFactory:
    """Spawns agents from templates with task-specific customization."""

    def __init__(self, templates: dict[str, AgentTemplate]):
        self.templates = templates
        self.spawned_count = 0

    def spawn(
        self,
        template_id: str,
        requirements: TaskRequirements,
        overrides: dict | None = None
    ) -> Agent:
        """Create agent instance from template."""

        template = self.templates[template_id]

        # 1. SELECT APPROPRIATE TIER
        tier = self._select_tier(template.tier_preference, requirements)

        # 2. CUSTOMIZE INSTRUCTIONS
        instructions = self._customize_instructions(
            template.default_instructions,
            requirements,
            overrides or {}
        )

        # 3. FILTER TOOLS (only what's needed)
        tools = self._select_tools(template.tools, requirements)

        # 4. CONFIGURE QUALITY GATES
        quality_gates = self._configure_gates(
            template.quality_gates,
            requirements
        )

        # 5. SPAWN INSTANCE
        self.spawned_count += 1
        instance_id = f"{template.id}_{self.spawned_count:04d}"

        return Agent(
            id=instance_id,
            template_id=template.id,
            role=template.role,
            tier=tier,
            instructions=instructions,
            tools=tools,
            quality_gates=quality_gates,
            capabilities=template.capabilities,
        )
Enter fullscreen mode Exit fullscreen mode

Intelligent Tier Selection

The factory doesn't blindly use the template's preferred tier—it right-sizes based on the task:

def _select_tier(
    self,
    preference: str,
    requirements: TaskRequirements
) -> str:
    """Select appropriate tier based on task requirements."""

    # Complexity scores: CHEAP=1, CAPABLE=2, PREMIUM=3
    tier_scores = {"CHEAP": 1, "CAPABLE": 2, "PREMIUM": 3}
    preference_score = tier_scores[preference]

    # Adjust based on requirements
    adjustments = 0

    # Critical tasks get upgraded
    if requirements.is_critical:
        adjustments += 1

    # Simple tasks get downgraded
    if requirements.complexity == "simple":
        adjustments -= 1

    # User override
    if requirements.force_tier:
        return requirements.force_tier

    # Calculate final tier
    final_score = max(1, min(3, preference_score + adjustments))
    return {1: "CHEAP", 2: "CAPABLE", 3: "PREMIUM"}[final_score]
Enter fullscreen mode Exit fullscreen mode

Examples of tier selection:

Template Task Selected Tier Why
security_auditor (PREMIUM) Production release PREMIUM Critical + matches preference
security_auditor (PREMIUM) Dev branch check CAPABLE Non-critical, downgraded
documentation_writer (CHEAP) API docs CHEAP Matches preference
documentation_writer (CHEAP) Executive summary CAPABLE Upgraded for quality

Dynamic Instruction Customization

Instructions are templates that get filled with task-specific context:

def _customize_instructions(
    self,
    template_instructions: str,
    requirements: TaskRequirements,
    overrides: dict
) -> str:
    """Generate task-specific instructions."""

    # Build context from requirements
    context = {
        "focus_areas": requirements.focus or "general",
        "severity": requirements.severity_threshold or "medium",
        "compliance": ", ".join(requirements.compliance or ["none"]),
        "modules": ", ".join(requirements.target_modules or ["all"]),
        "target_coverage": requirements.coverage_target or 80,
        "style": requirements.doc_style or "technical",
        "audience": requirements.audience or "developers",
        "sections": ", ".join(requirements.sections or ["overview", "api"]),
    }

    # Apply overrides
    context.update(overrides)

    # Generate customized instructions
    instructions = template_instructions.format(**context)

    # Add task-specific context
    if requirements.additional_context:
        instructions += f"\n\nAdditional context:\n{requirements.additional_context}"

    return instructions
Enter fullscreen mode Exit fullscreen mode

Example transformation:

# Template instructions
"""
Analyze code for security vulnerabilities.
Focus on: {focus_areas}
Severity threshold: {severity}
Compliance requirements: {compliance}
"""

# Requirements
requirements = TaskRequirements(
    focus="authentication, session management",
    severity_threshold="high",
    compliance=["OWASP", "GDPR"],
    additional_context="Recent breach attempt on login endpoint"
)

# Spawned agent receives:
"""
Analyze code for security vulnerabilities.
Focus on: authentication, session management
Severity threshold: high
Compliance requirements: OWASP, GDPR

Additional context:
Recent breach attempt on login endpoint
"""
Enter fullscreen mode Exit fullscreen mode

Spawning in Action

Let's trace a complete spawn operation:

# User request
result = orchestrate("security audit on auth module for release")

# Meta-orchestrator analyzes task
requirements = TaskRequirements(
    intent="security_audit",
    target_modules=["auth"],
    is_critical=True,  # Release = critical
    compliance=["OWASP", "PCI-DSS"],
    focus="authentication, authorization, session",
    severity_threshold="high",
    complexity="moderate"
)

# Factory spawns agent
factory = AgentFactory(TEMPLATES)
agent = factory.spawn(
    template_id="security_auditor",
    requirements=requirements
)

# Result:
print(agent)
# Agent(
#     id="security_auditor_0001",
#     template_id="security_auditor",
#     role="Security Expert",
#     tier="PREMIUM",  # Critical + PREMIUM preference
#     instructions="Analyze code for security vulnerabilities.
#                   Focus on: authentication, authorization, session
#                   Severity threshold: high
#                   Compliance requirements: OWASP, PCI-DSS",
#     tools=["code_reader", "vuln_scanner", "cve_database"],
#     quality_gates={"max_critical": 0, "max_high": 0},  # Stricter for release
#     capabilities=["vulnerability_scan", "threat_modeling", "compliance_check"]
# )
Enter fullscreen mode Exit fullscreen mode

Template Registry & Discovery

Templates are registered in a central library:

class TemplateRegistry:
    """Central registry for agent templates."""

    def __init__(self):
        self._templates: dict[str, AgentTemplate] = {}
        self._by_capability: dict[str, list[str]] = {}

    def register(self, template: AgentTemplate):
        """Register a template."""
        self._templates[template.id] = template

        # Index by capability
        for cap in template.capabilities:
            if cap not in self._by_capability:
                self._by_capability[cap] = []
            self._by_capability[cap].append(template.id)

    def find_by_capability(self, capability: str) -> list[AgentTemplate]:
        """Find templates that have a capability."""
        template_ids = self._by_capability.get(capability, [])
        return [self._templates[tid] for tid in template_ids]

    def find_best_match(self, requirements: TaskRequirements) -> AgentTemplate:
        """Find the best template for requirements."""
        candidates = []

        for cap in requirements.needed_capabilities:
            candidates.extend(self.find_by_capability(cap))

        # Score by capability coverage
        scored = []
        for template in set(candidates):
            coverage = len(set(template.capabilities) & set(requirements.needed_capabilities))
            scored.append((coverage, template))

        scored.sort(reverse=True, key=lambda x: x[0])
        return scored[0][1] if scored else None
Enter fullscreen mode Exit fullscreen mode

Capability-based discovery:

registry = TemplateRegistry()
registry.register(TEMPLATES["security_auditor"])
registry.register(TEMPLATES["test_coverage_analyzer"])

# Find templates for "vulnerability scanning"
matches = registry.find_by_capability("vulnerability_scan")
# → [security_auditor]

# Find best match for complex requirements
requirements = TaskRequirements(
    needed_capabilities=["vulnerability_scan", "compliance_check"]
)
best = registry.find_best_match(requirements)
# → security_auditor (has both capabilities)
Enter fullscreen mode Exit fullscreen mode

Creating Custom Templates

Users can define their own templates:

# Custom template for your organization
custom_template = AgentTemplate(
    id="internal_code_reviewer",
    role="Internal Code Standards Expert",
    capabilities=["style_check", "pattern_enforcement", "team_conventions"],
    tier_preference="CHEAP",
    tools=["code_reader", "style_checker"],
    default_instructions="""
        Review code against internal standards:
        - Naming conventions: {naming_style}
        - Architecture patterns: {patterns}
        - Forbidden patterns: {forbidden}

        Be strict about: {strict_rules}
        Be lenient about: {lenient_rules}
    """,
    quality_gates={"min_style_score": 0.9},
    category="code_review",
)

# Register it
registry.register(custom_template)

# Now available for orchestration
result = orchestrate("review PR for internal standards")
# → Uses internal_code_reviewer template
Enter fullscreen mode Exit fullscreen mode

Performance Characteristics

Operation Latency Notes
Template lookup <1ms Dictionary lookup
Tier selection <1ms Simple arithmetic
Instruction customization <5ms String formatting
Tool filtering <1ms Set intersection
Full spawn <10ms All operations combined

Spawning is cheap. The real cost is agent execution, not creation.

Key Takeaways

  1. Templates define capabilities, not behavior — The same security_auditor template spawns different agents for auth audits vs. API audits

  2. Dynamic customization at spawn time — Instructions, tier, tools all adapt to the task

  3. Right-sizing is automatic — Critical tasks get premium resources, simple tasks get efficient execution

  4. Capability-based discovery — System finds appropriate templates by matching capabilities to requirements

  5. Extensible by design — Add custom templates for your organization's needs

What's Next

In Part 3: Building Agent Teams, we'll see how multiple spawned agents combine into teams using composition patterns.

The Agent Factory code is available in Empathy Framework v4.4.0.

*Patrick Roebuck is the creator of the Empathy Framework.

Top comments (0)