subagent-driven-development（子 Agent 驱动开发）

Subagent-Driven Development: Multi-Agent Coordination

Overview

This skill provides comprehensive patterns for orchestrating multiple specialized agents in complex development workflows. It enables sophisticated coordination between autonomous agents, each with specific expertise and responsibilities.

When to Use This Skill

Use this skill when implementing:

• Multi-agent development workflows with specialized roles
• Complex task orchestration requiring multiple expertise areas
• Autonomous development teams with agent coordination
• Parallel workflow execution with dependency management
• Agent collaboration patterns for complex problems
• Distributed task management across multiple agents
• Quality assurance through specialized reviewers

Quick Start

1. Basic Multi-Agent Workflow

from subagent_framework import WorkflowOrchestrator, Task, AgentType, Priority

# Create orchestrator
orchestrator = WorkflowOrchestrator()

# Register specialized agents
orchestrator.register_agent(ArchitectAgent("architect_001"))
orchestrator.register_agent(DeveloperAgent("developer_001"))
orchestrator.register_agent(TesterAgent("tester_001"))

# Define task workflow
tasks = [
    Task(
        id="design_001",
        title="Design System Architecture",
        agent_type=AgentType.ARCHITECT,
        priority=Priority.HIGH,
        input_data={'requirements': {...}}
    ),
    Task(
        id="implement_001",
        title="Implement Feature",
        agent_type=AgentType.DEVELOPER,
        priority=Priority.HIGH,
        dependencies=["design_001"],  # Depends on architect
        input_data={'specification': {...}}
    ),
    Task(
        id="test_001",
        title="Test Implementation",
        agent_type=AgentType.TESTER,
        priority=Priority.MEDIUM,
        dependencies=["implement_001"],  # Depends on developer
        input_data={'test_type': 'integration'}
    )
]

# Execute workflow
results = await orchestrator.execute_workflow(tasks)

2. Parallel Agent Execution

# Create tasks that can run in parallel
parallel_tasks = [
    Task(
        id="frontend_dev",
        title="Develop Frontend",
        agent_type=AgentType.DEVELOPER,
        dependencies=["design_001"],
        input_data={'component': 'ui'}
    ),
    Task(
        id="backend_dev",
        title="Develop Backend API",
        agent_type=AgentType.DEVELOPER,
        dependencies=["design_001"],
        input_data={'component': 'api'}
    ),
    Task(
        id="database_dev",
        title="Develop Database Schema",
        agent_type=AgentType.DEVELOPER,
        dependencies=["design_001"],
        input_data={'component': 'database'}
    )
]

# These tasks will execute in parallel (same dependency level)
results = await orchestrator.execute_workflow(parallel_tasks)

3. Quality Gate Workflow

# Add quality gate for production-critical code
quality_workflow = [
    Task(id="develop", agent_type=AgentType.DEVELOPER, ...),
    Task(
        id="security_review",
        agent_type=AgentType.SECURITY,
        dependencies=["develop"],
        priority=Priority.CRITICAL,
        input_data={'scan_type': 'comprehensive'}
    ),
    Task(
        id="quality_gate",
        agent_type=AgentType.QUALITY_GATE,
        dependencies=["security_review"],
        priority=Priority.CRITICAL,
        input_data={'approval_required': True}
    )
]

results = await orchestrator.execute_workflow(quality_workflow)

Core Components

Agent Types

See: reference/agent-taxonomy.md for complete agent type specifications

Available Agent Types:

• ARCHITECT - System design and architecture
• DEVELOPER - Code implementation
• TESTER - Testing and quality assurance
• REVIEWER - Code review and analysis
• SECURITY - Security analysis and hardening
• PERFORMANCE - Performance optimization
• QUALITY_GATE - Quality validation and approval
• COORDINATOR - Workflow coordination

Task Management

See: reference/task-management.md for complete task data structures

Key Task Properties:

• agent_type - Which agent handles this task
• priority - Task priority (LOW to EMERGENCY)
• dependencies - Tasks that must complete first
• input_data - Task-specific input parameters
• metadata - Additional context and configuration

Workflow Orchestrator

Central coordination system for multi-agent workflows

Key Methods:

• register_agent() - Add agent to available pool
• execute_workflow() - Execute task workflow
• get_workflow_status() - Monitor progress
• handle_agent_failure() - Error recovery

Common Workflows

Sequential Pipeline

Linear workflow with strict dependencies:

Architect → Developer → Tester → Reviewer → Quality Gate

See: reference/orchestration-patterns.md#sequential-pipeline

Parallel Fan-Out

Multiple agents working independently:

                 ┌─> Developer A (Frontend)
Architect ──────┼─> Developer B (Backend)
                 └─> Developer C (Database)
                           ↓
                      [Integration]

See: reference/orchestration-patterns.md#parallel-fan-out

Iterative Refinement

Progressive improvement through feedback loops:

