如何用 Python 构建智能体蜂群：分步教程

swarm-tutorial/
  .env
  common.py            # step 0: setup, helpers, the task
  step1_agent.py
  step2_pipeline.py
  step3_concurrent.py
  step3_mixture.py
  step4_hierarchical.py
  ...

pip install aiohttp python-dotenv

SWARMS_API_KEY=your_api_key_here

# common.py: shared setup for every step in this tutorial
import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]  # fails fast with KeyError if the key is missing

BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}
# Multi-agent runs can take minutes, so give the whole request room.
TIMEOUT = aiohttp.ClientTimeout(total=600)


def new_session():
    """Open one session per program, inside a coroutine, and reuse it for every call."""
    return aiohttp.ClientSession(headers=HEADERS, timeout=TIMEOUT)


async def post(session, path, payload):
    """POST to the Swarms API. Returns the JSON body or raises with the error detail."""
    async with session.post(f"{BASE_URL}{path}", json=payload) as resp:
        if resp.status >= 400:
            body = await resp.text()
            raise RuntimeError(f"{path} returned {resp.status}: {body}")
        return await resp.json()


async def run_agent(session, payload):
    return await post(session, "/v1/agent/completions", payload)


async def run_swarm(session, payload):
    return await post(session, "/v1/swarm/completions", payload)


def print_turns(turns):
    """Print a list of {"role": ..., "content": ...} turns."""
    for turn in turns:
        print(f"\n=== {turn['role']} ===\n{turn['content']}")


def print_usage(result):
    """Agent completions report total_cost directly; swarm completions nest it in billing_info."""
    usage = result.get("usage") or {}
    cost = usage.get("total_cost", (usage.get("billing_info") or {}).get("total_cost"))
    print(f"\ntokens: {usage.get('total_tokens')}  cost: ${cost}")


PRODUCT_NOTES = """
Product: Acme Invoice (a made-up example), invoicing and expense tracking for freelancers.
Pricing: free for up to 5 clients, then $12 per month.
Strengths: invoices built from time logs, multi-currency, receipt capture on mobile.
Gaps: no payroll, no inventory, integrations limited to Stripe and PayPal.
Customers: solo freelancers and two-person studios in the US and EU.
Competitors to cover: a full small-business accounting suite, a free invoice
generator, and a payments platform with built-in invoicing.
"""


def make_task(notes):
    return (
        "Write a one-page competitive analysis brief for the product below. "
        "Cover positioning, a pricing comparison, the main risks, and three "
        "recommendations for the next two quarters. Use only the facts given "
        "and label anything else as an assumption.\n" + notes
    )


TASK = make_task(PRODUCT_NOTES)


async def main():
    async with new_session() as session:
        # Health check from the quickstart: is the API reachable?
        async with session.get(f"{BASE_URL}/health") as resp:
            print(await resp.json())
        # A 200 with your balance confirms the key works.
        async with session.get(f"{BASE_URL}/v1/account/credits") as resp:
            print(resp.status, await resp.json())


if __name__ == "__main__":
    asyncio.run(main())

# step1_agent.py: one agent writes the whole brief (uses common.py from step 0)
import asyncio

from common import TASK, new_session, print_usage, run_agent

payload = {
    "agent_config": {
        "agent_name": "Competitive Analyst",
        "description": "Writes competitive analysis briefs for product teams",
        "system_prompt": (
            "You are a product marketing analyst. Write short, specific "
            "competitive briefs with clear headings. Do not invent figures."
        ),
        "model_name": "claude-sonnet-5",
        "max_loops": 1,
        "max_tokens": 4096,
        "temperature": 0.4,
    },
    "task": TASK,
}


async def main():
    async with new_session() as session:
        result = await run_agent(session, payload)
    print(result["outputs"][-1]["content"])
    print_usage(result)


asyncio.run(main())

