Heavy Swarm Pattern

Dense collaboration network where all agents can interact with each other. Enables rich peer-to-peer communication and consensus building.

Overview

Heavy Swarm creates a fully connected network where every agent can communicate with every other agent. This pattern enables deep collaboration, debate, consensus building, and emergent intelligence through dense agent interactions.

When to Use

• Brainstorming: Generate diverse ideas through collaboration
• Consensus building: Reach agreement through discussion
• Peer review: Multiple agents review and critique each other
• Complex problem solving: Leverage collective intelligence
• Emergent solutions: Solutions emerge from interactions

Basic Usage

from azcore import HeavySwarm, Agent

# Create collaborative agents
agent1 = Agent(
    agent_name="Creative Thinker",
    system_prompt="Generate creative ideas and build on others' suggestions",
    llm=llm,
)

agent2 = Agent(
    agent_name="Critical Analyst",
    system_prompt="Critically analyze ideas and identify weaknesses",
    llm=llm,
)

agent3 = Agent(
    agent_name="Practical Implementer",
    system_prompt="Evaluate practical feasibility and implementation",
    llm=llm,
)

agent4 = Agent(
    agent_name="Risk Assessor",
    system_prompt="Identify and assess risks and challenges",
    llm=llm,
)

# Create heavy swarm
swarm = HeavySwarm(
    agents=[agent1, agent2, agent3, agent4],
    max_loops=5,  # 5 rounds of collaboration
)

# Run collaborative session
result = swarm.run("How should we approach building an AI education platform?")
print(result)

Configuration Options

max_loops

Number of collaboration rounds:

swarm = HeavySwarm(
    agents=[agent1, agent2, agent3],
    max_loops=10,  # More rounds = deeper collaboration
)

interaction_mode

How agents interact:

swarm = HeavySwarm(
    agents=[...],
    interaction_mode="round_robin",  # or "free_form" or "moderated"
)

consensus_threshold

When to stop collaborating:

swarm = HeavySwarm(
    agents=[...],
    max_loops=10,
    consensus_threshold=0.8,  # Stop when 80% agreement
)

moderator

Optional moderator agent:

moderator = Agent(
    agent_name="Moderator",
    system_prompt="Guide discussion and synthesize viewpoints",
    llm=llm,
)

swarm = HeavySwarm(
    agents=[agent1, agent2, agent3],
    moderator=moderator,
)

Advanced Examples

Design Review Board

from azcore import Agent, HeavySwarm

# Review board members
ux_designer = Agent(
    agent_name="UX Designer",
    system_prompt="""Review designs from user experience perspective.
    Comment on usability, accessibility, and user delight.
    Engage with other reviewers' feedback.""",
    llm=llm,
)

visual_designer = Agent(
    agent_name="Visual Designer",
    system_prompt="""Review visual design and aesthetics.
    Comment on color, typography, spacing, and brand alignment.
    Build on others' feedback.""",
    llm=llm,
)

frontend_engineer = Agent(
    agent_name="Frontend Engineer",
    system_prompt="""Review technical feasibility and implementation.
    Comment on complexity, performance, and maintainability.
    Consider others' concerns.""",
    llm=llm,
)

product_manager = Agent(
    agent_name="Product Manager",
    system_prompt="""Review business value and user needs.
    Comment on features, priorities, and trade-offs.
    Balance all perspectives.""",
    llm=llm,
)

accessibility_expert = Agent(
    agent_name="Accessibility Expert",
    system_prompt="""Review accessibility and inclusivity.
    Comment on WCAG compliance and assistive technology support.
    Ensure all voices are heard.""",
    llm=llm,
)

# Create review board
review_board = HeavySwarm(
    agents=[ux_designer, visual_designer, frontend_engineer, product_manager, accessibility_expert],
    max_loops=6,
)

# Conduct design review
review = review_board.run("Review the new checkout flow design")

Research Symposium

# Research perspectives
theorist = Agent(
    agent_name="Theorist",
    system_prompt="""Discuss theoretical frameworks and models.
    Propose theoretical explanations.
    Engage in scholarly debate.""",
    llm=llm,
)

empiricist = Agent(
    agent_name="Empiricist",
    system_prompt="""Focus on empirical evidence and data.
    Demand experimental validation.
    Challenge unsupported claims.""",
    llm=llm,
)

methodologist = Agent(
    agent_name="Methodologist",
    system_prompt="""Critique research methods and designs.
    Suggest methodological improvements.
    Ensure rigor.""",
    llm=llm,
)

applied_researcher = Agent(
    agent_name="Applied Researcher",
    system_prompt="""Consider practical applications.
    Bridge theory and practice.
    Focus on real-world impact.""",
    llm=llm,
)

skeptic = Agent(
    agent_name="Skeptic",
    system_prompt="""Question assumptions and conclusions.
    Play devil's advocate.
    Strengthen arguments through criticism.""",
    llm=llm,
)

# Create research symposium
symposium = HeavySwarm(
    agents=[theorist, empiricist, methodologist, applied_researcher, skeptic],
    max_loops=8,
)

