Swarm Intelligence

Overview

Build systems where multiple AI agents independently analyze a problem, then converge on predictions through voting, debate, or weighted aggregation. Inspired by biological swarms and ensemble methods — the collective intelligence of diverse agents consistently outperforms any single agent.

Core Patterns

1. Prediction Swarm (Vote & Aggregate) — Each agent analyzes independently with a different persona, then votes are aggregated
2. Debate Swarm (Argue & Converge) — Agents see each other's reasoning and update positions over multiple rounds
3. Specialist Swarm (Divide & Conquer) — Each agent handles a different domain aspect, then a synthesizer combines results

Instructions

When a user asks to build a swarm intelligence system, prediction ensemble, or multi-agent decision system:

1. Identify the pattern — Is it prediction (vote), debate (converge), or specialist (divide)?
2. Define agents — Each agent needs a unique persona/perspective and clear role
3. Choose aggregation — Weighted voting, median, debate rounds, or synthesis
4. Implement with LangGraph — Use parallel nodes for agents, then aggregation node

Prediction Swarm Implementation

"""Prediction swarm: N agents vote independently, aggregator combines."""
import json, operator
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

class SwarmState(TypedDict):
    question: str
    predictions: Annotated[list[dict], operator.add]
    final_answer: str

AGENT_PERSONAS = [
    {"name": "Optimist", "prompt": "You see opportunities and upside potential."},
    {"name": "Skeptic", "prompt": "You question assumptions and look for flaws."},
    {"name": "Analyst", "prompt": "You focus on data and historical patterns."},
    {"name": "Contrarian", "prompt": "You challenge consensus. Look for what others miss."},
    {"name": "Pragmatist", "prompt": "You focus on practical, real-world constraints."},
]

def make_agent_node(persona: dict):
    llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
    def agent_fn(state: SwarmState) -> dict:
        response = llm.invoke(
            f"You are the {persona['name']}. {persona['prompt']}\n\n"
            f"Question: {state['question']}\n\n"
            f"Respond with JSON: {{\"prediction\": \"...\", \"confidence\": 0.0-1.0, \"reasoning\": \"...\"}}"
        )
        prediction = json.loads(response.content)
        prediction["agent"] = persona["name"]
        return {"predictions": [prediction]}
    return agent_fn

def aggregator(state: SwarmState) -> dict:
    predictions = state["predictions"]
    votes: dict[str, float] = {}
    reasoning_parts = []
    for p in predictions:
        votes[p["prediction"]] = votes.get(p["prediction"], 0) + p["confidence"]
        reasoning_parts.append(f"- {p['agent']} ({p['confidence']:.0%}): {p['reasoning']}")
    winner = max(votes, key=votes.get)
    avg_conf = sum(p["confidence"] for p in predictions) / len(predictions)
    return {"final_answer": f"**Prediction:** {winner}\n**Confidence:** {avg_conf:.0%}\n**Breakdown:**\n" + "\n".join(reasoning_parts)}

# Build the graph
builder = StateGraph(SwarmState)
for persona in AGENT_PERSONAS:
    builder.add_node(persona["name"], make_agent_node(persona))
    builder.add_edge("__start__", persona["name"])
builder.add_node("aggregator", aggregator)
for persona in AGENT_PERSONAS:
    builder.add_edge(persona["name"], "aggregator")
builder.add_edge("aggregator", END)
swarm = builder.compile()

Debate Swarm (Multi-Round Convergence)

"""Debate swarm: agents see each other's reasoning and update positions."""
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.5)
DEBATE_AGENTS = [
    {"name": "Bull", "bias": "optimistic"},
    {"name": "Bear", "bias": "pessimistic"},
    {"name": "Quant", "bias": "data-driven"},
]

def run_debate(question: str, rounds: int = 3) -> dict:
    history = []
    for round_num in range(1, rounds + 1):
        round_responses = []
        for agent in DEBATE_AGENTS:
            context = ""
            if history:
                context = "Previous positions:\n" + "\n".join(
                    f"- {r['agent']}: {r['position']} (conf: {r['confidence']})" for r in history[-1]
                )
            response = llm.invoke(
                f"You are {agent['name']}, a {agent['bias']} analyst.\n"
                f"Question: {question}\nRound {round_num}/{rounds}.\n{context}\n\n"
                f"State your position, confidence (0-1), and reasoning."
            )
            round_responses.append({"agent": agent["name"], "position": response.content[:200], "confidence": 0.7, "full": response.content})
        history.append(round_responses)
    final = llm.invoke(
        f"Question: {question}\n\nFinal positions after {rounds} rounds:\n"
        + "\n".join(f"- {r['agent']}: {r['full']}" for r in history[-1])
        + "\n\nSynthesize a consensus answer."
    )
    return {"rounds": history, "consensus": final.content}

Aggregation Strategies

Examples

Example 1: Market Trend Prediction

result = swarm.invoke({"question": "Will AI agents replace 50% of SaaS tools by 2027?"})
print(result["final_answer"])
# Output:
# **Prediction:** Unlikely within that timeframe
# **Confidence:** 68%
# **Breakdown:**
# - Optimist (85%): AI agents will automate many workflows but full replacement takes longer
# - Skeptic (40%): Enterprise adoption is slow, regulatory hurdles remain
# - Analyst (65%): Historical tech adoption curves suggest 2029-2030
# - Contrarian (70%): The question is wrong — agents will augment, not replace
# - Pragmatist (55%): Integration complexity means gradual transition

Example 2: Multi-Domain Business Analysis

SPECIALISTS = {
    "market": "Analyze market size, competition, and demand signals.",
    "technical": "Assess technical feasibility and architecture risks.",
    "financial": "Model costs, revenue potential, and break-even timeline.",
    "legal": "Identify regulatory risks and compliance needs.",
}

def specialist_swarm(question: str) -> str:
    analyses = {}
    for domain, prompt in SPECIALISTS.items():
        response = llm.invoke(f"You are a {domain} specialist. {prompt}\n\nQuestion: {question}")
        analyses[domain] = response.content
    synthesis = llm.invoke(
        f"Specialist analyses for: {question}\n\n"
        + "\n\n".join(f"**{k.upper()}:**\n{v}" for k, v in analyses.items())
        + "\n\nSynthesize into a unified recommendation."
    )
    return synthesis.content

# Usage: specialist_swarm("Should we build a competitor to Notion using AI-native architecture?")

Guidelines

1. Diversity is key — Agents with identical prompts add noise, not intelligence. Give each a distinct perspective
2. Odd number of agents — Avoids ties in voting (5, 7, or 9 agents work best)
3. Confidence calibration — Ask agents to self-report confidence; use it for weighting
4. Cost control — Parallel calls are fast but expensive. Use cheaper models for screening, expensive for synthesis
5. Sweet spot is 5-7 agents — Beyond 9, gains plateau due to diminishing returns
6. Temperature variation — Use different temperatures per agent (0.3 for analytical, 0.9 for creative)
7. Use swarms for high-stakes decisions — For simple tasks, a single agent is faster and cheaper

Dependencies

pip install langgraph langchain-openai   # Python
npm install openai                        # Node.js