{
  "job_id": "agent-...",
  "success": true,
  "name": "Competitive Analyst",
  "description": "Writes competitive analysis briefs for product teams",
  "temperature": 0.4,
  "outputs": [
    {
      "role": "Competitive Analyst",
      "content": "# Acme Invoice: Competitive Brief\n\n## Positioning\nAcme Invoice sits between the free invoice generator and the accounting suite: more automation than the first, much less setup than the second..."
    }
  ],
  "usage": { "input_tokens": ..., "output_tokens": ..., "total_tokens": ..., "total_cost": ... },
  "timestamp": "..."
}

# step2_pipeline.py: researcher -> analyst -> writer (uses common.py from step 0)
import asyncio

from common import TASK, new_session, print_turns, print_usage, run_swarm

PIPELINE_AGENTS = [
    {
        "agent_name": "Market Researcher",
        "description": "Extracts facts and open questions from product notes",
        "system_prompt": (
            "List the facts in the product notes that matter for competition: "
            "pricing, features, gaps, customers, and each competitor type. "
            "End with open questions. Leave analysis and writing to later agents."
        ),
        "model_name": "gpt-5.4-mini",
        "max_loops": 1,
        "max_tokens": 4096,
        "temperature": 0.2,
    },
    {
        "agent_name": "Competitive Analyst",
        "description": "Compares the product with each competitor type",
        "system_prompt": (
            "Using the researcher's notes, compare the product with each "
            "competitor type on price, features, and target customer. "
            "Rank the three biggest risks. Leave the final write-up to the writer."
        ),
        "model_name": "gpt-5.4",
        "max_loops": 1,
        "max_tokens": 4096,
        "temperature": 0.3,
    },
    {
        "agent_name": "Brief Writer",
        "description": "Turns the analysis into a one-page brief",
        "system_prompt": (
            "Write a one-page competitive brief from the analysis: positioning, "
            "pricing comparison, risks, and three recommendations. Plain language. "
            "Use no figures that are missing from your input."
        ),
        "model_name": "claude-sonnet-5",
        "max_loops": 1,
        "max_tokens": 4096,
        "temperature": 0.5,
    },
]

payload = {
    "name": "Competitive Brief Pipeline",
    "description": "Research, analyze, then write a competitive brief",
    "swarm_type": "SequentialWorkflow",
    "max_loops": 1,
    "task": TASK,
    "agents": PIPELINE_AGENTS,
}


async def main():
    async with new_session() as session:
        result = await run_swarm(session, payload)
    print_turns(result["output"])
    print_usage(result)


if __name__ == "__main__":
    asyncio.run(main())

{
  "job_id": "swarm-...",
  "status": "success",
  "swarm_name": "Competitive Brief Pipeline",
  "description": "Research, analyze, then write a competitive brief",
  "swarm_type": "SequentialWorkflow",
  "output": [
    { "role": "Market Researcher", "content": "Facts: free for up to 5 clients, then $12 per month; invoices built from time logs... Open questions: how many users pass the 5-client limit?" },
    { "role": "Competitive Analyst", "content": "Accounting suite: covers payroll and inventory, heavier setup. Free generator: wins on price, no time-log invoicing... Top risks: 1. Payments platforms bundle invoicing..." },
    { "role": "Brief Writer", "content": "# Acme Invoice: Competitive Brief\n\n## Positioning\n..." }
  ],
  "number_of_agents": 3,
  "execution_time": ...,
  "usage": {
    "input_tokens": ...,
    "output_tokens": ...,
    "total_tokens": ...,
    "billing_info": { "cost_breakdown": { ... }, "total_cost": ... }
  }
}

# sdk_pipeline.py: step 2's pipeline through swarms-client (reuses common.py and step2_pipeline.py)
import os

from dotenv import load_dotenv
from swarms_client import SwarmsClient

from common import TASK
from step2_pipeline import PIPELINE_AGENTS

load_dotenv()

client = SwarmsClient(
    api_key=os.getenv("SWARMS_API_KEY"),
    base_url="https://api.swarms.world",
)

