多智能体协作模式：辩论、投票、Mixture of Agents 与委员会

pip install aiohttp python-dotenv
export SWARMS_API_KEY="your-api-key"

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def main() -> None:
    payload = {
        "name": "Database Split Debate",
        "description": "Pro and Con argue, a judge rules",
        "swarm_type": "DebateWithJudge",
        "task": (
            "Should a 10-engineer startup with one Postgres database split it into "
            "one database per service this quarter? Consider delivery speed, "
            "incident risk, data consistency and hiring."
        ),
        "agents": [
            {
                "agent_name": "Pro",
                "description": "Argues for the proposition",
                "system_prompt": (
                    "You argue FOR the proposition. Give your three strongest arguments "
                    "with concrete mechanisms and answer the other side's best point. "
                    "In later rounds, respond directly to the judge's feedback."
                ),
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
                "temperature": 0.5,
            },
            {
                "agent_name": "Con",
                "description": "Argues against the proposition",
                "system_prompt": (
                    "You argue AGAINST the proposition. Give your three strongest arguments "
                    "with concrete mechanisms and answer the other side's best point. "
                    "In later rounds, respond directly to the judge's feedback."
                ),
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
                "temperature": 0.5,
            },
            {
                "agent_name": "Judge",
                "description": "Weighs both sides and writes the verdict",
                "system_prompt": (
                    "You are an impartial judge. Assess each side on evidence and logic, "
                    "name the strongest point from each, then give a clear verdict and "
                    "the conditions under which it would change."
                ),
                "model_name": "gpt-5.4",
                "max_loops": 1,
                "temperature": 0.2,
            },
        ],
        "max_loops": 2,  # two rounds, three calls per round
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    judge_turns = [t for t in result["output"] if t["role"] == "Judge"]
    print(f"{len(judge_turns)} rounds in {result['execution_time']:.1f}s")
    print("Final verdict:\n", text(judge_turns[-1]))


asyncio.run(main())

import asyncio
import os
import re
from collections import Counter

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}

DIFF = '''
def get_user(conn, username):
    query = f"SELECT * FROM users WHERE name = '{username}'"
    return conn.execute(query).fetchone()
'''

VOTE_RULE = (
    "Start your reply with exactly 'VOTE: APPROVE' or 'VOTE: REJECT', "
    "then give at most three reasons."
)