# Discuss research question
discussion = symposium.run("Can AI achieve true understanding of language?")

Strategic Planning Session

# Strategy team
visionary = Agent(
    agent_name="Visionary",
    system_prompt="""Think big picture and long-term vision.
    Propose ambitious goals and strategies.
    Inspire bold thinking.""",
    llm=llm,
)

pragmatist = Agent(
    agent_name="Pragmatist",
    system_prompt="""Ground ideas in practical reality.
    Consider resources and constraints.
    Propose realistic paths forward.""",
    llm=llm,
)

analyst = Agent(
    agent_name="Analyst",
    system_prompt="""Analyze data and market trends.
    Provide evidence-based insights.
    Challenge assumptions with facts.""",
    llm=llm,
    tools=[market_research_tools],
)

innovator = Agent(
    agent_name="Innovator",
    system_prompt="""Propose creative solutions.
    Challenge conventional wisdom.
    Find novel approaches.""",
    llm=llm,
)

risk_manager = Agent(
    agent_name="Risk Manager",
    system_prompt="""Identify risks and challenges.
    Propose mitigation strategies.
    Ensure balanced risk-taking.""",
    llm=llm,
)

customer_advocate = Agent(
    agent_name="Customer Advocate",
    system_prompt="""Represent customer perspective.
    Ensure customer value is central.
    Challenge internally-focused thinking.""",
    llm=llm,
)

# Create strategy session
strategy_session = HeavySwarm(
    agents=[visionary, pragmatist, analyst, innovator, risk_manager, customer_advocate],
    max_loops=10,
)

# Develop strategy
strategy = strategy_session.run("Develop our 3-year product strategy")

Code Architecture Debate

# Architecture perspectives
architect_1 = Agent(
    agent_name="Microservices Advocate",
    system_prompt="""Advocate for microservices architecture.
    Highlight benefits of service decomposition.
    Address concerns raised by others.""",
    llm=llm,
)

architect_2 = Agent(
    agent_name="Monolith Advocate",
    system_prompt="""Advocate for monolithic architecture.
    Highlight benefits of simplicity.
    Challenge unnecessary complexity.""",
    llm=llm,
)

architect_3 = Agent(
    agent_name="Modular Monolith Advocate",
    system_prompt="""Advocate for modular monolith approach.
    Find middle ground between extremes.
    Balance trade-offs.""",
    llm=llm,
)

operations_engineer = Agent(
    agent_name="Operations Engineer",
    system_prompt="""Focus on operational concerns.
    Consider deployment, monitoring, debugging.
    Evaluate operational complexity.""",
    llm=llm,
)

developer = Agent(
    agent_name="Developer",
    system_prompt="""Consider developer experience.
    Evaluate development velocity and ease.
    Think about testing and debugging.""",
    llm=llm,
)

# Create architecture debate
architecture_debate = HeavySwarm(
    agents=[architect_1, architect_2, architect_3, operations_engineer, developer],
    max_loops=7,
)

# Decide architecture
decision = architecture_debate.run("Choose architecture for our new platform")

Interaction Modes

Round Robin

Agents take turns sequentially:

swarm = HeavySwarm(
    agents=[agent1, agent2, agent3],
    interaction_mode="round_robin",
    max_loops=5,
)

# Round 1: agent1, agent2, agent3
# Round 2: agent1, agent2, agent3
# ...

Free Form

Dynamic interaction based on relevance:

swarm = HeavySwarm(
    agents=[agent1, agent2, agent3],
    interaction_mode="free_form",
    max_loops=10,
)

# Agents speak when they have something to contribute

Moderated

Moderator controls the flow:

moderator = Agent(
    system_prompt="Guide discussion and call on agents",
    llm=llm,
)

swarm = HeavySwarm(
    agents=[agent1, agent2, agent3],
    moderator=moderator,
    interaction_mode="moderated",
)

Debate

Structured argumentation:

swarm = HeavySwarm(
    agents=[proposer, opposer, judge],
    interaction_mode="debate",
    max_loops=6,
)

# Proposer presents
# Opposer counters
# Back and forth
# Judge evaluates

Collaboration Patterns

1. Brainstorming

Generate and build on ideas:

agent_prompt = """Brainstorming mode:
- Propose new ideas
- Build on others' ideas
- Combine concepts
- No criticism in early rounds"""

2. Critique and Improve

Critical feedback loops:

agent_prompt = """Critique mode:
- Analyze others' proposals
- Identify weaknesses
- Suggest improvements
- Be constructive"""

3. Consensus Building

Converge on solutions:

agent_prompt = """Consensus mode:
- Find common ground
- Compromise when needed
- Synthesize viewpoints
- Move toward agreement"""

4. Debate

Structured argumentation:

agent_prompt = """Debate mode:
- Present clear arguments
- Counter opposing views
- Use evidence
- Maintain logical consistency"""

Best Practices

1. Diverse Perspectives

Ensure agents have different viewpoints:

# Good: Varied perspectives
[creative, critical, practical, visionary, skeptic]