response = client.swarms.run(
    name="Competitive Brief Pipeline",
    description="Research, analyze, then write a competitive brief",
    swarm_type="SequentialWorkflow",
    max_loops=1,
    task=TASK,
    agents=PIPELINE_AGENTS,
)
print(response)

# step3_concurrent.py: three specialists draft sections in parallel (uses common.py)
import asyncio

from common import TASK, new_session, print_usage, run_swarm


def specialist(name, focus):
    return {
        "agent_name": name,
        "description": f"Drafts the {focus} section of a competitive brief",
        "system_prompt": (
            f"You are a {focus} specialist. Write only the {focus} section of a "
            "competitive brief, in under 250 words, using only the facts given."
        ),
        "model_name": "gpt-5.4-mini",
        "max_loops": 1,
        "max_tokens": 2048,
        "temperature": 0.4,
    }


SPECIALISTS = [
    specialist("Pricing Analyst", "pricing"),
    specialist("Positioning Analyst", "positioning"),
    specialist("Risk Analyst", "risk"),
]

payload = {
    "name": "Competitive Brief Specialists",
    "description": "Pricing, positioning, and risk drafts in parallel",
    "swarm_type": "ConcurrentWorkflow",
    "max_loops": 1,
    "task": TASK,
    "agents": SPECIALISTS,
}


async def main():
    async with new_session() as session:
        result = await run_swarm(session, payload)
    # Turns arrive in the order the agents finished, so look them up by role.
    drafts = {turn["role"]: turn["content"] for turn in result["output"]}
    for name in ("Positioning Analyst", "Pricing Analyst", "Risk Analyst"):
        print(f"\n=== {name} ===\n{drafts.get(name, '(no output)')}")
    print_usage(result)


if __name__ == "__main__":
    asyncio.run(main())

{
  "status": "success",
  "swarm_name": "Competitive Brief Specialists",
  "swarm_type": "ConcurrentWorkflow",
  "output": [
    { "role": "Risk Analyst", "content": "Risk 1: payments platforms include invoicing at no extra charge..." },
    { "role": "Pricing Analyst", "content": "Acme Invoice is free for up to 5 clients, then $12 per month..." },
    { "role": "Positioning Analyst", "content": "For freelancers who bill by the hour, Acme Invoice turns time logs into invoices..." }
  ],
  "number_of_agents": 3,
  "usage": { ... }
}

# step3_mixture.py: same specialists plus an aggregator (reuses common.py and step3_concurrent.py)
import asyncio

from common import TASK, new_session, print_usage, run_swarm
from step3_concurrent import SPECIALISTS

payload = {
    "name": "Competitive Brief Mixture",
    "description": "Specialist drafts merged into one brief",
    "swarm_type": "MixtureOfAgents",
    "max_loops": 1,
    "task": TASK,
    "agents": SPECIALISTS,
}


async def main():
    async with new_session() as session:
        result = await run_swarm(session, payload)
    merged = result["output"][-1]  # the aggregator's synthesis is the final turn
    print(f"=== {merged['role']} ===\n{merged['content']}")
    print_usage(result)


asyncio.run(main())

{
  "status": "success",
  "swarm_name": "Competitive Brief Mixture",
  "swarm_type": "MixtureOfAgents",
  "output": [
    { "role": "Pricing Analyst", "content": "..." },
    { "role": "Positioning Analyst", "content": "..." },
    { "role": "Risk Analyst", "content": "..." },
    { "role": "<aggregator>", "content": "# Acme Invoice: Competitive Brief\n\nPositioning: ...\nPricing: ...\nRisks: ...\nRecommendations: 1. ..." }
  ],
  "number_of_agents": 3,
  "usage": { ... }
}

# step4_hierarchical.py: a director delegates to four workers (uses common.py)
import asyncio

from common import TASK, new_session, print_turns, print_usage, run_swarm


def worker(name, description, prompt, model="gpt-5.4-mini"):
    return {
        "agent_name": name,
        "description": description,
        "system_prompt": prompt,
        "model_name": model,
        "max_loops": 1,
        "max_tokens": 4096,
        "temperature": 0.3,
    }


