Forem: Patrick Roebuck

How composition patterns transform individual agents into collaborative teams

Patrick Roebuck — Mon, 19 Jan 2026 13:25:16 +0000

Part 3 of the Grammar of AI Collaboration series.

In Part 1, we introduced the grammar metaphor. In Part 2, we explored how agents spawn dynamically. Now we tackle the heart of the system: composition patterns—the grammar rules that turn individual agents into coordinated teams.

Why Composition Matters

Having great agents isn't enough. Consider:

🤖 Security Auditor: "Found 3 critical vulnerabilities"
🤖 Code Reviewer: "Code quality is excellent"
🤖 Test Analyzer: "Coverage is 45%"

❓ Human: "Should I release?"

Three agents, three perspectives, no synthesis. Who resolves conflicts? Who prioritizes? Who decides?

Composition patterns answer these questions.

The 6 Core Patterns

Think of these as the verbs of agent orchestration—they describe how agents collaborate:

┌─────────────────────────────────────────────────────────────────┐
│                     COMPOSITION PATTERNS                         │
├─────────────────────────────────────────────────────────────────┤
│  Sequential  │  A → B → C        │  Pipeline, dependencies     │
│  Parallel    │  A ‖ B ‖ C        │  Independent, speed         │
│  Debate      │  A ⇄ B → Synth    │  Multiple perspectives      │
│  Teaching    │  Junior → Expert  │  Cost + quality             │
│  Refinement  │  Draft → Polish   │  Iterative improvement      │
│  Adaptive    │  Route → Spec     │  Right-size by complexity   │
└─────────────────────────────────────────────────────────────────┘

Let's explore each one.

Pattern 1: Sequential (A → B → C)

When to use: Each step depends on the previous step's output.

coverage_analyzer → test_generator → quality_validator
       ↓                  ↓                  ↓
  "45% coverage"    "15 new tests"    "All tests pass"

Implementation

class SequentialStrategy(ExecutionStrategy):
    """Execute agents one after another, passing results forward."""

    async def execute(
        self,
        agents: list[Agent],
        context: dict
    ) -> StrategyResult:
        results = []
        current_context = context.copy()

        for agent in agents:
            # Execute agent with accumulated context
            result = await agent.execute(current_context)
            results.append(result)

            # Pass results forward
            current_context = {**current_context, **result.output}

        return StrategyResult(
            success=all(r.success for r in results),
            outputs=results,
            aggregated_output=current_context,
            total_duration=sum(r.duration for r in results)
        )

Real-World Example: Test Coverage Boost

# Define the team
team = AgentTeam(
    agents=[
        spawn("coverage_analyzer", focus="auth module"),
        spawn("test_generator", style="pytest"),
        spawn("quality_validator", min_score=0.8)
    ],
    strategy="sequential"
)

# Execute
result = await team.execute({
    "code_path": "src/auth/",
    "target_coverage": 80
})

# Flow:
# 1. coverage_analyzer identifies gaps → {gaps: ["login", "logout", "refresh"]}
# 2. test_generator receives gaps → {tests: ["test_login.py", ...]}
# 3. quality_validator receives tests → {quality_score: 0.92, passed: True}

Strengths: Clear dependencies, traceable flow, each agent has full context from predecessors.

Weaknesses: Slow (serial execution), one failure stops the pipeline.

Pattern 2: Parallel (A ‖ B ‖ C)

When to use: Independent checks that can run simultaneously.

┌─────────────────────────────────────────┐
│           START                          │
└─────────────┬───────────────────────────┘
              │
    ┌─────────┼─────────┐
    ↓         ↓         ↓
┌───────┐ ┌───────┐ ┌───────┐
│ Sec   │ │ Perf  │ │ Docs  │
│ Audit │ │ Check │ │ Check │
└───┬───┘ └───┬───┘ └───┬───┘
    │         │         │
    └─────────┼─────────┘
              ↓
    ┌─────────────────┐
    │   AGGREGATOR    │
    └─────────────────┘

Implementation