# Bad: Similar agents
[analyst1, analyst2, analyst3]

2. Balanced Team Size

Optimal number of agents:

# Good: 4-6 agents
swarm = HeavySwarm(agents=[a1, a2, a3, a4, a5])

# Too few: 2 agents (use Sequential)
# Too many: 10+ agents (hard to coordinate)

3. Set Appropriate max_loops

Balance depth vs. efficiency:

# Quick brainstorm: 3-5 loops
swarm = HeavySwarm(agents=[...], max_loops=4)

# Deep collaboration: 8-12 loops
swarm = HeavySwarm(agents=[...], max_loops=10)

# Extended session: 15+ loops (monitor costs)

4. Use Moderator for Structure

For complex discussions:

moderator = Agent(
    system_prompt="""Guide productive discussion:
    - Keep focused on topic
    - Ensure all voices heard
    - Summarize progress
    - Drive toward conclusion""",
)

5. Design for Collaboration

Agents should be collaborative:

agent_prompt = """Collaborative agent guidelines:
- Listen to others
- Build on their ideas
- Acknowledge good points
- Disagree respectfully
- Work toward common goal"""

State Management

All agents see full conversation history:

# Each agent sees:
# - Original task
# - All previous agent outputs
# - Full discussion thread

# State grows linearly with interactions
# Can become very large

Manage state growth:

class EfficientHeavySwarm(HeavySwarm):
    def prune_history(self, state, max_messages=50):
        # Keep only recent messages
        if len(state["messages"]) > max_messages:
            state["messages"] = state["messages"][-max_messages:]
        return state

Error Handling

Handle unproductive discussions:

class RobustHeavySwarm(HeavySwarm):
    def detect_circular_discussion(self, state):
        # Detect when agents repeat themselves
        if self.is_circular(state):
            return self.summarize_and_conclude(state)
        return state

    def handle_disagreement(self, state):
        # Handle persistent disagreement
        if not self.is_converging(state):
            return self.call_vote(state)
        return state

Performance Considerations

Complexity

Heavy Swarm is the most complex pattern:

# Interactions = max_loops × num_agents
# State size = sum of all agent outputs
# Can be slow and expensive

Cost Optimization

# Total LLM calls = max_loops × num_agents

# For 5 agents, 10 loops: 50 LLM calls
# Monitor costs carefully

# Optimization strategies:
# 1. Use cheaper models
# 2. Reduce max_loops
# 3. Implement early stopping
# 4. Prune conversation history

When to Stop

Implement early termination:

swarm = HeavySwarm(
    agents=[...],
    max_loops=20,
    consensus_threshold=0.85,  # Stop at 85% agreement
    convergence_check=True,     # Stop when converged
)

Comparison with Other Patterns

Debugging

Monitor collaboration dynamics:

swarm = HeavySwarm(
    agents=[agent1, agent2, agent3, agent4],
    max_loops=10,
    verbose=True,
)

result = swarm.run("Task")

# Analyze interaction patterns
for round_num, round_outputs in enumerate(result["rounds"]):
    print(f"\n=== Round {round_num} ===")
    for agent_name, output in round_outputs.items():
        print(f"{agent_name}: {output[:100]}...")

# Check convergence
convergence = result["convergence_score"]
print(f"Convergence: {convergence}")

# Identify dominant agents
agent_contributions = result["agent_statistics"]
for agent, stats in agent_contributions.items():
    print(f"{agent}: {stats['word_count']} words, {stats['interactions']} turns")

Visualize discussion flow:

def visualize_discussion(result):
    import matplotlib.pyplot as plt

    # Track sentiment/agreement over rounds
    rounds = []
    agreement_scores = []

    for i, round_data in enumerate(result["rounds"]):
        rounds.append(i)
        agreement_scores.append(round_data["agreement_score"])

    plt.plot(rounds, agreement_scores)
    plt.xlabel("Round")
    plt.ylabel("Agreement Score")
    plt.title("Convergence Over Time")
    plt.show()

Common Patterns

Deliberation

Careful consideration of options:

# Present options
# Discuss pros/cons
# Debate trade-offs
# Reach decision

Peer Review

Mutual critique and improvement:

# Each agent reviews others' work
# Provide constructive feedback
# Iterate based on feedback
# Converge on quality output

Collective Problem Solving

Emergent solutions:

# Agents propose partial solutions
# Build on each other's ideas
# Combine approaches
# Emerge with integrated solution

Advanced Techniques

Consensus Detection

Automatically detect agreement:

def detect_consensus(agent_outputs):
    # Use embeddings to measure similarity
    # Calculate agreement score
    # Return True when threshold reached
    pass

Role Rotation

Agents change roles over time:

def rotate_roles(agents, round_num):
    # Agents switch perspectives
    # Prevents groupthink
    # Explores different angles
    pass

Sub-Group Formation

Create temporary sub-groups:

def form_subgroups(agents, topic):
    # Group agents by expertise
    # Have sub-group discussions
    # Report back to main group
    pass