Developer → Reviewer → [Feedback] → Developer → Reviewer → [Approved]

See: reference/orchestration-patterns.md#iterative-refinement

Orchestration Patterns

Complete patterns: reference/orchestration-patterns.md

Pattern Selection Guide

Error Handling

Retry Strategy

task = Task(
    id="task_001",
    title="Implement Feature",
    agent_type=AgentType.DEVELOPER,
    priority=Priority.HIGH,
    max_retries=3,  # Retry up to 3 times
    input_data={...}
)

Fallback Agents

# Register multiple developer agents for fallback
orchestrator.register_agent(DeveloperAgent("dev_primary"))
orchestrator.register_agent(DeveloperAgent("dev_fallback_1"))
orchestrator.register_agent(DeveloperAgent("dev_fallback_2"))

# Orchestrator automatically uses fallback if primary fails

See: reference/orchestration-patterns.md#error-handling-patterns

Monitoring and Status

Get Workflow Status

status = orchestrator.get_workflow_status(workflow_id)

print(f"Progress: {status['progress']['percentage']}%")
print(f"Completed: {status['progress']['completed']}/{status['progress']['total']}")
print(f"Active Agents: {status['active_agents']}")

Monitor Agent Health

for agent_id, agent_status in status['agent_details'].items():
    print(f"{agent_id}: {agent_status['status']}")
    if agent_status['current_task']:
        print(f"  Working on: {agent_status['current_task']}")

Real Estate Platform Example

Complete example: examples/real-estate-workflow.py

async def create_property_matching_workflow():
    """Multi-agent workflow for property matching feature."""
    orchestrator = WorkflowOrchestrator()

    # Register agents
    orchestrator.register_agent(ArchitectAgent("architect_001"))
    orchestrator.register_agent(DeveloperAgent("developer_001"))
    orchestrator.register_agent(DeveloperAgent("developer_002"))
    orchestrator.register_agent(TesterAgent("tester_001"))
    orchestrator.register_agent(QualityGateAgent("quality_gate_001"))

    # Define workflow
    workflow = [
        # Phase 1: Architecture
        Task(
            id="arch_001",
            title="Design Property Matching System",
            agent_type=AgentType.ARCHITECT,
            priority=Priority.HIGH,
            input_data={'requirements': {'ai_integration': True}}
        ),

        # Phase 2: Parallel Development
        Task(
            id="api_dev",
            title="Develop Matching API",
            agent_type=AgentType.DEVELOPER,
            dependencies=["arch_001"],
            input_data={'framework': 'FastAPI'}
        ),
        Task(
            id="model_dev",
            title="Develop Data Models",
            agent_type=AgentType.DEVELOPER,
            dependencies=["arch_001"],
            input_data={'orm': 'SQLAlchemy'}
        ),

        # Phase 3: Testing
        Task(
            id="integration_test",
            title="Integration Testing",
            agent_type=AgentType.TESTER,
            dependencies=["api_dev", "model_dev"],
            input_data={'coverage_target': 0.80}
        ),

        # Phase 4: Quality Gate
        Task(
            id="quality_gate",
            title="Production Approval",
            agent_type=AgentType.QUALITY_GATE,
            dependencies=["integration_test"],
            priority=Priority.CRITICAL
        )
    ]

    # Execute
    results = await orchestrator.execute_workflow(workflow)
    return results

Best Practices

1. Clear Agent Responsibilities: Each agent should have well-defined, non-overlapping responsibilities
2. Proper Dependency Management: Ensure task dependencies are correctly specified and enforced
3. Error Handling: Implement robust error handling and retry mechanisms
4. Status Monitoring: Provide comprehensive status monitoring and reporting
5. Resource Management: Prevent resource contention and ensure efficient agent utilization
6. Quality Gates: Implement quality validation at appropriate workflow stages
7. Scalability: Design for horizontal scaling of agent instances

Advanced Usage

For advanced multi-agent orchestration scenarios, see:

• reference/agent-taxonomy.md - Complete agent type specifications
• reference/task-management.md - Task and workflow state management
• reference/orchestration-patterns.md - Advanced coordination patterns
• examples/real-estate-workflow.py - Real Estate platform workflows
• examples/emergency-workflow.py - Emergency fast-track patterns
• scripts/workflow-monitor.py - Monitoring and observability tools

Troubleshooting

Agents Not Executing Tasks

Cause: No registered agents matching task's agent_type Solution: Ensure agents are registered before workflow execution

Circular Dependencies

Cause: Task A depends on Task B which depends on Task A Solution: Use dependency resolver to validate task graph before execution

Workflow Stuck

Cause: Agent failed but retry limit not reached Solution: Check agent logs, increase retry limit or add fallback agents

Quick Reference

Version: 2.0.0 (Token-Optimized) Original: 1,395 lines Optimized: ~400 lines (71% reduction) Full Documentation: See reference/ directory