class ParallelStrategy(ExecutionStrategy):
    """Execute all agents simultaneously, aggregate results."""

    async def execute(
        self,
        agents: list[Agent],
        context: dict
    ) -> StrategyResult:
        # Launch all agents concurrently
        tasks = [agent.execute(context) for agent in agents]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Handle failures
        successful = [r for r in results if not isinstance(r, Exception)]
        failed = [r for r in results if isinstance(r, Exception)]

        return StrategyResult(
            success=len(failed) == 0,
            outputs=successful,
            aggregated_output=self._aggregate(successful),
            total_duration=max(r.duration for r in successful),
            errors=failed
        )

    def _aggregate(self, results: list[AgentResult]) -> dict:
        """Combine results with weighted scoring."""
        combined = {}
        for result in results:
            for key, value in result.output.items():
                combined[f"{result.agent_id}.{key}"] = value

        # Calculate overall score
        scores = [r.output.get("score", 0) for r in results]
        combined["overall_score"] = sum(scores) / len(scores)

        return combined

Real-World Example: Release Preparation

# Define parallel checks
team = AgentTeam(
    agents=[
        spawn("security_auditor", severity="high"),
        spawn("performance_validator", sla="100ms"),
        spawn("documentation_checker", completeness=0.9),
        spawn("test_coverage_analyzer", target=80)
    ],
    strategy="parallel"
)

# Execute - all run at once
result = await team.execute({"release_candidate": "v4.4.0"})

# Aggregated result:
# {
#     "security_auditor.vulnerabilities": 0,
#     "security_auditor.score": 95,
#     "performance_validator.p99_latency": 82,
#     "performance_validator.score": 90,
#     "documentation_checker.completeness": 0.94,
#     "documentation_checker.score": 94,
#     "test_coverage_analyzer.coverage": 87,
#     "test_coverage_analyzer.score": 87,
#     "overall_score": 91.5  # Weighted average
# }

Strengths: Fast (parallel execution), comprehensive (multiple perspectives).

Weaknesses: No inter-agent communication, all agents see same input.

Pattern 3: Debate (A ⇄ B → Synthesis)

When to use: Complex decisions needing multiple expert perspectives.

┌──────────────┐          ┌──────────────┐
│  Architect   │ ⇄ ⇄ ⇄ ⇄  │  Architect   │
│  (Scale)     │          │  (Cost)      │
└──────┬───────┘          └──────┬───────┘
       │                         │
       └───────────┬─────────────┘
                   ↓
           ┌──────────────┐
           │  Synthesizer │
           │  (Decision)  │
           └──────────────┘

Implementation

class DebateStrategy(ExecutionStrategy):
    """Multiple perspectives with synthesis."""

    def __init__(self, rounds: int = 2, synthesizer: Agent | None = None):
        self.rounds = rounds
        self.synthesizer = synthesizer

    async def execute(
        self,
        agents: list[Agent],
        context: dict
    ) -> StrategyResult:
        debate_history = []
        current_context = context.copy()

        # Debate rounds
        for round_num in range(self.rounds):
            round_results = []

            for agent in agents:
                # Agent sees other agents' previous responses
                agent_context = {
                    **current_context,
                    "debate_history": debate_history,
                    "round": round_num + 1
                }
                result = await agent.execute(agent_context)
                round_results.append(result)

            debate_history.append({
                "round": round_num + 1,
                "responses": [r.output for r in round_results]
            })

        # Synthesis
        if self.synthesizer:
            synthesis_context = {
                **context,
                "debate_history": debate_history
            }
            synthesis = await self.synthesizer.execute(synthesis_context)
            return StrategyResult(
                success=True,
                outputs=debate_history,
                aggregated_output=synthesis.output,
                total_duration=sum(...)
            )

        return self._auto_synthesize(debate_history)

Real-World Example: Architecture Decision

# Define debaters
team = AgentTeam(
    agents=[
        spawn("architect", focus="scalability"),
        spawn("architect", focus="cost_efficiency"),
        spawn("architect", focus="simplicity")
    ],
    strategy=DebateStrategy(
        rounds=2,
        synthesizer=spawn("decision_maker", style="consensus")
    )
)

# Execute debate
result = await team.execute({
    "decision": "How should we handle caching?",
    "constraints": ["budget: $500/mo", "latency: <50ms", "team_size: 3"]
})

