Swarms Multi-Agent Framework

from swarms import Agent, GraphWorkflow

wf = GraphWorkflow(name="research-dag", max_loops=1)

for a in (ingestion, branch_a, branch_b, merger):
    wf.add_node(a)

wf.add_edge(ingestion, branch_a)      # fan out
wf.add_edge(ingestion, branch_b)
wf.add_edge(branch_a, merger)         # fan in
wf.add_edge(branch_b, merger)

wf.set_entry_points(["Ingestion"])
wf.set_end_points(["Merger"])

def on_done(node: str, result) -> None:
    print(f"[{node}] {len(str(result))} chars")

results = wf.run(task="Analyze this dataset two ways and merge.", on_node_complete=on_done)

from swarms import Agent, SwarmRouter

router = SwarmRouter(agents=agents, swarm_type="SequentialWorkflow", max_loops=1)
router.run("Write a post about transformers.")

from swarms import Agent, MixtureOfAgents

moa = MixtureOfAgents(
    agents=[worker_gpt, worker_claude, worker_llama],
    aggregator_agent=aggregator,       # optional; falls back to aggregator_model_name
    layers=3,
    max_loops=1,
)
moa.run("Best practices for securing a Kubernetes cluster?")

from swarms import Agent, HierarchicalSwarm

swarm = HierarchicalSwarm(
    agents=[data_worker, writing_worker, review_worker],
    director=director,            # optional; else built from director_model_name
    max_loops=2,
    planning_enabled=True,
    parallel_execution=True,
    director_feedback_on=True,
)
swarm.run("Produce a competitive analysis of the AI chip market.")

from swarms import Agent, GroupChat

chat = GroupChat(
    agents=[optimist, pessimist, realist],   # at least 2 required
    max_loops=10,
    threshold=0.5,           # raise for a more selective room
    recency_penalty=0.3,     # discourages one agent dominating
    idle_timeout=8.0,
)
chat.run("Should we adopt AI for medical diagnosis?")

from swarms import Agent, MajorityVoting

mv = MajorityVoting(
    agents=voters,
    consensus_agent_model_name="gpt-5.4",
    max_loops=1,
)
mv.run("Python or Rust for a high-performance web server?")

from swarms import CouncilAsAJudge

council = CouncilAsAJudge(
    model_name="gpt-5.4",
    aggregation_model_name="gpt-5.4",
    random_model_name=True,
    max_loops=1,
)
council.run("Should we store biometric data on-device only?")

from swarms import Agent, DebateWithJudge

debate = DebateWithJudge(
    pro_agent=pro,
    con_agent=con,
    judge_agent=judge,
    max_loops=3,          # rounds
)
debate.run("Motion: open-source LLMs will surpass closed-source by 2027.")

from swarms import HeavySwarm

swarm = HeavySwarm(
    question_agent_model_name="gpt-5.4",
    worker_model_name="gpt-5.4",
    max_loops=1,
    timeout=900,
    show_dashboard=True,
    worker_tools=[search_web],
)
swarm.run("Analyze the implications of AGI on global labour markets.")

from swarms.structs.planner_worker_swarm import PlannerWorkerSwarm

swarm = PlannerWorkerSwarm(
    agents=workers,                    # workers only — the planner is built internally
    planner_model_name="gpt-5.4",
    judge_model_name="gpt-5.4",
    max_planner_depth=1,
    max_loops=1,
)
swarm.run("Build a go-to-market strategy for a B2B SaaS product.")

from swarms import (
    MultiAgentRouter,      # routes each task to the best-fit agent
    RoundRobinSwarm,       # fixed rotation
    LLMCouncil,            # members answer, rank peers anonymously, chairman synthesizes
    BatchedGridWorkflow,   # agent i runs task i
    AutoSwarmBuilder,      # generates the agents and architecture from a description
    SpreadSheetSwarm,      # structured tabular processing
    AdvisorSwarm, SelfMoASeq, HybridHierarchicalClusterSwarm,
)

builder = AutoSwarmBuilder(name="MarketResearch", description="...", max_loops=1)
builder.run("Research the EV market and find growth opportunities.")

from swarms import (
    run_agents_concurrently,
    run_agents_with_different_tasks,
    run_agents_concurrently_async,
    batch_agent_execution,
    run_single_agent,
    aggregate,
)

run_agents_concurrently(agents=agents, task="Summarize today's news.", max_workers=8)
run_agents_with_different_tasks([(agent_a, "task A"), (agent_b, "task B")])  # list of tuples
batch_agent_execution(agents=agents, tasks=tasks, max_workers=10)
aggregate(workers=agents, task="...", aggregator_model_name="gpt-5.4")

from swarms import CronJob

job = CronJob(agent=agent, interval="10minutes", job_id="market-check")
job.run(task="Check for unusual market activity.")

from swarms import AgentLoader

loader = AgentLoader(concurrent=True)
agents = loader.load_agents_from_markdown("agents/")   # also: _from_yaml, _from_csv
agent = loader.load_agent_from_markdown("agents/researcher.md")

agent = Agent(
    agent_name="ProductionAgent",
    agent_description="...",
    model_name="gpt-5.4",
    max_loops=3,
    context_length=32000,
    context_compression=True,
    persistent_memory=True,
    autosave=True,
    retry_attempts=3,
    fallback_models=["claude-sonnet-4-6"],
    verbose=False,
)