Multi-Agent System Example¶

def create_architect_agent(provider: Bundle) -> Bundle:
    """Designs system architecture and specifications"""
    return Bundle(
        name="architect",
        tools=[{"module": "tool-filesystem", "source": "..."}],
        instruction="""You are a Software Architect expert.

        Your role:
        - Design system architectures and specifications
        - Break down complex problems into modules
        - Define clear interfaces and contracts
        - Consider scalability, maintainability, and best practices

        When given a task:
        1. Analyze the requirements
        2. Design the architecture (components, interfaces, data flow)
        3. Create a specification document
        4. Return clear instructions for implementation

        Focus on design, not implementation. Be thorough but concise."""
    ).compose(provider)


def create_implementer_agent(provider: Bundle) -> Bundle:
    """Writes code based on specifications"""
    return Bundle(
        name="implementer",
        tools=[
            {"module": "tool-filesystem", "source": "..."},
            {"module": "tool-bash", "source": "..."}
        ],
        instruction="""You are a Software Implementation expert.

        Your role:
        - Implement code based on specifications
        - Write clean, tested, documented code
        - Follow the architecture and contracts provided
        - Run tests to verify correctness

        When given a specification:
        1. Understand the requirements
        2. Implement the code
        3. Write tests
        4. Verify it works
        5. Document the implementation

        Focus on clean, working code that matches the spec exactly."""
    ).compose(provider)


def create_reviewer_agent(provider: Bundle) -> Bundle:
    """Reviews code for quality, security, and best practices"""
    return Bundle(
        name="reviewer",
        tools=[{"module": "tool-filesystem", "source": "..."}],
        instruction="""You are a Code Review expert.

        Your role:
        - Review code for correctness, security, and best practices
        - Identify bugs, vulnerabilities, and improvement opportunities
        - Provide actionable feedback

        When reviewing code:
        1. Check correctness (does it match the spec?)
        2. Check security (any vulnerabilities?)
        3. Check quality (readable, maintainable?)
        4. Check tests (adequate coverage?)
        5. Provide specific, actionable feedback

        Be thorough but constructive. Focus on important issues."""
    ).compose(provider)

workflow = [
    (
        "architect",
        """Design a rate limiting module for API requests.

        Requirements:
        - Support different rate limit strategies (token bucket, sliding window)
        - Thread-safe for concurrent requests
        - Configurable limits per endpoint
        - Include usage tracking

        Provide:
        1. Module structure
        2. Class definitions and interfaces
        3. Key algorithms
        4. Example usage""",
        architect,
    ),
    (
        "implementer",
        """Implement the rate limiting module based on the architect's design.

        Create:
        1. rate_limiter.py with all classes
        2. A simple test to verify it works
        3. Brief usage example

        Keep it simple and focused.""",
        implementer,
    ),
    (
        "reviewer",
        """Review the rate limiter implementation.

        Check:
        1. Does it match the architecture?
        2. Are there any bugs or edge cases?
        3. Is it thread-safe?
        4. Is the code readable and maintainable?

        Provide specific feedback for improvement.""",
        reviewer,
    ),
]

# Execute workflow
results = await system.execute_workflow(task, workflow)

def _format_context(self, context: dict[str, Any], previous_results: dict[str, str]) -> str:
    """Format context from previous agents for the current agent."""
    if not previous_results:
        return f"Task: {context['task']}"

    sections = [f"Task: {context['task']}", "\nPrevious work:"]
    for agent_name, result in previous_results.items():
        sections.append(f"\n{agent_name.upper()} OUTPUT:")
        sections.append(result[:500] + "..." if len(result) > 500 else result)

    return "\n".join(sections)

class MultiAgentSystem:
    """Orchestrates multiple specialized agents to complete complex tasks."""

    def __init__(self, foundation: Bundle, provider: Bundle):
        self.foundation = foundation
        self.provider = provider

    async def execute_workflow(
        self, 
        task: str, 
        workflow: list[tuple[str, str, Bundle]]
    ) -> dict[str, Any]:
        """Execute a multi-agent workflow.

        Args:
            task: The overall task description
            workflow: List of (agent_name, instruction, agent_bundle) tuples

        Returns:
            Dict with results from each agent
        """
        results = {}
        context = {"task": task}

        for agent_name, instruction, agent_bundle in workflow:
            # Compose foundation + agent
            composed = self.foundation.compose(agent_bundle)

            # Prepare and execute
            prepared = await composed.prepare()
            session = await prepared.create_session()

            # Add context from previous agents
            context_str = self._format_context(context, results)
            full_instruction = f"{context_str}\n\n{instruction}"

            # Execute
            async with session:
                response = await session.execute(full_instruction)
                results[agent_name] = response

        return results

async def parallel_workflow():
    """Execute multiple independent agents in parallel."""

    # Create tasks
    tasks = [
        analyze_code(code_agent),
        check_security(security_agent),
        verify_tests(test_agent),
    ]

    # Run in parallel
    results = await asyncio.gather(*tasks)

    return results

# Clear contract
architect_output: Specification
implementer_input: Specification
implementer_output: Implementation
reviewer_input: Implementation

# Same agents, different workflows
workflow_1 = [architect, implementer, reviewer]
workflow_2 = [architect, implementer, tester, reviewer]
workflow_3 = [researcher, implementer, documenter]