# Debate flow:
# Round 1:
#   scale_architect: "Use Redis cluster for horizontal scaling"
#   cost_architect: "In-memory cache sufficient, Redis overkill"
#   simple_architect: "Start with functools.lru_cache"
#
# Round 2 (responds to each other):
#   scale_architect: "lru_cache doesn't share across instances"
#   cost_architect: "Single Redis instance balances cost/capability"
#   simple_architect: "Agree with single Redis, simpler than cluster"
#
# Synthesis:
#   "Recommendation: Single Redis instance with local LRU fallback.
#    Rationale: Balances scalability needs with cost constraints.
#    Migration path: Start simple, scale cluster when >1000 RPS."

Strengths: Rich decision-making, surfaces trade-offs, reduces single-agent bias.

Weaknesses: Expensive (multiple rounds × multiple agents), slower.

Pattern 4: Teaching (Junior → Expert Validation)

When to use: Cost optimization with quality assurance.

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Junior Writer  │  →  │  Quality Gate   │  →  │  Expert Review  │
│    (CHEAP)      │     │    (check)      │     │   (CAPABLE)     │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                              ↓ pass                    ↓
                         ┌─────────┐              ┌─────────┐
                         │  DONE   │              │ REFINED │
                         └─────────┘              └─────────┘

Implementation

class TeachingStrategy(ExecutionStrategy):
    """Junior generates, expert validates/refines."""

    def __init__(
        self,
        junior: Agent,
        expert: Agent,
        quality_threshold: float = 0.8
    ):
        self.junior = junior
        self.expert = expert
        self.threshold = quality_threshold

    async def execute(
        self,
        agents: list[Agent],  # Ignored, uses junior/expert
        context: dict
    ) -> StrategyResult:
        # Junior attempt (cheap tier)
        junior_result = await self.junior.execute(context)

        # Quality check
        quality_score = self._assess_quality(junior_result)

        if quality_score >= self.threshold:
            # Passed! Junior output sufficient
            return StrategyResult(
                success=True,
                outputs=[junior_result],
                aggregated_output=junior_result.output,
                metadata={"tier_used": "CHEAP", "expert_needed": False}
            )

        # Expert refinement needed
        expert_context = {
            **context,
            "junior_output": junior_result.output,
            "quality_issues": self._identify_issues(junior_result)
        }
        expert_result = await self.expert.execute(expert_context)

        return StrategyResult(
            success=True,
            outputs=[junior_result, expert_result],
            aggregated_output=expert_result.output,
            metadata={"tier_used": "CAPABLE", "expert_needed": True}
        )

Real-World Example: Documentation Generation

# Cost-optimized documentation
strategy = TeachingStrategy(
    junior=spawn("documentation_writer", tier="CHEAP"),
    expert=spawn("documentation_writer", tier="CAPABLE"),
    quality_threshold=0.85
)

result = await strategy.execute([], {
    "code": api_module,
    "style": "technical",
    "audience": "developers"
})

# Scenario A: Junior passes (85%+ quality)
#   → Cost: $0.002 (Haiku only)
#   → Output: Junior's documentation
#
# Scenario B: Junior fails (< 85% quality)
#   → Cost: $0.002 + $0.015 = $0.017 (Haiku + Sonnet)
#   → Output: Expert-refined documentation

Cost savings: 60-80% when junior passes frequently.

Strengths: Cost-effective, quality maintained, automatic escalation.

Weaknesses: Requires good quality assessment, two-tier latency on escalation.

Pattern 5: Refinement (Draft → Review → Polish)

When to use: Iterative improvement for high-quality output.

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   Drafter    │  →  │   Reviewer   │  →  │   Polisher   │
│   (CHEAP)    │     │  (CAPABLE)   │     │  (PREMIUM)   │
└──────────────┘     └──────────────┘     └──────────────┘
       ↓                    ↓                    ↓
   "Raw draft"      "Improved + notes"    "Publication-ready"

Implementation

class RefinementStrategy(ExecutionStrategy):
    """Progressive quality improvement ladder."""

    def __init__(self, stages: list[tuple[Agent, str]]):
        # [(agent, role), ...]
        self.stages = stages

    async def execute(
        self,
        agents: list[Agent],
        context: dict
    ) -> StrategyResult:
        current_output = None
        stage_results = []

        for agent, role in self.stages:
            stage_context = {
                **context,
                "previous_output": current_output,
                "role": role,
                "stage": len(stage_results) + 1
            }

            result = await agent.execute(stage_context)
            stage_results.append(result)
            current_output = result.output

        return StrategyResult(
            success=True,
            outputs=stage_results,
            aggregated_output=current_output,  # Final polished version
            metadata={"stages_completed": len(stage_results)}
        )