voters = [
    ("Security-Reviewer", "You review code for security flaws such as injection and unsafe input handling.", "gpt-5.4"),
    ("Correctness-Reviewer", "You review code for bugs, edge cases and wrong results.", "claude-sonnet-5"),
    ("Maintainability-Reviewer", "You review code for readability, naming and testability.", "gpt-5.4-mini"),
]


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def main() -> None:
    payload = {
        "name": "PR Approval Vote",
        "description": "Independent reviewers vote on a pull request",
        "swarm_type": "MajorityVoting",
        "task": f"Vote on whether this change can merge.\n{VOTE_RULE}\n\n{DIFF}",
        "agents": [
            {
                "agent_name": name,
                "description": prompt,
                "system_prompt": f"{prompt} {VOTE_RULE}",
                "model_name": model,
                "max_loops": 1,
                "temperature": 0.2,
            }
            for name, prompt, model in voters
        ],
        "max_loops": 1,
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    voter_names = {name for name, _, _ in voters}
    tally = Counter()
    for turn in result["output"]:
        match = re.match(r"\s*\**VOTE:\s*(APPROVE|REJECT)", text(turn), re.IGNORECASE)
        if turn["role"] in voter_names and match:
            tally[match.group(1).upper()] += 1

    consensus = [t for t in result["output"] if t["role"] == "Consensus-Agent"]
    print("Our tally:", dict(tally))
    print("Consensus agent:\n", text(consensus[-1]) if consensus else "(none returned)")


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def main() -> None:
    payload = {
        "name": "Region Migration Plan",
        "description": "Three proposers, one synthesizer",
        "swarm_type": "MixtureOfAgents",
        "task": (
            "Write a step-by-step plan to move a 2 TB Postgres database to a new "
            "cloud region with under 5 minutes of write downtime. Include the "
            "rollback plan and the checks that gate each step."
        ),
        "agents": [
            {
                "agent_name": "Database-Engineer",
                "description": "Replication and cutover mechanics",
                "system_prompt": "Focus on replication setup, lag monitoring and the cutover sequence.",
                "model_name": "gpt-5.4",
                "max_loops": 1,
                "temperature": 0.3,
            },
            {
                "agent_name": "SRE",
                "description": "Operational risk and observability",
                "system_prompt": "Focus on monitoring, alerting, runbooks and what can go wrong at 3 a.m.",
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
                "temperature": 0.3,
            },
            {
                "agent_name": "Risk-Reviewer",
                "description": "Failure cases and rollback",
                "system_prompt": "Focus on data loss scenarios, rollback paths and go or no-go criteria.",
                "model_name": "gpt-5.4-mini",
                "max_loops": 1,
                "temperature": 0.3,
            },
            {
                # The last agent is reused as the aggregator.
                "agent_name": "Synthesizer",
                "description": "Merges the proposals into one plan",
                "system_prompt": (
                    "You receive the task plus other engineers' proposals. Merge them into "
                    "one numbered plan, resolve conflicts explicitly, and drop repetition."
                ),
                "model_name": "gpt-5.4",
                "max_loops": 1,
                "temperature": 0.2,
            },
        ],
        "max_loops": 1,  # one proposer layer
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    # The synthesizer appears twice: once as a proposer, then as the aggregator.
    synthesis = [t for t in result["output"] if t["role"] == "Synthesizer"][-1]
    print(text(synthesis))
    usage = result.get("usage") or {}
    print("Total cost:", usage.get("billing_info", {}).get("total_cost"))


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def main() -> None:
    payload = {
        "name": "Moderation Latency Council",
        "description": "Members answer, review each other, chairman decides",
        "swarm_type": "LLMCouncil",
        "chairman_model": "gpt-5.4",
        "task": (
            "Our API's p99 latency doubled after we added an LLM moderation step "
            "before every response. Options: (a) run moderation asynchronously and "
            "retract flagged responses, (b) switch to a smaller moderation model, "
            "(c) cache verdicts for repeated content. Recommend one and say why."
        ),
        "agents": [
            {
                "agent_name": "Latency-Engineer",
                "description": "Optimizes for response time",
                "system_prompt": "You care most about latency budgets and tail behavior.",
                "model_name": "gpt-5.4",
                "max_loops": 1,
                "temperature": 0.4,
            },
            {
                "agent_name": "Trust-and-Safety",
                "description": "Optimizes for harm prevention",
                "system_prompt": "You care most about harmful content reaching users, even briefly.",
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
                "temperature": 0.4,
            },
            {
                "agent_name": "Cost-Analyst",
                "description": "Optimizes for spend",
                "system_prompt": "You care most about cost per request and engineering effort.",
                "model_name": "gpt-5.4-mini",
                "max_loops": 1,
                "temperature": 0.4,
            },
        ],
        "max_loops": 1,
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    for turn in result["output"]:
        if turn["role"].endswith("-Evaluation"):
            print(f"--- {turn['role']} ---\n{text(turn)[:400]}\n")

    chairman = [t for t in result["output"] if t["role"] == "Chairman"]
    print("Chairman:\n", text(chairman[-1]) if chairman else "(none returned)")


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}

original_task = "Explain to a junior engineer when to add a database index."
candidate = (
    "Indexes always make queries faster, so add one to every column. "
    "They have no downside because the database maintains them for free."
)


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def main() -> None:
    payload = {
        "name": "Answer Review Council",
        "description": "Six judges grade one response, an aggregator rules",
        "swarm_type": "CouncilAsAJudge",
        "council_judge_model_name": "gpt-5.4",
        "task": f"Task: {original_task}\n\nResponse: {candidate}",
        "agents": [
            {
                # Required by the API for this type, but never executed.
                "agent_name": "placeholder",
                "description": "Not run by CouncilAsAJudge",
                "model_name": "gpt-5.4-mini",
                "max_loops": 1,
            }
        ],
        "max_loops": 1,
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    for turn in result["output"]:
        if turn["role"].endswith("_judge"):
            print(f"{turn['role']}: {text(turn)[:200]}...")

    ruling = [t for t in result["output"] if t["role"] == "aggregator_agent"]
    print("\nRuling:\n", text(ruling[-1]) if ruling else "(none returned)")


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def main() -> None:
    payload = {
        "name": "Reliability Sprint Planning",
        "description": "Three roles plan a one-week sprint together",
        "swarm_type": "GroupChat",
        "task": (
            "Plan a one-week reliability sprint for our payments API, which had two "
            "outages last month (a bad config push and a database connection storm). "
            "Agree on at most five work items, an owner role for each, and what we drop."
        ),
        "agents": [
            {
                "agent_name": "SRE",
                "description": "Owns incident data and operational risk",
                "system_prompt": "Ground every proposal in the two outages. Push back on vague items.",
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
                "temperature": 0.5,
            },
            {
                "agent_name": "Backend-Engineer",
                "description": "Estimates effort and feasibility",
                "system_prompt": "Estimate effort for each proposal and flag what will not fit in a week.",
                "model_name": "gpt-5.4",
                "max_loops": 1,
                "temperature": 0.5,
            },
            {
                "agent_name": "Product-Manager",
                "description": "Owns priorities and trade-offs",
                "system_prompt": "Decide what gets cut, and summarize the agreed plan when the group converges.",
                "model_name": "gpt-5.4-mini",
                "max_loops": 1,
                "temperature": 0.5,
            },
        ],
        "max_loops": 6,  # total messages in the conversation
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    for turn in result["output"]:
        print(f"[{turn['role']}] {text(turn)[:300]}\n")


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}

TASK = (
    "Our checkout service times out on 2% of calls at peak hours. Should we "
    "put a cache in front of the pricing service or scale the pricing service "
    "horizontally? Recommend one and say why."
)

MEMBERS = [
    {
        "agent_name": "Performance-Engineer",
        "description": "Focuses on latency",
        "system_prompt": "You care most about latency and tail behavior.",
        "model_name": "gpt-5.4",
        "max_loops": 1,
    },
    {
        "agent_name": "Reliability-Engineer",
        "description": "Focuses on failure modes",
        "system_prompt": "You care most about stale data, failure modes and recovery.",
        "model_name": "claude-sonnet-5",
        "max_loops": 1,
    },
    {
        "agent_name": "Cost-Analyst",
        "description": "Focuses on spend",
        "system_prompt": "You care most about infrastructure cost and engineering effort.",
        "model_name": "gpt-5.4-mini",
        "max_loops": 1,
    },
]

SYNTHESIZER = {
    "agent_name": "Synthesizer",
    "description": "Merges the proposals into one recommendation",
    "system_prompt": "Merge the other agents' answers into one recommendation with its reasons.",
    "model_name": "gpt-5.4",
    "max_loops": 1,
}

PAYLOADS = [
    {"name": "Compare: voting", "swarm_type": "MajorityVoting",
     "task": TASK, "agents": MEMBERS, "max_loops": 1},
    {"name": "Compare: council", "swarm_type": "LLMCouncil", "chairman_model": "gpt-5.4",
     "task": TASK, "agents": MEMBERS, "max_loops": 1},
    {"name": "Compare: mixture", "swarm_type": "MixtureOfAgents",
     "task": TASK, "agents": MEMBERS + [SYNTHESIZER], "max_loops": 1},
]


def text(turn):
    content = turn["content"]
    return content if isinstance(content, str) else " ".join(map(str, content))


async def run_swarm(session: aiohttp.ClientSession, payload: dict) -> dict:
    async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
        if resp.status >= 400:
            body = await resp.text()
            raise RuntimeError(f"{resp.status}: {body}")
        return await resp.json()


async def main() -> None:
    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        # One shared session; the three runs are in flight at the same time.
        results = await asyncio.gather(
            *(run_swarm(session, p) for p in PAYLOADS), return_exceptions=True
        )

    for payload, result in zip(PAYLOADS, results):
        if isinstance(result, Exception):
            print(f"--- {payload['swarm_type']} failed: {result}\n")
            continue
        usage = result.get("usage") or {}
        cost = usage.get("billing_info", {}).get("total_cost")
        print(f"--- {payload['swarm_type']}: {result['execution_time']:.1f}s, cost {cost}")
        # The last turn holds each pattern's final answer.
        print(text(result["output"][-1])[:500], "\n")


asyncio.run(main())