terminalskills 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
- Prediction Swarm (Vote & Aggregate) — Each agent analyzes independently with a different persona, then votes are aggregated
- Debate Swarm (Argue & Converge) — Agents see each other's reasoning and update positions over multiple rounds
- 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:
- Identify the pattern — Is it prediction (vote), debate (converge), or specialist (divide)?
- Define agents — Each agent needs a unique persona/perspective and clear role
- Choose aggregation — Weighted voting, median, debate rounds, or synthesis
- 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
| Strategy | Best For | How It Works |
|---|---|---|
| Majority Vote | Binary/categorical predictions | Most common answer wins |
| Weighted Vote | Varying agent confidence | Weight by confidence scores |
| Median | Numerical predictions | Take the median value |
| Debate | Complex reasoning | Multiple rounds of argumentation |
| Synthesis | Open-ended analysis | LLM combines all perspectives |
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
- Diversity is key — Agents with identical prompts add noise, not intelligence. Give each a distinct perspective
- Odd number of agents — Avoids ties in voting (5, 7, or 9 agents work best)
- Confidence calibration — Ask agents to self-report confidence; use it for weighting
- Cost control — Parallel calls are fast but expensive. Use cheaper models for screening, expensive for synthesis
- Sweet spot is 5-7 agents — Beyond 9, gains plateau due to diminishing returns
- Temperature variation — Use different temperatures per agent (0.3 for analytical, 0.9 for creative)
- 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