Real-World Example: API Documentation Pipeline

strategy = RefinementStrategy(stages=[
    (spawn("doc_writer", tier="CHEAP"), "drafter"),
    (spawn("doc_reviewer", tier="CAPABLE"), "reviewer"),
    (spawn("doc_editor", tier="PREMIUM"), "polisher")
])

result = await strategy.execute([], {
    "code": payment_api,
    "standard": "OpenAPI 3.0"
})

# Stage 1 - Drafter (CHEAP):
#   "Generates basic structure, extracts endpoints"
#
# Stage 2 - Reviewer (CAPABLE):
#   "Adds examples, improves descriptions, checks accuracy"
#
# Stage 3 - Polisher (PREMIUM):
#   "Perfects language, ensures consistency, adds edge cases"

Strengths: Highest quality output, clear improvement stages.

Weaknesses: Most expensive, slowest, not always necessary.

Pattern 6: Adaptive Routing (Classifier → Specialist)

When to use: Variable complexity tasks that need right-sizing.

           ┌─────────────────┐
           │   Classifier    │
           │    (CHEAP)      │
           └────────┬────────┘
                    │
        ┌───────────┼───────────┐
        ↓           ↓           ↓
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │ Simple  │ │ Medium  │ │ Complex │
   │ (CHEAP) │ │(CAPABLE)│ │(PREMIUM)│
   └─────────┘ └─────────┘ └─────────┘

Implementation

class AdaptiveStrategy(ExecutionStrategy):
    """Route to appropriate specialist based on complexity."""

    def __init__(
        self,
        classifier: Agent,
        specialists: dict[str, Agent]  # complexity → agent
    ):
        self.classifier = classifier
        self.specialists = specialists

    async def execute(
        self,
        agents: list[Agent],
        context: dict
    ) -> StrategyResult:
        # Classify task complexity (cheap operation)
        classification = await self.classifier.execute(context)
        complexity = classification.output["complexity"]

        # Route to appropriate specialist
        specialist = self.specialists.get(complexity)
        if not specialist:
            specialist = self.specialists["default"]

        result = await specialist.execute({
            **context,
            "classification": classification.output
        })

        return StrategyResult(
            success=result.success,
            outputs=[classification, result],
            aggregated_output=result.output,
            metadata={
                "classified_as": complexity,
                "specialist_used": specialist.id
            }
        )

Real-World Example: Bug Triage

strategy = AdaptiveStrategy(
    classifier=spawn("bug_classifier", tier="CHEAP"),
    specialists={
        "simple": spawn("bug_fixer", tier="CHEAP"),      # Typos, config
        "moderate": spawn("bug_fixer", tier="CAPABLE"),   # Logic errors
        "complex": spawn("bug_fixer", tier="PREMIUM"),    # Architecture issues
        "default": spawn("bug_fixer", tier="CAPABLE")
    }
)

# Bug: "Login button doesn't work"
result = await strategy.execute([], {"bug_report": bug})

# Classifier: "simple" (missing event handler)
# → Routes to CHEAP tier fixer
# → Cost: $0.005
#
# Bug: "Race condition in payment processing"
# Classifier: "complex" (concurrency issue)
# → Routes to PREMIUM tier fixer
# → Cost: $0.10

Strengths: Cost-optimized, right-sized automatically.

Weaknesses: Classifier accuracy is critical, misclassification wastes resources.

Choosing the Right Pattern

Scenario	Recommended Pattern
Steps depend on each other	Sequential
Independent validations	Parallel
Need multiple expert opinions	Debate
Cost matters, quality negotiable	Teaching
Highest quality required	Refinement
Variable task complexity	Adaptive

Pattern Selection Heuristic

def select_pattern(requirements: TaskRequirements) -> str:
    # Dependencies → Sequential
    if requirements.has_dependencies:
        return "sequential"

    # Multiple perspectives needed → Debate
    if requirements.needs_consensus:
        return "debate"

    # Cost-sensitive → Teaching or Adaptive
    if requirements.cost_sensitive:
        if requirements.variable_complexity:
            return "adaptive"
        return "teaching"

    # Highest quality → Refinement
    if requirements.quality_critical:
        return "refinement"

    # Default: Parallel for speed
    return "parallel"

