Why Multi-Agent Systems

1# Problem: Single agent trying to analyze entire codebase
2class SingleAgentCodeAnalyzer:
3    """Hits context limits quickly with large codebases."""
4
5    def __init__(self, llm_client):
6        self.llm = llm_client
7        self.max_tokens = 128000  # Context window limit
8
9    def analyze_codebase(self, codebase_path: str):
10        # Attempt to load entire codebase
11        all_files = self.load_all_files(codebase_path)
12        total_tokens = self.count_tokens(all_files)
13
14        if total_tokens > self.max_tokens:
15            # Context overflow!
16            # Options: truncate (lose info) or summarize (lose detail)
17            print(f"Warning: Codebase has [total_tokens] tokens")
18            print(f"Context limit: [self.max_tokens] tokens")
19            return self.partial_analysis(all_files)  # Compromised result
20
21        return self.full_analysis(all_files)
22
23    def load_all_files(self, path: str):
24        """Load all source files from directory."""
25        files = {}
26        import os
27        for root, _, filenames in os.walk(path):
28            for filename in filenames:
29                if filename.endswith(('.py', '.js', '.ts')):
30                    filepath = os.path.join(root, filename)
31                    with open(filepath, 'r') as f:
32                        files[filepath] = f.read()
33        return files
34
35    def count_tokens(self, files: dict) -> int:
36        """Estimate token count."""
37        return sum(len(content) // 4 for content in files.values())
38
39    def partial_analysis(self, files: dict):
40        """Compromised analysis with truncated context."""
41        # Must choose what to include, what to exclude
42        return {"status": "partial", "warning": "context overflow"}
43
44    def full_analysis(self, files: dict):
45        """Full analysis when context permits."""
46        return {"status": "complete"}

1# Problem: Single agent handling diverse expertise areas
2class AllPurposeAgent:
3    """One agent trying to handle everything."""
4
5    def __init__(self):
6        self.system_prompt = """
7        You are an expert in:
8        - Python, JavaScript, TypeScript, Rust, Go, Java, C++
9        - Frontend (React, Vue, Angular, Svelte)
10        - Backend (FastAPI, Django, Express, Actix)
11        - Databases (PostgreSQL, MongoDB, Redis, Cassandra)
12        - DevOps (Docker, Kubernetes, Terraform, AWS, GCP, Azure)
13        - Security (OWASP, authentication, encryption)
14        - Machine Learning (PyTorch, TensorFlow, scikit-learn)
15        - System Design (microservices, distributed systems)
16
17        Handle all user requests with expert-level knowledge.
18        """
19        # Problem: Diluted expertise across too many domains
20        # Result: Mediocre performance in specialized tasks
21
22# Compare to specialized human teams:
23# - Security Expert: Deep knowledge of vulnerabilities, attack vectors
24# - ML Engineer: Deep knowledge of model architectures, training
25# - DevOps Engineer: Deep knowledge of infrastructure, deployments
26# Each specialist excels in their domain

1# Problem: Complex multi-step task for single agent
2class SingleAgentProjectBuilder:
3    """Single agent attempting complex project creation."""
4
5    async def build_full_stack_app(self, requirements: str):
6        """
7        Required subtasks (in dependency order):
8        1. Analyze requirements
9        2. Design database schema
10        3. Create API specifications
11        4. Implement backend (depends on 2, 3)
12        5. Implement frontend (depends on 3, 4)
13        6. Write tests (depends on 4, 5)
14        7. Configure deployment (depends on 4, 5, 6)
15        8. Write documentation (depends on all above)
16
17        Single agent challenges:
18        - Must hold entire context in working memory
19        - Errors propagate without checks
20        - No parallel execution possible
21        - Difficult to resume if interrupted
22        """
23
24        # Sequential execution with accumulated context
25        analysis = await self.analyze(requirements)
26        schema = await self.design_database(analysis)
27        api_spec = await self.create_api_spec(analysis, schema)
28        backend = await self.implement_backend(schema, api_spec)
29        frontend = await self.implement_frontend(api_spec, backend)
30        tests = await self.write_tests(backend, frontend)
31        deployment = await self.configure_deployment(backend, frontend)
32        docs = await self.write_documentation(
33            analysis, schema, api_spec, backend,
34            frontend, tests, deployment
35        )
36
37        # Context grows with each step
38        # Risk of hallucination increases
39        # No independent verification
40        return {
41            "analysis": analysis,
42            "schema": schema,
43            "api_spec": api_spec,
44            "backend": backend,
45            "frontend": frontend,
46            "tests": tests,
47            "deployment": deployment,
48            "docs": docs
49        }

1# Multi-agent approach: Specialized experts
2from dataclasses import dataclass
3from typing import Protocol
4from abc import abstractmethod
5
6
7class SpecializedAgent(Protocol):
8    """Protocol for specialized agents."""
9
10    @property
11    def expertise(self) -> str:
12        """Agent's area of expertise."""
13        ...
14
15    @abstractmethod
16    async def analyze(self, context: dict) -> dict:
17        """Perform specialized analysis."""
18        ...
19
20
21@dataclass
22class SecurityAgent:
23    """Agent specialized in security analysis."""
24    expertise: str = "security"
25
26    system_prompt: str = """
27    You are a security expert specializing in:
28    - OWASP Top 10 vulnerabilities
29    - Authentication and authorization patterns
30    - Encryption and key management
31    - Input validation and sanitization
32    - Security headers and CORS
33    - Rate limiting and DDoS protection
34
35    Analyze code and architectures for security issues.
36    Provide specific, actionable security recommendations.
37    """
38
39    async def analyze(self, code: str) -> dict:
40        """Deep security analysis with expert knowledge."""
41        return {
42            "vulnerabilities": await self.find_vulnerabilities(code),
43            "recommendations": await self.generate_recommendations(code),
44            "severity_scores": await self.calculate_severities(code),
45        }
46
47    async def find_vulnerabilities(self, code: str) -> list:
48        # Focused security vulnerability detection
49        return []
50
51    async def generate_recommendations(self, code: str) -> list:
52        # Expert security recommendations
53        return []
54
55    async def calculate_severities(self, code: str) -> dict:
56        # CVSS-style severity scoring
57        return {}
58
59
60@dataclass
61class PerformanceAgent:
62    """Agent specialized in performance optimization."""
63    expertise: str = "performance"
64
65    system_prompt: str = """
66    You are a performance optimization expert specializing in:
67    - Time complexity analysis
68    - Memory optimization patterns
69    - Database query optimization
70    - Caching strategies
71    - Async/concurrent programming
72    - Profiling and benchmarking
73
74    Analyze code for performance bottlenecks.
75    Provide specific optimization recommendations.
76    """
77
78    async def analyze(self, code: str) -> dict:
79        """Deep performance analysis with expert knowledge."""
80        return {
81            "bottlenecks": await self.find_bottlenecks(code),
82            "optimizations": await self.suggest_optimizations(code),
83            "complexity_analysis": await self.analyze_complexity(code),
84        }
85
86    async def find_bottlenecks(self, code: str) -> list:
87        return []
88
89    async def suggest_optimizations(self, code: str) -> list:
90        return []
91
92    async def analyze_complexity(self, code: str) -> dict:
93        return {}
94
95
96class MultiAgentCodeReviewer:
97    """Orchestrates multiple specialized agents."""
98
99    def __init__(self):
100        self.agents = [
101            SecurityAgent(),
102            PerformanceAgent(),
103            # Additional specialized agents...
104        ]
105
106    async def comprehensive_review(self, code: str) -> dict:
107        """Each agent provides deep expertise in their domain."""
108        import asyncio
109
110        # Parallel specialized analysis
111        results = await asyncio.gather(*[
112            agent.analyze(code) for agent in self.agents
113        ])
114
115        return {
116            agent.expertise: result
117            for agent, result in zip(self.agents, results)
118        }

1import asyncio
2from dataclasses import dataclass, field
3from typing import Callable, Any
4import time
5
6
7@dataclass
8class TaskResult:
9    """Result from a task execution."""
10    agent_id: str
11    result: Any
12    duration: float
13
14
15@dataclass
16class ParallelAgentSystem:
17    """System for parallel agent execution."""
18    agents: dict = field(default_factory=dict)
19
20    def register_agent(self, agent_id: str, agent: Any):
21        """Register an agent in the system."""
22        self.agents[agent_id] = agent
23
24    async def execute_parallel(
25        self,
26        tasks: list[tuple[str, str, dict]]  # (agent_id, method, params)
27    ) -> list[TaskResult]:
28        """Execute tasks across agents in parallel."""
29
30        async def run_task(agent_id: str, method: str, params: dict):
31            start = time.time()
32            agent = self.agents[agent_id]
33            result = await getattr(agent, method)(**params)
34            duration = time.time() - start
35            return TaskResult(agent_id, result, duration)
36
37        results = await asyncio.gather(*[
38            run_task(agent_id, method, params)
39            for agent_id, method, params in tasks
40        ])
41
42        return results
43
44
45# Example: Parallel code analysis
46async def parallel_analysis_demo():
47    system = ParallelAgentSystem()
48
49    # Register specialized agents
50    system.register_agent("security", SecurityAgent())
51    system.register_agent("performance", PerformanceAgent())
52
53    code_to_analyze = "def process(data): return data"
54
55    # Define parallel tasks
56    tasks = [
57        ("security", "analyze", {"code": code_to_analyze}),
58        ("performance", "analyze", {"code": code_to_analyze}),
59    ]
60
61    # Execute in parallel - 2x faster than sequential
62    start = time.time()
63    results = await system.execute_parallel(tasks)
64    total_time = time.time() - start
65
66    print(f"Total parallel execution: [total_time:.2f]s")
67    for result in results:
68        print(f"  [result.agent_id]: [result.duration:.2f]s")
69
70    # Compare to sequential execution
71    sequential_time = sum(r.duration for r in results)
72    speedup = sequential_time / total_time
73    print(f"Speedup: [speedup:.1f]x")

1from dataclasses import dataclass, field
2from typing import Optional
3from enum import Enum
4
5
6class AgentStatus(Enum):
7    RUNNING = "running"
8    COMPLETED = "completed"
9    FAILED = "failed"
10    RETRYING = "retrying"
11
12
13@dataclass
14class AgentResult:
15    """Result from agent execution with status."""
16    status: AgentStatus
17    data: Optional[dict] = None
18    error: Optional[str] = None
19    retries: int = 0
20
21
22@dataclass
23class ResilientAgentSystem:
24    """Multi-agent system with error isolation."""
25
26    max_retries: int = 3
27    agents: dict = field(default_factory=dict)
28    results: dict = field(default_factory=dict)
29
30    async def execute_with_isolation(
31        self,
32        agent_id: str,
33        method: str,
34        params: dict
35    ) -> AgentResult:
36        """Execute agent task with error isolation and retry."""
37
38        retries = 0
39        last_error = None
40
41        while retries < self.max_retries:
42            try:
43                agent = self.agents[agent_id]
44                result = await getattr(agent, method)(**params)
45                return AgentResult(
46                    status=AgentStatus.COMPLETED,
47                    data=result,
48                    retries=retries
49                )
50            except Exception as e:
51                retries += 1
52                last_error = str(e)
53                print(f"Agent [agent_id] failed (attempt [retries]): [e]")
54
55                if retries < self.max_retries:
56                    # Optionally: exponential backoff
57                    import asyncio
58                    await asyncio.sleep(2 ** retries)
59
60        return AgentResult(
61            status=AgentStatus.FAILED,
62            error=last_error,
63            retries=retries
64        )
65
66    async def execute_all_with_fallback(
67        self,
68        tasks: list[tuple[str, str, dict]]
69    ) -> dict:
70        """Execute all tasks, continue even if some fail."""
71        import asyncio
72
73        async def safe_execute(agent_id: str, method: str, params: dict):
74            return agent_id, await self.execute_with_isolation(
75                agent_id, method, params
76            )
77
78        results_list = await asyncio.gather(*[
79            safe_execute(agent_id, method, params)
80            for agent_id, method, params in tasks
81        ])
82
83        # Categorize results
84        succeeded = {}
85        failed = {}
86
87        for agent_id, result in results_list:
88            if result.status == AgentStatus.COMPLETED:
89                succeeded[agent_id] = result.data
90            else:
91                failed[agent_id] = result.error
92
93        return {
94            "succeeded": succeeded,
95            "failed": failed,
96            "success_rate": len(succeeded) / len(tasks) if tasks else 0
97        }

1from dataclasses import dataclass
2from enum import Enum
3
4
5class ArchitectureRecommendation(Enum):
6    SINGLE_AGENT = "single_agent"
7    MULTI_AGENT = "multi_agent"
8    HYBRID = "hybrid"
9
10
11@dataclass
12class TaskCharacteristics:
13    """Characteristics to evaluate for architecture decision."""
14
15    # Task complexity
16    requires_multiple_expertise_domains: bool
17    has_independent_subtasks: bool
18    subtask_count: int
19
20    # Context requirements
21    total_context_needed: int  # tokens
22    context_window_limit: int
23    requires_different_tools: bool
24
25    # Reliability requirements
26    can_tolerate_single_failure: bool
27    needs_verification_steps: bool
28
29    # Performance requirements
30    latency_critical: bool
31    can_benefit_from_parallelism: bool
32
33
34def recommend_architecture(
35    task: TaskCharacteristics
36) -> ArchitectureRecommendation:
37    """Recommend architecture based on task characteristics."""
38
39    multi_agent_score = 0
40
41    # Expertise diversity
42    if task.requires_multiple_expertise_domains:
43        multi_agent_score += 3
44
45    # Subtask independence
46    if task.has_independent_subtasks:
47        multi_agent_score += 2
48    if task.subtask_count > 5:
49        multi_agent_score += 2
50
51    # Context constraints
52    if task.total_context_needed > task.context_window_limit * 0.7:
53        multi_agent_score += 3
54
55    # Tool requirements
56    if task.requires_different_tools:
57        multi_agent_score += 2
58
59    # Reliability
60    if not task.can_tolerate_single_failure:
61        multi_agent_score += 2
62    if task.needs_verification_steps:
63        multi_agent_score += 2
64
65    # Performance
66    if task.can_benefit_from_parallelism and not task.latency_critical:
67        multi_agent_score += 1
68
69    # Decision thresholds
70    if multi_agent_score >= 8:
71        return ArchitectureRecommendation.MULTI_AGENT
72    elif multi_agent_score >= 4:
73        return ArchitectureRecommendation.HYBRID
74    else:
75        return ArchitectureRecommendation.SINGLE_AGENT
76
77
78# Example evaluations
79def evaluate_examples():
80    # Simple Q&A chatbot
81    chatbot = TaskCharacteristics(
82        requires_multiple_expertise_domains=False,
83        has_independent_subtasks=False,
84        subtask_count=1,
85        total_context_needed=4000,
86        context_window_limit=128000,
87        requires_different_tools=False,
88        can_tolerate_single_failure=True,
89        needs_verification_steps=False,
90        latency_critical=True,
91        can_benefit_from_parallelism=False
92    )
93    print(f"Chatbot: [recommend_architecture(chatbot)]")
94    # Result: SINGLE_AGENT
95
96    # Complex code review system
97    code_review = TaskCharacteristics(
98        requires_multiple_expertise_domains=True,  # Security, performance, style
99        has_independent_subtasks=True,  # Each review type is independent
100        subtask_count=6,
101        total_context_needed=100000,
102        context_window_limit=128000,
103        requires_different_tools=True,  # Linters, analyzers, etc.
104        can_tolerate_single_failure=False,
105        needs_verification_steps=True,
106        latency_critical=False,
107        can_benefit_from_parallelism=True
108    )
109    print(f"Code Review: [recommend_architecture(code_review)]")
110    # Result: MULTI_AGENT
111
112    # Research assistant
113    research = TaskCharacteristics(
114        requires_multiple_expertise_domains=True,
115        has_independent_subtasks=True,  # Search, read, synthesize
116        subtask_count=4,
117        total_context_needed=50000,
118        context_window_limit=128000,
119        requires_different_tools=True,
120        can_tolerate_single_failure=True,
121        needs_verification_steps=True,
122        latency_critical=False,
123        can_benefit_from_parallelism=True
124    )
125    print(f"Research: [recommend_architecture(research)]")
126    # Result: HYBRID or MULTI_AGENT

1# Multi-agent customer support
2from dataclasses import dataclass, field
3from typing import Optional
4from enum import Enum
5
6
7class TicketCategory(Enum):
8    BILLING = "billing"
9    TECHNICAL = "technical"
10    GENERAL = "general"
11    ESCALATION = "escalation"
12
13
14@dataclass
15class SupportTicket:
16    """Customer support ticket."""
17    id: str
18    content: str
19    category: Optional[TicketCategory] = None
20    priority: int = 0
21    resolution: Optional[str] = None
22
23
24@dataclass
25class TriageAgent:
26    """Classifies and prioritizes incoming tickets."""
27
28    async def process(self, ticket: SupportTicket) -> SupportTicket:
29        # Analyze ticket content
30        category = await self.classify_category(ticket.content)
31        priority = await self.assess_priority(ticket.content)
32
33        ticket.category = category
34        ticket.priority = priority
35        return ticket
36
37    async def classify_category(self, content: str) -> TicketCategory:
38        # LLM-based classification
39        return TicketCategory.GENERAL
40
41    async def assess_priority(self, content: str) -> int:
42        # Priority scoring
43        return 1
44
45
46@dataclass
47class BillingAgent:
48    """Specialized in billing issues."""
49
50    system_prompt: str = """
51    You are a billing support specialist with access to:
52    - Subscription and payment history
53    - Refund policies and procedures
54    - Invoice generation systems
55
56    Resolve billing inquiries professionally.
57    """
58
59    async def process(self, ticket: SupportTicket) -> SupportTicket:
60        resolution = await self.resolve_billing_issue(ticket.content)
61        ticket.resolution = resolution
62        return ticket
63
64    async def resolve_billing_issue(self, content: str) -> str:
65        return "Billing issue resolved"
66
67
68@dataclass
69class TechnicalAgent:
70    """Specialized in technical issues."""
71
72    system_prompt: str = """
73    You are a technical support specialist with access to:
74    - Product documentation and knowledge base
75    - System logs and diagnostics
76    - Bug tracking and feature requests
77
78    Diagnose and resolve technical issues.
79    """
80
81    async def process(self, ticket: SupportTicket) -> SupportTicket:
82        resolution = await self.resolve_technical_issue(ticket.content)
83        ticket.resolution = resolution
84        return ticket
85
86    async def resolve_technical_issue(self, content: str) -> str:
87        return "Technical issue resolved"
88
89
90@dataclass
91class SupportOrchestrator:
92    """Routes tickets to appropriate specialist agents."""
93
94    triage: TriageAgent = field(default_factory=TriageAgent)
95    billing: BillingAgent = field(default_factory=BillingAgent)
96    technical: TechnicalAgent = field(default_factory=TechnicalAgent)
97
98    async def handle_ticket(self, ticket: SupportTicket) -> SupportTicket:
99        # Step 1: Triage
100        ticket = await self.triage.process(ticket)
101
102        # Step 2: Route to specialist
103        if ticket.category == TicketCategory.BILLING:
104            ticket = await self.billing.process(ticket)
105        elif ticket.category == TicketCategory.TECHNICAL:
106            ticket = await self.technical.process(ticket)
107        else:
108            # Handle with general agent or escalate
109            pass
110
111        return ticket

1# Multi-agent development pipeline
2from dataclasses import dataclass, field
3from typing import Optional
4from enum import Enum
5
6
7class ReviewStatus(Enum):
8    PENDING = "pending"
9    APPROVED = "approved"
10    NEEDS_CHANGES = "needs_changes"
11
12
13@dataclass
14class CodeChange:
15    """Represents a code change for review."""
16    files: list[str]
17    diff: str
18    security_review: Optional[str] = None
19    performance_review: Optional[str] = None
20    style_review: Optional[str] = None
21    status: ReviewStatus = ReviewStatus.PENDING
22
23
24@dataclass
25class ArchitectAgent:
26    """Designs system architecture."""
27
28    async def design(self, requirements: str) -> dict:
29        return {
30            "components": [],
31            "data_flow": {},
32            "api_contracts": {}
33        }
34
35
36@dataclass
37class SecurityReviewAgent:
38    """Reviews code for security issues."""
39
40    async def review(self, change: CodeChange) -> str:
41        # Security-focused analysis
42        return "Security review: No critical issues found"
43
44
45@dataclass
46class PerformanceReviewAgent:
47    """Reviews code for performance issues."""
48
49    async def review(self, change: CodeChange) -> str:
50        # Performance-focused analysis
51        return "Performance review: Acceptable"
52
53
54@dataclass
55class StyleReviewAgent:
56    """Reviews code for style compliance."""
57
58    async def review(self, change: CodeChange) -> str:
59        # Style and maintainability analysis
60        return "Style review: Follows conventions"
61
62
63@dataclass
64class DevelopmentPipeline:
65    """Orchestrates development workflow."""
66
67    architect: ArchitectAgent = field(default_factory=ArchitectAgent)
68    security: SecurityReviewAgent = field(default_factory=SecurityReviewAgent)
69    performance: PerformanceReviewAgent = field(default_factory=PerformanceReviewAgent)
70    style: StyleReviewAgent = field(default_factory=StyleReviewAgent)
71
72    async def review_change(self, change: CodeChange) -> CodeChange:
73        """Parallel review by all specialized agents."""
74        import asyncio
75
76        # Run all reviews in parallel
77        security_task = asyncio.create_task(self.security.review(change))
78        perf_task = asyncio.create_task(self.performance.review(change))
79        style_task = asyncio.create_task(self.style.review(change))
80
81        change.security_review = await security_task
82        change.performance_review = await perf_task
83        change.style_review = await style_task
84
85        # Determine overall status
86        change.status = self.aggregate_reviews(change)
87
88        return change
89
90    def aggregate_reviews(self, change: CodeChange) -> ReviewStatus:
91        """Aggregate individual reviews into final status."""
92        # If any review has critical issues, needs changes
93        reviews = [
94            change.security_review or "",
95            change.performance_review or "",
96            change.style_review or ""
97        ]
98
99        if any("critical" in r.lower() for r in reviews):
100            return ReviewStatus.NEEDS_CHANGES
101        return ReviewStatus.APPROVED