WORKERS = [
    worker(
        "Market Researcher",
        "Extracts facts, competitor types, and open questions from product notes",
        "You extract facts and open questions from product notes. You write no recommendations.",
    ),
    worker(
        "Pricing Analyst",
        "Compares pricing and packaging against each competitor type",
        "You compare pricing and packaging. Use only the prices given and label estimates.",
    ),
    worker(
        "Risk Analyst",
        "Identifies and ranks the biggest competitive risks",
        "You identify the three to five biggest competitive risks and rank them.",
    ),
    worker(
        "Brief Writer",
        "Writes the final one-page brief from the team's findings",
        "You write the final one-page competitive brief from the other agents' findings.",
        model="claude-sonnet-5",
    ),
]


def hierarchical_payload(task):
    return {
        "name": "Competitive Brief Team",
        "description": "A director plans the brief and delegates to specialists",
        "swarm_type": "HierarchicalSwarm",
        "max_loops": 1,
        "task": task,
        "agents": WORKERS,
        "director_model_name": "gpt-5.4",
        "director_settings": {"temperature": 0.2, "max_tokens": 8000},
        "list_all_agents": True,
    }


async def main():
    async with new_session() as session:
        result = await run_swarm(session, hierarchical_payload(TASK))
    print_turns(result["output"])
    print_usage(result)


if __name__ == "__main__":
    asyncio.run(main())

{
  "job_id": "swarm-...",
  "status": "success",
  "swarm_name": "Competitive Brief Team",
  "swarm_type": "HierarchicalSwarm",
  "output": [
    { "role": "Market Researcher", "content": "Facts from the notes: ... Missing: churn data, competitor prices..." },
    { "role": "Pricing Analyst", "content": "Against the free generator, Acme Invoice charges for volume (over 5 clients)..." },
    { "role": "Risk Analyst", "content": "1. Payments platforms bundle invoicing (high). 2. ..." },
    { "role": "Brief Writer", "content": "# Acme Invoice: Competitive Brief\n\n## Positioning\n..." }
  ],
  "number_of_agents": 4,
  "execution_time": ...,
  "usage": { "input_tokens": ..., "output_tokens": ..., "total_tokens": ..., "billing_info": { ... } }
}

# step5_auto_builder.py: generate a team, review it, then run it (uses common.py)
import asyncio

from common import TASK, new_session, post, print_turns, print_usage, run_swarm


async def main():
    async with new_session() as session:
        built = await post(
            session,
            "/v1/auto-agent-builder/completions",
            {"task": TASK, "max_agents": 4, "agent_kwargs": {"max_loops": 1, "max_tokens": 4096}},
        )

        for agent in built["agents"]:
            print(f"{agent['agent_name']} ({agent['model_name']}): {agent['description']}")

        if input("\nRun this team? [y/N] ").strip().lower() == "y":
            result = await run_swarm(session, {
                "name": "Auto-built Brief Team",
                "swarm_type": "HierarchicalSwarm",
                "director_model_name": "gpt-5.4",
                "max_loops": 1,
                "task": TASK,
                "agents": built["agents"],
            })
            print_turns(result["output"])
            print_usage(result)


asyncio.run(main())

{
  "job_id": "auto-agent-builder-...",
  "name": "auto-agent-builder",
  "status": "success",
  "agents": [
    { "agent_name": "CompetitorResearcher", "description": "Maps each competitor type...", "system_prompt": "You are...", "model_name": "gpt-5.4" },
    { "agent_name": "PricingStrategist", "description": "Compares pricing tiers...", "system_prompt": "You are...", "model_name": "gpt-5.4" },
    { "agent_name": "BriefWriter", "description": "Writes the final brief...", "system_prompt": "You are...", "model_name": "claude-sonnet-5" }
  ],
  "usage": { "input_tokens": ..., "output_tokens": ..., "total_tokens": ..., "total_cost": ..., "cost_per_agent": ... },
  "timestamp": "..."
}

