什么是多智能体编排？模式、状态与故障处理

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"}

CHANGELOG = """
- Added CSV export to the reports page
- Fixed a bug where saved filters were lost after logout
- Deprecated the v1 webhooks endpoint; v2 replaces it
"""

swarm_config = {
    "name": "Release Notes Pipeline",
    "description": "Extract changes, write release notes, then edit them",
    "swarm_type": "SequentialWorkflow",
    "task": f"Write customer-facing release notes for this changelog:\n{CHANGELOG}",
    "max_loops": 1,
    "agents": [
        {
            "agent_name": "Change Extractor",
            "description": "Turns a raw changelog into a structured list",
            "system_prompt": (
                "List every change as: type (feature, fix, deprecation), "
                "what changed, and who is affected. Output only the list."
            ),
            "model_name": "gpt-5.4-mini",
            "max_loops": 1,
            "max_tokens": 2000,
            "temperature": 0.2,
        },
        {
            "agent_name": "Release Writer",
            "description": "Writes release notes from the structured list",
            "system_prompt": (
                "Write short release notes from the list you receive. Group them "
                "by type and explain each change in one or two plain sentences."
            ),
            "model_name": "claude-sonnet-5",
            "max_loops": 1,
            "max_tokens": 3000,
            "temperature": 0.5,
        },
        {
            "agent_name": "Editor",
            "description": "Tightens the notes for publication",
            "system_prompt": (
                "Edit the release notes you receive for clarity and length. "
                "Remove marketing language. Return only the final notes."
            ),
            "model_name": "gpt-5.4",
            "max_loops": 1,
            "max_tokens": 3000,
            "temperature": 0.3,
        },
    ],
}


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:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=swarm_config) as resp:
            resp.raise_for_status()
            result = await resp.json()

    # One turn per agent, in pipeline order: the list is an execution trace.
    for turn in result["output"]:
        print(f"--- {turn['role']}\n{turn['content']}\n")

    usage = result["usage"]
    cost = usage.get("billing_info", {}).get("total_cost")
    print(f"job {result['job_id']}: {result['execution_time']:.1f}s, "
          f"{usage['total_tokens']} tokens, ${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"}

REVIEWER_RULE = "Answer only the question addressed to you in the scope, in at most 200 words."

swarm_config = {
    "name": "Vendor Evaluation",
    "description": "Scope the decision, review it three ways in parallel, then recommend",
    "swarm_type": "AgentRearrange",
    # "->" hands off to the next step; commas run agents in parallel within a step.
    "rearrange_flow": "Scoper -> Security, Pricing, Integration -> Recommender",
    "task": (
        "Should we move our support search to a hosted vector database? "
        "We index 2 million help-center documents, need SOC 2 compliance, "
        "and have a budget of $3,000 per month."
    ),
    "max_loops": 1,
    "agents": [
        {
            "agent_name": "Scoper",
            "system_prompt": (
                "Restate the decision and its constraints in five bullet points, "
                "then write one question each for Security, Pricing and Integration."
            ),
            "model_name": "gpt-5.4-mini",
            "max_tokens": 1500,
        },
        {
            "agent_name": "Security",
            "system_prompt": f"You review security and compliance risk. {REVIEWER_RULE}",
            "model_name": "claude-sonnet-5",
            "max_tokens": 1500,
        },
        {
            "agent_name": "Pricing",
            "system_prompt": f"You review cost at the stated scale. {REVIEWER_RULE}",
            "model_name": "gpt-5.4-mini",
            "max_tokens": 1500,
        },
        {
            "agent_name": "Integration",
            "system_prompt": f"You review migration and integration effort. {REVIEWER_RULE}",
            "model_name": "gpt-5.4-mini",
            "max_tokens": 1500,
        },
        {
            "agent_name": "Recommender",
            "system_prompt": (
                "Read the three reviews. Recommend adopt, trial or reject, give the "
                "two reasons that matter most, and list the open questions."
            ),
            "model_name": "gpt-5.4",
            "max_tokens": 2000,
        },
    ],
}


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:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=swarm_config) as resp:
            resp.raise_for_status()
            result = await resp.json()

    # Key turns by agent name: parallel agents can finish in any order.
    by_agent = {turn["role"]: turn["content"] for turn in result["output"]}
    for name in ("Security", "Pricing", "Integration"):
        print(f"--- {name}\n{by_agent.get(name, '(no output)')}\n")
    print("=== Recommendation\n", by_agent.get("Recommender", "(no output)"))


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"}


async def run_swarm(session: aiohttp.ClientSession, spec: dict, attempts: int = 3) -> dict:
    """Run one swarm. Retry 429 and 500 with backoff; raise on anything else."""
    for attempt in range(1, attempts + 1):
        # A request that outlives the session timeout raises asyncio.TimeoutError,
        # which is not retried here.
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=spec) as resp:
            if resp.status == 200:
                return await resp.json()
            body = await resp.text()
            retry_after = resp.headers.get("Retry-After")
        if resp.status in (429, 500) and attempt < attempts:
            await asyncio.sleep(int(retry_after or 2 ** attempt))
            continue
        # 401, 402, 403 and 422 fail the same way on a retry, so stop here.
        raise RuntimeError(
            f"{spec['swarm_type']} failed with HTTP {resp.status}: {body[:300]}"
        )


