[TRACK3_005] Capstone: Autonomous Multi-Agent System
[TRACK3_005] Capstone: Autonomous Multi-Agent System
📚 Capstone: Autonomous Multi-Agent System
[VIDEO-015] Track 3 Capstone: Build an Autonomous Multi-Agent System
Track: 3 - Multi-Agent Systems Module: 5 (Capstone) Duration: 18 minutes Level requirement: 42 XP reward: 600 XP (+ 3000 XP for completion)
---
Scene 1: The Ultimate Multi-Agent Challenge (0:00-2:00)
[Visual]: Epic capstone badge with multiple agent avatars [Animation]: Agents forming into a coordinated swarm
[Audio/Script]:
"You've learned to orchestrate agents. Build communication systems. Create specialists.>
Now it's time to build something that can work autonomously.>
Your Track 3 Capstone: An Autonomous Project Assistant>
This system will:
- Accept high-level project goals
- Break them into tasks automatically
- Assign tasks to specialist agents
- Coordinate execution
- Handle failures gracefully
- Report results comprehensively>
All without human intervention.>
This is real-world AI automation. Let's build it."
[Lower third]: "Track 3 Capstone | Levels 42-45"
---
Scene 2: System Architecture (2:00-4:30)
[Visual]: Full system architecture diagram [Animation]: Components connecting and data flowing
[Audio/Script]:
"Your autonomous system has five layers:"
[Diagram]:
┌─────────────────────────────────────────────────────────────┐
│ PROJECT ASSISTANT │
├─────────────────────────────────────────────────────────────┤
│ Layer 1: INPUT PROCESSOR │
│ - Parse project requirements │
│ - Extract goals and constraints │
│ - Validate inputs │
├─────────────────────────────────────────────────────────────┤
│ Layer 2: STRATEGIC PLANNER │
│ - Decompose goals into phases │
│ - Create dependency graph │
│ - Estimate resources │
├─────────────────────────────────────────────────────────────┤
│ Layer 3: TASK ORCHESTRATOR │
│ - Assign tasks to agents │
│ - Manage execution order │
│ - Handle parallelization │
├─────────────────────────────────────────────────────────────┤
│ Layer 4: SPECIALIST POOL │
│ - Research Agent │
│ - Code Agent │
│ - Writer Agent │
│ - Review Agent │
├─────────────────────────────────────────────────────────────┤
│ Layer 5: MONITORING & REPORTING │
│ - Progress tracking │
│ - Error handling │
│ - Final synthesis │
└─────────────────────────────────────────────────────────────┘---
Scene 3: Core Framework (4:30-7:00)
[Visual]: Framework code structure [Animation]: Classes being assembled
[Audio/Script]:
"Let's build the core framework."
[Demo - Core Classes]:
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional
from enum import Enum
from datetime import datetime
import asyncio
import uuidclass TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
BLOCKED = "blocked"
class TaskPriority(Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
CRITICAL = 4
@dataclass
class ProjectGoal:
"""A high-level project goal"""
goal_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
description: str = ""
constraints: List[str] = field(default_factory=list)
deadline: Optional[datetime] = None
success_criteria: List[str] = field(default_factory=list)
@dataclass
class AgentTask:
"""A task assigned to an agent"""
task_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
description: str = ""
assigned_agent: str = ""
status: TaskStatus = TaskStatus.PENDING
priority: TaskPriority = TaskPriority.MEDIUM
dependencies: List[str] = field(default_factory=list)
result: Optional[Dict] = None
error: Optional[str] = None
created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
@dataclass
class ProjectPhase:
"""A phase of the project"""
phase_id: str
name: str
description: str
tasks: List[AgentTask] = field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
@dataclass
class ExecutionPlan:
"""Complete execution plan for a project"""
plan_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
goal: ProjectGoal = None
phases: List[ProjectPhase] = field(default_factory=list)
estimated_duration_minutes: int = 0
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class ExecutionReport:
"""Final report of project execution"""
plan_id: str
success: bool
phases_completed: int
phases_total: int
tasks_completed: int
tasks_failed: int
duration_minutes: float
outputs: Dict[str, Any]
errors: List[str]
recommendations: List[str]
---
Scene 4: Input Processor (7:00-8:30)
[Visual]: Input parsing flow [Animation]: Raw input becoming structured goal
[Audio/Script]:
"Layer 1: The Input Processor turns natural language into structured goals."
[Demo - Input Processor]:
class InputProcessor:
"""Processes and validates project inputs""" async def process(self, raw_input: str) -> ProjectGoal:
"""Convert raw input to structured ProjectGoal"""
# Extract the main goal
goal_description = await self._extract_goal(raw_input)
# Extract constraints
constraints = await self._extract_constraints(raw_input)
# Extract success criteria
success_criteria = await self._extract_success_criteria(raw_input)
# Validate
self._validate(goal_description, constraints, success_criteria)
return ProjectGoal(
description=goal_description,
constraints=constraints,
success_criteria=success_criteria
)
async def _extract_goal(self, text: str) -> str:
"""Extract the main goal from input"""
# In production, use Claude to extract
# For now, simple heuristics
lines = text.strip().split('\n')
return lines[0] if lines else text[:200]
async def _extract_constraints(self, text: str) -> List[str]:
"""Extract constraints from input"""
constraints = []
text_lower = text.lower()
# Look for constraint keywords
constraint_patterns = [
("must", "requirement"),
("should", "preference"),
("cannot", "restriction"),
("within", "resource"),
("by", "deadline"),
]
for pattern, constraint_type in constraint_patterns:
if pattern in text_lower:
# Extract the relevant sentence
for sentence in text.split('.'):
if pattern in sentence.lower():
constraints.append(sentence.strip())
return constraints
async def _extract_success_criteria(self, text: str) -> List[str]:
"""Extract success criteria"""
criteria = []
text_lower = text.lower()
# Look for success indicators
if "success" in text_lower or "complete" in text_lower:
for sentence in text.split('.'):
if any(w in sentence.lower() for w in ["success", "complete", "deliver", "achieve"]):
criteria.append(sentence.strip())
# Default criteria if none found
if not criteria:
criteria = ["Task completed without errors", "All deliverables produced"]
return criteria
def _validate(self, goal: str, constraints: List, criteria: List):
"""Validate the processed input"""
if not goal or len(goal) < 10:
raise ValueError("Goal description too short")
if len(goal) > 2000:
raise ValueError("Goal description too long")
---
Scene 5: Strategic Planner (8:30-10:30)
[Visual]: Goal decomposition tree [Animation]: Goal breaking into phases and tasks
[Audio/Script]:
"Layer 2: The Strategic Planner decomposes goals into executable phases."
[Demo - Strategic Planner]:
class StrategicPlanner:
"""Creates execution plans from goals""" def __init__(self, agent_registry: Dict[str, List[str]]):
self.agent_capabilities = agent_registry
async def create_plan(self, goal: ProjectGoal) -> ExecutionPlan:
"""Create a full execution plan for a goal"""
# Analyze goal to determine required phases
phases = await self._decompose_goal(goal)
# Create tasks for each phase
for phase in phases:
phase.tasks = await self._create_phase_tasks(phase, goal)
# Establish dependencies
self._establish_dependencies(phases)
# Estimate duration
duration = self._estimate_duration(phases)
return ExecutionPlan(
goal=goal,
phases=phases,
estimated_duration_minutes=duration
)
async def _decompose_goal(self, goal: ProjectGoal) -> List[ProjectPhase]:
"""Decompose goal into phases"""
# Standard phases for most projects
phase_templates = [
("research", "Research & Discovery", "Gather information and understand requirements"),
("design", "Design & Planning", "Design the solution architecture"),
("implement", "Implementation", "Build the solution"),
("test", "Testing & Validation", "Verify the solution works"),
("document", "Documentation", "Create documentation and reports")
]
phases = []
goal_lower = goal.description.lower()
for phase_id, name, description in phase_templates:
# Include phase if relevant to goal
if self._phase_relevant(phase_id, goal_lower):
phases.append(ProjectPhase(
phase_id=phase_id,
name=name,
description=description
))
return phases
def _phase_relevant(self, phase_id: str, goal: str) -> bool:
"""Determine if a phase is relevant to the goal"""
# Research is always relevant
if phase_id == "research":
return True
# Implement if building something
if phase_id == "implement" and any(w in goal for w in ["build", "create", "develop", "make"]):
return True
# Test if implementing
if phase_id == "test" and any(w in goal for w in ["build", "create", "develop", "code"]):
return True
# Document if report needed
if phase_id == "document":
return True
# Design for complex projects
if phase_id == "design" and any(w in goal for w in ["system", "architecture", "complex"]):
return True
return False
async def _create_phase_tasks(self, phase: ProjectPhase, goal: ProjectGoal) -> List[AgentTask]:
"""Create tasks for a phase"""
tasks = []
if phase.phase_id == "research":
tasks.append(AgentTask(
description=f"Research: {goal.description}",
assigned_agent="research_agent",
priority=TaskPriority.HIGH
))
elif phase.phase_id == "design":
tasks.append(AgentTask(
description="Create solution design and architecture",
assigned_agent="architect_agent",
priority=TaskPriority.HIGH
))
elif phase.phase_id == "implement":
tasks.append(AgentTask(
description="Implement the solution based on design",
assigned_agent="code_agent",
priority=TaskPriority.HIGH
))
elif phase.phase_id == "test":
tasks.append(AgentTask(
description="Test and validate the implementation",
assigned_agent="test_agent",
priority=TaskPriority.MEDIUM
))
elif phase.phase_id == "document":
tasks.append(AgentTask(
description="Create documentation and final report",
assigned_agent="writer_agent",
priority=TaskPriority.MEDIUM
))
return tasks
def _establish_dependencies(self, phases: List[ProjectPhase]):
"""Establish task dependencies across phases"""
previous_phase_tasks = []
for phase in phases:
for task in phase.tasks:
# Each task depends on previous phase completion
task.dependencies = [t.task_id for t in previous_phase_tasks]
previous_phase_tasks = phase.tasks
def _estimate_duration(self, phases: List[ProjectPhase]) -> int:
"""Estimate total duration in minutes"""
return sum(len(phase.tasks) * 5 for phase in phases) # 5 min per task estimate
---
Scene 6: Task Orchestrator (10:30-13:00)
[Visual]: Orchestrator managing task flow [Animation]: Tasks being assigned and executed
[Audio/Script]:
"Layer 3: The Task Orchestrator executes the plan."
[Demo - Task Orchestrator]:
class TaskOrchestrator:
"""Orchestrates task execution across agents""" def __init__(self, agents: Dict[str, Any]):
self.agents = agents
self.task_results: Dict[str, Any] = {}
self.failed_tasks: List[str] = []
async def execute_plan(self, plan: ExecutionPlan) -> ExecutionReport:
"""Execute a complete plan"""
start_time = datetime.now()
phases_completed = 0
for phase in plan.phases:
print(f"\n{'='*50}")
print(f"Starting Phase: {phase.name}")
print(f"{'='*50}")
phase.status = TaskStatus.IN_PROGRESS
phase_success = await self._execute_phase(phase)
if phase_success:
phase.status = TaskStatus.COMPLETED
phases_completed += 1
else:
phase.status = TaskStatus.FAILED
print(f"Phase {phase.name} failed. Attempting recovery...")
# Try to recover
recovered = await self._attempt_recovery(phase)
if recovered:
phase.status = TaskStatus.COMPLETED
phases_completed += 1
else:
# Check if we can continue
if not self._can_continue(phase, plan):
print("Critical phase failed. Stopping execution.")
break
# Calculate duration
duration = (datetime.now() - start_time).total_seconds() / 60
# Generate report
return self._generate_report(plan, phases_completed, duration)
async def _execute_phase(self, phase: ProjectPhase) -> bool:
"""Execute all tasks in a phase"""
# Group tasks by dependencies for parallel execution
task_groups = self._group_by_dependencies(phase.tasks)
for group in task_groups:
# Execute group in parallel
results = await asyncio.gather(
*[self._execute_task(task) for task in group],
return_exceptions=True
)
# Check for failures
for task, result in zip(group, results):
if isinstance(result, Exception):
task.status = TaskStatus.FAILED
task.error = str(result)
self.failed_tasks.append(task.task_id)
else:
task.status = TaskStatus.COMPLETED
task.result = result
self.task_results[task.task_id] = result
# Phase succeeds if no critical failures
return all(t.status == TaskStatus.COMPLETED for t in phase.tasks)
async def _execute_task(self, task: AgentTask) -> Dict[str, Any]:
"""Execute a single task"""
print(f" Executing: {task.description[:50]}...")
task.status = TaskStatus.IN_PROGRESS
task.started_at = datetime.now()
# Get the agent
agent = self.agents.get(task.assigned_agent)
if not agent:
raise ValueError(f"Agent not found: {task.assigned_agent}")
# Build context from dependency results
context = {
"dependencies": {
dep_id: self.task_results.get(dep_id)
for dep_id in task.dependencies
}
}
# Execute
try:
result = await agent.execute(task.description, context)
task.completed_at = datetime.now()
print(f" Completed: {task.description[:50]}")
return result
except Exception as e:
task.error = str(e)
print(f" Failed: {task.description[:50]} - {e}")
raise
def _group_by_dependencies(self, tasks: List[AgentTask]) -> List[List[AgentTask]]:
"""Group tasks by dependencies for parallel execution"""
groups = []
remaining = list(tasks)
completed = set()
while remaining:
# Find tasks with satisfied dependencies
ready = [
t for t in remaining
if all(d in completed for d in t.dependencies)
]
if not ready:
# Deadlock - force progress
ready = [remaining[0]]
groups.append(ready)
for task in ready:
remaining.remove(task)
completed.add(task.task_id)
return groups
async def _attempt_recovery(self, phase: ProjectPhase) -> bool:
"""Attempt to recover from phase failure"""
failed_tasks = [t for t in phase.tasks if t.status == TaskStatus.FAILED]
for task in failed_tasks:
# Try up to 2 retries
for attempt in range(2):
print(f" Retry {attempt + 1} for: {task.description[:30]}...")
try:
result = await self._execute_task(task)
task.result = result
task.status = TaskStatus.COMPLETED
task.error = None
break
except Exception:
continue
return all(t.status == TaskStatus.COMPLETED for t in phase.tasks)
def _can_continue(self, failed_phase: ProjectPhase, plan: ExecutionPlan) -> bool:
"""Determine if execution can continue after a failure"""
# Research and design failures are critical
if failed_phase.phase_id in ["research", "design"]:
return False
# Other phases can potentially be skipped
return True
def _generate_report(self, plan: ExecutionPlan, phases_completed: int, duration: float) -> ExecutionReport:
"""Generate final execution report"""
all_tasks = [t for p in plan.phases for t in p.tasks]
completed_tasks = [t for t in all_tasks if t.status == TaskStatus.COMPLETED]
failed_tasks = [t for t in all_tasks if t.status == TaskStatus.FAILED]
return ExecutionReport(
plan_id=plan.plan_id,
success=len(failed_tasks) == 0,
phases_completed=phases_completed,
phases_total=len(plan.phases),
tasks_completed=len(completed_tasks),
tasks_failed=len(failed_tasks),
duration_minutes=duration,
outputs=self.task_results,
errors=[t.error for t in failed_tasks if t.error],
recommendations=self._generate_recommendations(failed_tasks)
)
def _generate_recommendations(self, failed_tasks: List[AgentTask]) -> List[str]:
"""Generate recommendations based on failures"""
recommendations = []
if failed_tasks:
recommendations.append(f"{len(failed_tasks)} tasks failed - review error logs")
return recommendations
---
Scene 7: Putting It All Together (13:00-16:00)
[Visual]: Complete system running [Animation]: End-to-end execution flow
[Audio/Script]:
"Now let's assemble the complete autonomous system."
[Demo - Complete System]:
class AutonomousProjectAssistant:
"""Complete autonomous multi-agent project assistant""" def __init__(self):
# Initialize specialists
self.agents = {
"research_agent": ResearchAgent(),
"code_agent": CodeAgent(),
"writer_agent": WriterAgent(),
"architect_agent": ArchitectAgent(),
"test_agent": TestAgent()
}
# Initialize layers
self.input_processor = InputProcessor()
self.planner = StrategicPlanner(
{name: agent.capabilities for name, agent in self.agents.items()}
)
self.orchestrator = TaskOrchestrator(self.agents)
async def execute_project(self, project_description: str) -> ExecutionReport:
"""Execute a complete project autonomously"""
print("\n" + "="*60)
print("AUTONOMOUS PROJECT ASSISTANT")
print("="*60)
# Layer 1: Process input
print("\n[1/4] Processing project requirements...")
goal = await self.input_processor.process(project_description)
print(f"Goal: {goal.description[:100]}...")
print(f"Constraints: {len(goal.constraints)}")
print(f"Success criteria: {len(goal.success_criteria)}")
# Layer 2: Create plan
print("\n[2/4] Creating execution plan...")
plan = await self.planner.create_plan(goal)
print(f"Plan has {len(plan.phases)} phases")
for phase in plan.phases:
print(f" - {phase.name}: {len(phase.tasks)} tasks")
# Layer 3: Execute plan
print("\n[3/4] Executing plan...")
report = await self.orchestrator.execute_plan(plan)
# Layer 4: Generate final output
print("\n[4/4] Generating final report...")
final_output = self._synthesize_output(report)
# Print summary
self._print_summary(report, final_output)
return report
def _synthesize_output(self, report: ExecutionReport) -> str:
"""Synthesize final output from all results"""
parts = ["# Project Execution Results\n"]
if report.success:
parts.append("Status: SUCCESS\n")
else:
parts.append("Status: PARTIAL SUCCESS\n")
parts.append(f"\n## Summary")
parts.append(f"- Phases completed: {report.phases_completed}/{report.phases_total}")
parts.append(f"- Tasks completed: {report.tasks_completed}")
parts.append(f"- Tasks failed: {report.tasks_failed}")
parts.append(f"- Duration: {report.duration_minutes:.1f} minutes")
if report.outputs:
parts.append("\n## Outputs")
for task_id, output in report.outputs.items():
if isinstance(output, dict) and "summary" in output:
parts.append(f"\n### {task_id}")
parts.append(output["summary"][:500])
if report.errors:
parts.append("\n## Errors")
for error in report.errors:
parts.append(f"- {error}")
if report.recommendations:
parts.append("\n## Recommendations")
for rec in report.recommendations:
parts.append(f"- {rec}")
return "\n".join(parts)
def _print_summary(self, report: ExecutionReport, output: str):
"""Print execution summary"""
print("\n" + "="*60)
print("EXECUTION COMPLETE")
print("="*60)
print(f"Success: {'YES' if report.success else 'PARTIAL'}")
print(f"Duration: {report.duration_minutes:.1f} minutes")
print(f"Tasks: {report.tasks_completed} completed, {report.tasks_failed} failed")
print("\nOutput preview:")
print(output[:500] + "..." if len(output) > 500 else output)
Usage
async def main():
assistant = AutonomousProjectAssistant() project = """
Create a comprehensive analysis of current AI agent frameworks.
Requirements:
- Research top 5 frameworks (LangChain, AutoGPT, CrewAI, etc.)
- Compare features, performance, and use cases
- Create code examples for each
- Write a technical report with recommendations
- Include architecture diagrams
Must be completed with working examples.
Success when: Report is comprehensive and examples run successfully.
"""
report = await assistant.execute_project(project)
# Save report
with open("project_report.md", "w") as f:
f.write(assistant._synthesize_output(report))
asyncio.run(main())
---
Scene 8: Submission & Celebration (16:00-18:00)
[Visual]: Achievement unlock sequence [Animation]: XP climbing, badge appearing, Track 4 unlocking
[Audio/Script]:
"Your submission requirements:>
| Component | Points |
|-----------|--------|
| Input Processor functional | 400 |
| Strategic Planner with phases | 500 |
| Task Orchestrator with parallel execution | 600 |
| 4+ Specialist Agents | 600 |
| Error handling & recovery | 400 |
| Final report synthesis | 300 |
| Bonus: Monitoring dashboard | 200 |>
Total possible: 3000 XP>
Completing this capstone:
- Awards the 'Multi-Agent Architect' badge
- Unlocks Track 4: Production Mastery
- Proves you can build autonomous AI systems>
You started Track 3 wondering how to coordinate agents.>
Now you've built a system that coordinates itself.>
Track 4 is the final frontier. Taking these systems to production.>
Deployment. Scaling. Monitoring. Enterprise-grade AI.>
You're almost there."
[Closing animation]: Badge unlock, confetti, level up to 45, Track 4 preview
---
Post-Video Challenge
Challenge ID: TRACK3_CAPSTONE Type: Project Submission Time Limit: None (complete at your pace)
Requirements:
1. Input Processor (400 XP)
- Process natural language project descriptions
- Extract goals, constraints, success criteria
- Input validation
- Decompose goals into phases
- Create tasks for each phase
- Establish dependencies
- Estimate durations
- Execute phases sequentially
- Execute independent tasks in parallel
- Handle failures with retries
- Generate execution report
- ResearchAgent - information gathering
- CodeAgent - development
- WriterAgent - content creation
- One additional specialist of your choice
- Graceful failure handling
- Recovery attempts
- Comprehensive error reporting
- Combine all outputs
- Summary statistics
- Recommendations
- Real-time monitoring dashboard (+200)
- Agent performance metrics (+150)
- Configuration file support (+150)
- All code files in your workspace
- Run:
/validate track3-capstone - Demo showing autonomous execution of a complex project
- Base XP: 600
- Challenge XP: Up to 3000
- Achievement: "Multi-Agent Architect" (Track 3 Complete)
- Unlock: Track 4 - Production Mastery
SEO Metadata
Alt-text: Track 3 Capstone - build a complete autonomous multi-agent system with strategic planning, task orchestration, and specialist agents working together autonomously.
Tags: autonomous AI, multi-agent capstone, AI orchestration, autonomous systems, agent coordination, AI project assistant
Keywords: autonomous multi-agent system, ai capstone project, build autonomous ai, multi-agent orchestration, ai project assistant, autonomous agent coordination