# step5_rearrange.py: research, then pricing and risk in parallel, then write
# (reuses common.py and the WORKERS list from step4_hierarchical.py)
import asyncio

from common import TASK, new_session, print_turns, print_usage, run_swarm
from step4_hierarchical import WORKERS

payload = {
    "name": "Competitive Brief Flow",
    "description": "Research first, pricing and risk in parallel, then write",
    "swarm_type": "AgentRearrange",
    "rearrange_flow": "Market Researcher -> Pricing Analyst, Risk Analyst -> Brief Writer",
    "max_loops": 1,
    "task": TASK,
    "agents": WORKERS,
}


async def main():
    async with new_session() as session:
        result = await run_swarm(session, payload)
    print_turns(result["output"])
    print_usage(result)


asyncio.run(main())

# step6_structured.py: end the pipeline with a JSON scorecard (reuses common.py and step2_pipeline.py)
import asyncio
import json

from common import TASK, new_session, print_usage, run_swarm
from step2_pipeline import PIPELINE_AGENTS

SCORECARD_FORMAT = {
    "type": "json_schema",
    "json_schema": {
        "name": "competitive_scorecard",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "product": {"type": "string"},
                "threats": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "competitor": {"type": "string"},
                            "level": {"type": "string", "enum": ["low", "medium", "high"]},
                            "reason": {"type": "string"},
                        },
                        "required": ["competitor", "level", "reason"],
                        "additionalProperties": False,
                    },
                },
                "recommendations": {"type": "array", "items": {"type": "string"}},
            },
            "required": ["product", "threats", "recommendations"],
            "additionalProperties": False,
        },
    },
}

formatter = {
    "agent_name": "Scorecard Formatter",
    "description": "Converts the brief into a JSON scorecard",
    "system_prompt": "Convert the competitive brief you receive into the scorecard schema. Return only JSON.",
    "model_name": "gpt-4.1",
    "max_loops": 1,
    "max_tokens": 2048,
    "temperature": 0.0,
    "llm_args": {"response_format": SCORECARD_FORMAT},
}


async def main():
    async with new_session() as session:
        result = await run_swarm(session, {
            "name": "Competitive Brief Pipeline (JSON)",
            "swarm_type": "SequentialWorkflow",
            "max_loops": 1,
            "task": TASK,
            "agents": PIPELINE_AGENTS + [formatter],
        })

    scorecard = json.loads(result["output"][-1]["content"])  # content is a JSON string
    print(json.dumps(scorecard, indent=2))
    print_usage(result)


asyncio.run(main())

{
  "product": "Acme Invoice",
  "threats": [
    { "competitor": "Payments platform with built-in invoicing", "level": "high", "reason": "..." },
    { "competitor": "Free invoice generator", "level": "medium", "reason": "..." },
    { "competitor": "Small-business accounting suite", "level": "low", "reason": "..." }
  ],
  "recommendations": ["...", "...", "..."]
}

# step6_cost.py: where the money went in one swarm run (reuses common.py and step2_pipeline.py)
import asyncio

from common import new_session, run_swarm
from step2_pipeline import payload


async def main():
    async with new_session() as session:
        result = await run_swarm(session, payload)
    usage = result["usage"]
    billing = usage["billing_info"]
    breakdown = billing["cost_breakdown"]

    print(f"time:          {result['execution_time']}s")
    print(f"input tokens:  {usage['input_tokens']}")
    print(f"output tokens: {usage['output_tokens']}")
    print(f"agent fees:    ${breakdown['agent_cost']}")
    print(f"input cost:    ${breakdown['input_token_cost']}")
    print(f"output cost:   ${breakdown['output_token_cost']}")
    print(f"total:         ${billing['total_cost']}")
    if billing.get("discount_active"):
        print(f"discount:      {billing['discount_type']} ({billing['discount_percentage']}%)")


asyncio.run(main())