def checked_turns(result: dict) -> list:
    """Fail loudly instead of passing an empty run downstream."""
    output = result.get("output")
    turns = [t for t in output if isinstance(t, dict)] if isinstance(output, list) else []
    empty = [t.get("role") for t in turns if not str(t.get("content", "")).strip()]
    if result.get("status") != "success" or not turns or empty:
        raise ValueError(
            f"unusable output: status={result.get('status')}, "
            f"turns={len(turns)}, empty={empty}"
        )
    return turns


TASK = (
    "Write a one-page market brief on AI code review tools for a seed-stage "
    "startup deciding whether to enter that market."
)

workers = [
    {
        "agent_name": "Market Researcher",
        "system_prompt": (
            "Research buyers, competitors and trends for the question you are "
            "given. Flag anything you are unsure of."
        ),
        "model_name": "claude-sonnet-5",
        "fallback_models": ["gpt-5.4"],
        "max_tokens": 3000,
    },
    {
        "agent_name": "Pricing Analyst",
        "system_prompt": "Analyze pricing models and unit economics for the question you are given.",
        "model_name": "gpt-5.4",
        "fallback_models": ["claude-sonnet-5"],
        "max_tokens": 3000,
    },
    {
        "agent_name": "Brief Writer",
        "system_prompt": "Write a one-page brief from the findings you are given. End with open questions.",
        "model_name": "claude-sonnet-5",
        "fallback_models": ["gpt-5.4"],
        "max_tokens": 4000,
    },
]

spec = {
    "name": "Market Brief",
    "swarm_type": "HierarchicalSwarm",
    "task": TASK,
    "agents": workers,
    "max_loops": 2,  # up to two rounds of plan, delegate and review
    "director_model_name": "gpt-5.4",
    "director_settings": {"temperature": 0.2, "max_tokens": 8000},
}


async def main() -> None:
    # Applies to each request on the session; a two-loop hierarchy can take minutes.
    timeout = aiohttp.ClientTimeout(total=900)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        # 1. Run the hierarchy; if it fails, fall back to a fixed pipeline of the same agents.
        try:
            result = await run_swarm(session, spec)
            turns = checked_turns(result)
        except (RuntimeError, ValueError, asyncio.TimeoutError) as err:
            print(f"HierarchicalSwarm failed ({err!r}); falling back to SequentialWorkflow")
            fallback = {**spec, "swarm_type": "SequentialWorkflow", "max_loops": 1}
            result = await run_swarm(session, fallback)
            turns = checked_turns(result)

        by_role = {t["role"]: t["content"] for t in turns}
        brief = by_role.get("Brief Writer", turns[-1]["content"])

        # 2. Gate the brief with a council of judges before anyone reads it.
        review = await run_swarm(session, {
            "name": "Brief Review",
            "swarm_type": "CouncilAsAJudge",
            "council_judge_model_name": "gpt-5.4",
            "task": f"Task: {TASK}\n\nResponse: {brief}",
            # Required by the API, billed, and never executed by CouncilAsAJudge.
            "agents": [{"agent_name": "placeholder", "model_name": "gpt-5.4-mini", "max_loops": 1}],
            "max_loops": 1,
        })
        verdict = checked_turns(review)[-1]["content"]  # the aggregator's ruling comes last

    for run in (result, review):
        cost = (run.get("usage") or {}).get("billing_info", {}).get("total_cost")
        print(f"{run['swarm_type']} {run['job_id']}: {run['execution_time']:.1f}s, ${cost}")
    print("\nJudge verdict:\n", verdict)


asyncio.run(main())

{
  "usage": {
    "input_tokens": 420,
    "output_tokens": 1850,
    "total_tokens": 2270,
    "billing_info": {
      "cost_breakdown": {
        "agent_cost": 0.02,
        "input_token_cost": 0.00137,
        "output_token_cost": 0.01713,
        "num_agents": 2
      },
      "total_cost": 0.0385
    }
  }
}

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"}


async def get_json(session: aiohttp.ClientSession, path: str) -> dict:
    async with session.get(f"{BASE_URL}{path}") as resp:
        if resp.status >= 400:
            body = await resp.text()
            raise RuntimeError(f"GET {path} failed with HTTP {resp.status}: {body}")
        return await resp.json()


async def main() -> None:
    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        # The three reads are independent, so send them at once over one session.
        metrics, logs, pricing = await asyncio.gather(
            get_json(session, "/v1/account/metrics/summary"),
            get_json(session, "/v1/account/logs"),
            get_json(session, "/v1/usage/costs"),
        )

    print(f"{metrics['successful_completions']} of {metrics['total_completion_calls']} completions "
          f"succeeded; {metrics['completions_last_24h']} in the last 24 hours")

    print(f"{logs['count']} logged requests, newest first:")
    for entry in logs["logs"][:5]:
        print(entry)

    print("per-agent fee:", pricing["usage_pricing"]["swarm_completions_agent_cost"])


asyncio.run(main())