Combining Patterns

Patterns can be nested. A release preparation workflow might use:

parallel([
    sequential([coverage_analyzer, test_generator, validator]),
    teaching(junior_security, expert_security),
    debate([arch_scale, arch_cost], synthesizer)
]) → weighted_aggregation

This creates a sophisticated team that:

Runs test boost sequentially (dependencies)
Runs security with cost optimization (teaching)
Debates architecture decisions (debate)
All three branches run in parallel (parallel outer)

What's Next

In Part 4: Advanced Grammar, we explore the newest patterns: conditional branching, nested workflows, and learning grammar that improves from experience.

Composition pattern code is available in Empathy Framework v4.4.0.

*Patrick Roebuck is the creator of the Empathy Framework.

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

Patrick Roebuck — Mon, 19 Jan 2026 13:18:30 +0000

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

What's wrong with this?

Over-engineering: Simple security checks use premium models
Under-powered: Complex audits don't get enhanced capabilities
Rigidity: Can't adapt to project-specific requirements
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}                │   │
│  └──────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘

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

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",
    ),
}

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,
        )

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]

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

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
"""

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"]
# )

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

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)

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

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

Templates define capabilities, not behavior — The same security_auditor template spawns different agents for auth audits vs. API audits
Dynamic customization at spawn time — Instructions, tier, tools all adapt to the task
Right-sizing is automatic — Critical tasks get premium resources, simple tasks get efficient execution
Capability-based discovery — System finds appropriate templates by matching capabilities to requirements
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.

What if AI agents could compose themselves like words form sentences?

Patrick Roebuck — Mon, 19 Jan 2026 13:14:30 +0000

Most AI frameworks treat agents as black boxes. You define them once, hard-code their behaviors, and pray they work together. When requirements change, you rewrite everything.

We took a different approach.

What if agent orchestration followed the same rules as language? Words combine into sentences through grammar. Agents combine into solutions through composition patterns. Learn the grammar, and you can express any idea.

The Language Metaphor

Think about how language works:

Language	Agent Orchestration
Words	Individual agents (test_analyzer, security_auditor, code_reviewer)
Grammar Rules	Composition patterns (sequential, parallel, debate, teaching)
Sentences	Complete solutions (release preparation, test coverage boost)

A child who knows 500 words and basic grammar can express millions of ideas. An orchestration system with 20 agent templates and 10 composition patterns can solve thousands of unique tasks.

That's the power of composability.

From Static Workflows to Dynamic Teams

The Old Way (Static)

# Fixed workflow - hard-coded agents, fixed order
workflow = Workflow([
    SecurityAuditor(),
    CodeReviewer(),
    TestValidator()
])
workflow.run()  # Same agents, same order, every time

Problems:

Want to add documentation? Rewrite the workflow
Different project needs security focus? New workflow
Task simpler than expected? Still runs full pipeline

The New Way (Dynamic)

# Dynamic composition - agents spawn based on requirements
result = orchestrate("prepare for release")

# System analyzes task, spawns appropriate team:
# → Security Auditor (PREMIUM tier - critical for release)
# → Test Coverage Analyzer (CAPABLE tier)
# → Documentation Checker (CHEAP tier - fast validation)
# → Code Quality Reviewer (CAPABLE tier)
#
# Executes in PARALLEL for speed
# Aggregates with weighted scoring

The system:

Analyzes what "prepare for release" requires
Selects appropriate agents from a template library
Chooses the optimal composition pattern
Spawns agents with task-specific instructions
Executes and aggregates results
Learns from the outcome for next time

The 10 Grammar Rules

We've implemented 10 composition patterns—the "grammar" of agent collaboration:

Basic Patterns (Rules 1-6)

Rule 1: Sequential (A → B → C)
  └─ Tasks done in order, each depends on previous
  └─ Example: analyze_coverage → generate_tests → validate_quality

Rule 2: Parallel (A || B || C)
  └─ Independent tasks run simultaneously
  └─ Example: [security || performance || docs || tests] → aggregate

Rule 3: Debate (A ⇄ B ⇄ C → Synthesis)
  └─ Multiple perspectives, synthesize consensus
  └─ Example: architect_scale ⇄ architect_cost → synthesizer

Rule 4: Teaching (Junior → Expert Validation)
  └─ Cost-effective generation with quality assurance
  └─ Example: junior_writer(CHEAP) → expert_review(CAPABLE)

Rule 5: Refinement (Draft → Review → Polish)
  └─ Iterative improvement ladder
  └─ Example: drafter → reviewer → polisher

Rule 6: Adaptive (Classifier → Specialist)
  └─ Route based on complexity
  └─ Example: classifier → route(simple|complex) → specialist

Advanced Patterns (Rules 7-10)

These are the new patterns we just released:

Rule 7: Conditional (IF condition THEN A ELSE B)
  └─ Branching based on runtime conditions
  └─ Example: IF confidence < 0.8 THEN expert_review ELSE auto_approve

Rule 8: Multi-Conditional (SWITCH/CASE)
  └─ Multiple condition branches
  └─ Example: CASE severity: critical→emergency | high→urgent | *→normal

Rule 9: Nested (Workflows within workflows)
  └─ Hierarchical composition with depth limits
  └─ Example: release_prep contains [security_deep_dive, test_coverage_boost]

Rule 10: Learning (Patterns that improve from experience)
  └─ Track success rates, recommend optimal patterns
  └─ Example: "For security tasks, parallel pattern has 92% success rate"

Dynamic Agent Creation in Action

Here's what happens when you ask the system to "boost test coverage":

# You write this:
result = orchestrate("boost test coverage for auth module")

# The system does this:
#
# 1. TASK ANALYSIS
#    └─ Intent: improve_tests
#    └─ Scope: auth module
#    └─ Complexity: moderate
#
# 2. AGENT SELECTION (from template library)
#    └─ coverage_analyzer (CAPABLE) - identify gaps
#    └─ test_generator (CAPABLE) - create tests
#    └─ quality_validator (CAPABLE) - verify quality
#
# 3. PATTERN SELECTION
#    └─ Sequential: analyze → generate → validate
#    └─ Why: Each step depends on previous
#
# 4. AGENT SPAWNING (dynamic customization)
#    coverage_analyzer_a7f2:
#      role: "Coverage Expert"
#      instructions: "Focus on auth module, identify untested paths"
#      tier: CAPABLE
#      tools: [coverage_analyzer, ast_parser]
#      quality_gates: {min_coverage: 80}
#
# 5. EXECUTION
#    coverage_analyzer → test_generator → quality_validator
#
# 6. LEARNING
#    └─ Record: sequential pattern, 3 agents, success=True
#    └─ Update: success_rate for "test_boost" pattern
#    └─ Store: for future reuse

The Agent Template System

Agents are spawned from reusable templates:

@dataclass
class AgentTemplate:
    """Reusable agent archetype."""
    id: str                        # "test_coverage_analyzer"
    role: str                      # "Test Coverage Expert"
    capabilities: list[str]        # ["analyze_gaps", "suggest_tests"]
    tier_preference: str           # "CAPABLE"
    tools: list[str]               # ["coverage_analyzer", "ast_parser"]
    default_instructions: str      # Base prompt
    quality_gates: dict[str, Any]  # {"min_coverage": 80}

When a template spawns an agent, it's customized for the specific task:

# Template defines archetype
template = TEMPLATES["security_auditor"]

# Factory spawns customized instance
agent = factory.spawn(template, requirements={
    "focus": "authentication",
    "severity_threshold": "high",
    "compliance": ["OWASP", "GDPR"]
})

# Resulting agent has task-specific instructions:
# "Audit authentication code for vulnerabilities.
#  Flag HIGH severity issues and above.
#  Check compliance with OWASP Top 10 and GDPR requirements."

20 templates + dynamic customization = unlimited agents

Why This Matters

1. Reduced Boilerplate

Define templates once, reuse everywhere. No more copy-pasting agent configurations.

2. Right-Sizing

System automatically selects appropriate tier (CHEAP/CAPABLE/PREMIUM) based on task complexity.

3. Automatic Learning

Every execution teaches the system. Success rates tracked. Optimal patterns discovered.

4. Composability

Combine patterns freely. Nest workflows. Add conditions. Build complex behaviors from simple rules.

5. Transparency

Clear execution plans. Visible decisions. Understand why agents were chosen.

Getting Started

# Install
pip install empathy-framework

# Dynamic orchestration
empathy orchestrate "prepare for release"

# Specific task with context
empathy orchestrate "boost test coverage" --context '{"module": "auth"}'

# See what patterns have been learned
empathy patterns report

What's Next

This is Part 1 of a 4-part series on the Grammar of AI Collaboration:

The Grammar of AI Collaboration ← You are here
Dynamic Agent Creation Deep Dive — Inside the Agent Factory
Building Agent Teams — Composition patterns in action
Advanced Grammar — Conditional logic, nesting, and learning

The code for this article is available in Empathy Framework v4.4.0 on GitHub.

*Patrick Roebuck is the creator of the Empathy Framework.

Enterprise-Ready AI Workflows: Formatted Reports + 80% Cost Savings

Patrick Roebuck — Fri, 26 Dec 2025 13:38:53 +0000

Just shipped v3.3.0 of Empathy Framework with features I wish existed when I was running AI at scale:

Formatted reports for every workflow (finally, readable output)
Cost guardrails so your doc-gen doesn't blow $50 overnight
File export because 50k character terminal limits are real

Here's what changed—and why it matters.

**The Problem with AI Workflows

Most AI libraries return raw JSON or unstructured text. Fine for prototypes. Terrible for:

Reports you need to share with stakeholders
Outputs you need to audit
Results that exceed terminal/UI display limits

The Solution: Formatted Reports for All Workflows

Every workflow in v3.3.0 now includes a formatted_report with consistent structure:

from empathy_os.workflows import SecurityAuditWorkflow

workflow = SecurityAuditWorkflow()
result = await workflow.execute(code=your_code)

print(result.final_output["formatted_report"])

Output:

============================================================
SECURITY AUDIT REPORT
============================================================

Status: NEEDS_ATTENTION
Risk Score: 7.2/10
Vulnerabilities Found: 3

------------------------------------------------------------
CRITICAL FINDINGS
------------------------------------------------------------
- SQL injection in user_query() at line 42
- Hardcoded credentials in config.py
- Missing input validation in API handler

------------------------------------------------------------
RECOMMENDATIONS
------------------------------------------------------------
1. Use parameterized queries
2. Move secrets to environment variables
3. Add input sanitization layer

============================================================

This works across all 10 workflows: security-audit, code-review, perf-audit, doc-gen, test-gen, and more.

Enterprise Doc-Gen: Built for Large Projects

The doc-gen workflow got a major upgrade for enterprise use:

from empathy_os.workflows import DocumentGenerationWorkflow

workflow = DocumentGenerationWorkflow(
    export_path="docs/generated",     # Auto-save to disk
    max_cost=5.0,                     # Stop at $5 (prevent runaway costs)
    chunked_generation=True,          # Handle large codebases
    graceful_degradation=True,        # Partial results on errors
)

result = await workflow.execute(
    source_code=your_large_codebase,
    doc_type="api_reference",
    audience="developers"
)

# Full docs saved to disk automatically
print(f"Saved to: {result.final_output['export_path']}")

What's New:

Feature	What It Does
Auto-scaling tokens	2000 tokens/section, scales to 64k for large projects
Chunked generation	Generates in chunks of 3 sections to avoid truncation
Cost guardrails	Stops at configurable limit ($5 default)
File export	Saves .md and report to disk automatically
Output chunking	Splits large reports for terminal display

Cost Savings: 80-96%

Smart tier routing still saves 80-96% on API costs:

from empathy_llm_toolkit import EmpathyLLM

llm = EmpathyLLM(provider="hybrid", enable_model_routing=True)

# Automatically routes to the right model
await llm.interact(user_id="dev", task_type="summarize")     # → Haiku ($0.25/M)
await llm.interact(user_id="dev", task_type="fix_bug")       # → Sonnet ($3/M)
await llm.interact(user_id="dev", task_type="architecture")  # → Opus ($15/M)

Real savings:

Without routing: $4.05/complex task
With routing: $0.83/complex task
80% saved

Persistent Memory

Your AI remembers across sessions:

llm = EmpathyLLM(provider="anthropic", memory_enabled=True)

# Preference survives across sessions
response = await llm.interact(
    user_id="dev_123",
    user_input="I prefer Python with type hints"
)

Next session—even days later—it remembers.

Quick Start

# Install
pip install empathy-framework==3.3.0

# Configure provider
python -m empathy_os.models.cli provider --set anthropic

# See all commands
empathy cheatsheet

What's in v3.3.0

Formatted Reports — Consistent output across all 10 workflows
Enterprise Doc-Gen — Auto-scaling, cost guardrails, file export
Output Chunking — Large reports split for display
Smart Router — Natural language wizard dispatch
Memory Graph — Cross-wizard knowledge sharing

Resources

What would you build with enterprise-ready AI workflows that cost 80% less?

Give Claude Persistent Memory in 10 Lines of Python

Patrick Roebuck — Fri, 19 Dec 2025 13:18:09 +0000

Every conversation with Claude starts from scratch. Tell it you prefer concise code examples, and next session? It's forgotten.

Here's how to fix that—plus save 80% on API costs.

The Problem

Claude's API is stateless. Each request is independent. For simple Q&A, that's fine. But for:

Development assistants that learn your coding style
Customer support that remembers history
Personal tools that adapt to preferences

...you need memory that persists.

The Solution

from empathy_llm_toolkit import EmpathyLLM

llm = EmpathyLLM(
    provider="anthropic",
    api_key="your-key",
    memory_enabled=True
)

# This preference survives across sessions
response = await llm.interact(
    user_id="dev_123",
    user_input="I prefer Python with type hints, no docstrings"
)

That's it. Next time this user connects—even days later—Claude remembers.

Real-World Example: Debugging Wizard

Here's what persistent memory enables. I built a debugging wizard that correlates current bugs with historical patterns:

from empathy_software_plugin.wizards import MemoryEnhancedDebuggingWizard

wizard = MemoryEnhancedDebuggingWizard()

result = await wizard.analyze({
    "error_message": "TypeError: Cannot read property 'map' of undefined",
    "file_path": "src/components/UserList.tsx"
})

print(result["historical_matches"])
# Shows: "This looks like bug #247 from 3 months ago"
# Suggests: "Add null check: data?.items ?? []"
# Time saved: ~12 minutes

Without persistent memory, every bug starts from zero. With it, your AI assistant remembers every fix and suggests proven solutions.

How It Works

The Empathy Framework stores user context in a memory layer that:

Persists across sessions - Preferences survive restarts
Scopes by user - Each user has isolated memory
Supports projects - Different contexts for different work
Includes privacy controls - Clear memory, forget specific info

Five Levels of Empathy

The framework implements five collaboration levels:

Level	Behavior	Example
1 - Reactive	Standard request-response	Basic Q&A
2 - Informed	Uses stored preferences	Remembers coding style
3 - Proactive	Offers help when stuck	Detects struggle patterns
4 - Anticipatory	Predicts needs	"This will break in 3 days"
5 - Collaborative	Full partnership	Cross-domain learning

# Level 4: Claude predicts and warns
response = await llm.interact(
    user_id="dev_123",
    user_input="Starting a new FastAPI project",
    empathy_level=4
)
# Might warn: "You had async issues last time—here's a pattern that worked"

Privacy Built In

# Clear all memory
await llm.clear_memory(user_id="dev_123")

# Forget specific information
await llm.forget(user_id="dev_123", pattern="email")

Results

On a real codebase (364 debt items, 81 security findings):

Bug correlation: 100% similarity matching with proven fixes
Security noise reduction: 84% (81 → 13 findings after learning)
Tech debt tracking: Trajectory predicts 2x growth in 170 days

NEW in v2.3: Smart Model Routing (80% Cost Savings)

Why pay Opus prices for simple tasks? The new ModelRouter automatically picks the right model:

llm = EmpathyLLM(
    provider="anthropic",
    enable_model_routing=True  # NEW!
)

# Summarization → Haiku ($0.25/M tokens)
await llm.interact(user_id="dev", user_input="Summarize this", task_type="summarize")

# Code generation → Sonnet ($3/M tokens)
await llm.interact(user_id="dev", user_input="Write a function", task_type="generate_code")

# Architecture → Opus ($15/M tokens)
await llm.interact(user_id="dev", user_input="Design the system", task_type="architectural_decision")

Cost comparison on real workload:

Without routing (all Opus): $4.05/complex task
With routing (tiered): $0.83/complex task
Savings: 80%

Get Started

pip install empathy-framework

Resources:

What would you build with an AI that remembers—and costs 80% less? Drop a comment below.