{
  "input_tokens": ...,
  "output_tokens": ...,
  "total_tokens": ...,
  "billing_info": {
    "cost_breakdown": {
      "agent_cost": ...,
      "input_token_cost": ...,
      "output_token_cost": ...,
      "token_counts": { "total_input_tokens": ..., "total_output_tokens": ..., "total_tokens": ... },
      "num_agents": 3,
      "night_time_discount_applied": false
    },
    "total_cost": ...,
    "discount_active": false,
    "discount_type": "none",
    "discount_percentage": 0
  }
}

# retry.py: retry 429s using the Retry-After header (reuses common.py)
import asyncio

from common import BASE_URL


async def post_with_retry(session, path, payload, max_retries=3):
    for attempt in range(max_retries + 1):
        async with session.post(f"{BASE_URL}{path}", json=payload) as resp:
            if resp.status == 429 and attempt < max_retries:
                wait = int(resp.headers.get("Retry-After", 60))
            elif resp.status >= 400:
                body = await resp.text()
                raise RuntimeError(f"{path} returned {resp.status}: {body}")
            else:
                remaining = resp.headers.get("X-RateLimit-Remaining-Minute")
                if remaining is not None:
                    print(f"requests left this minute: {remaining}")
                return await resp.json()
        print(f"Rate limited, retrying in {wait}s")
        await asyncio.sleep(wait)

# step6_batch.py: one brief per notes file in a single request
# (premium plans; reuses common.py and step4_hierarchical.py)
import asyncio
import json
from pathlib import Path

from common import make_task, new_session, post
from step4_hierarchical import hierarchical_payload


async def main():
    notes_files = sorted(Path("product_notes").glob("*.txt"))[:50]  # 50 specs per request max
    batch = []
    for path in notes_files:
        spec = hierarchical_payload(make_task(path.read_text()))
        spec["name"] = f"Brief: {path.stem}"
        batch.append(spec)

    async with new_session() as session:
        results = await post(session, "/v1/swarm/batch/completions", batch)

    out = Path("briefs")
    out.mkdir(exist_ok=True)
    for path, item in zip(notes_files, results):
        if item["status"] == "success":
            (out / f"{path.stem}.json").write_text(json.dumps(item["result"], indent=2))
            print(f"{path.stem}: ${item['usage']['billing_info']['total_cost']}")
        else:
            print(f"{path.stem}: failed: {item.get('detail')}")


asyncio.run(main())

[
  { "status": "success", "swarm_name": "Brief: acme-invoice", "result": { ... }, "usage": { "billing_info": { "total_cost": ... } } },
  { "status": "error", "swarm_name": "Brief: other-product", "detail": "Failed to run swarm: ..." }
]

# step6_parallel.py: free tier, one swarm request per notes file, a few at a time
# (reuses common.py, retry.py and step4_hierarchical.py)
import asyncio
import json
from pathlib import Path

from common import make_task, new_session
from retry import post_with_retry
from step4_hierarchical import hierarchical_payload

MAX_IN_FLIGHT = 4  # swarm runs in progress at once; retry.py handles any 429


async def run_brief(session, semaphore, path):
    async with semaphore:
        payload = hierarchical_payload(make_task(path.read_text()))
        return await post_with_retry(session, "/v1/swarm/completions", payload)


async def main():
    notes_files = sorted(Path("product_notes").glob("*.txt"))
    semaphore = asyncio.Semaphore(MAX_IN_FLIGHT)
    async with new_session() as session:
        results = await asyncio.gather(
            *(run_brief(session, semaphore, path) for path in notes_files),
            return_exceptions=True,  # collect failures instead of raising on the first one
        )

    out = Path("briefs")
    out.mkdir(exist_ok=True)
    for path, result in zip(notes_files, results):
        if isinstance(result, Exception):
            print(f"{path.stem}: failed: {result}")
        else:
            (out / f"{path.stem}.json").write_text(json.dumps(result["output"], indent=2))
            print(f"{path.stem}: ${result['usage']['billing_info']['total_cost']}")


asyncio.run(main())