如何在 Python 中并行运行 100 个 AI 智能体

pip install -U swarms
# optional: RouteHub for model calls and rustworkx for graphs
pip install -U "swarms[fast]"

import time

from swarms import Agent, ConcurrentWorkflow

# Cheap, fast models from four providers, 25 agents each.
models = [
    "gpt-5.4-mini",
    "anthropic/claude-haiku-5-5",
    "groq/openai/gpt-oss-120b",
    "mistral/mistral-large-latest",
]
sectors = ["energy", "banking", "retail", "logistics", "healthcare"]

agents = [
    Agent(
        agent_name=f"Analyst-{n:03d}",
        system_prompt=(
            f"You are an equity analyst covering {sectors[n % len(sectors)]}. "
            "Answer in two sentences."
        ),
        model_name=models[n % len(models)],
        max_loops=1,
        print_on=False,
    )
    for n in range(100)
]

workflow = ConcurrentWorkflow(agents=agents)  # max_workers defaults to len(agents)

start = time.perf_counter()
results = workflow.run("What is the biggest risk to your sector next quarter?")
print(f"{len(results)} answers in {time.perf_counter() - start:.2f} s")

for message in results[:3]:
    print(message["role"], "->", message["content"][:80])

import time

from swarms import Agent, GraphWorkflow

def make_agent(name, model, prompt):
    return Agent(
        agent_name=name,
        system_prompt=prompt,
        model_name=model,
        max_loops=1,
        output_type="final",  # record each node's answer, not its transcript
        print_on=False,
    )

brief = make_agent(
    "Brief-Writer", "gpt-5.4",
    "Turn the task into a short brief that a reviewer can act on.",
)
reviewers = [
    make_agent(
        f"Reviewer-{n:03d}",
        ["gpt-5.4-mini", "anthropic/claude-haiku-5-5", "groq/openai/gpt-oss-120b"][n % 3],
        f"You review product feedback as customer persona #{n}. Give one concrete objection.",
    )
    for n in range(100)
]
editor = make_agent(
    "Editor", "anthropic/claude-sonnet-5-5",
    "Group the reviewers' objections into themes and rank them by how often they appear.",
)

graph = GraphWorkflow(name="persona-review")  # max_parallel_nodes=None: full layer width
for agent in [brief, *reviewers, editor]:
    graph.add_node(agent)
graph.add_edges_from_source("Brief-Writer", [r.agent_name for r in reviewers])
graph.add_edges_to_target([r.agent_name for r in reviewers], "Editor")

start = time.perf_counter()
outputs = graph.run("Launch plan: a $9/month tier with no API access.")
print(f"{len(outputs)} nodes in {time.perf_counter() - start:.2f} s")
print(outputs["Editor"][:200])
print(graph.usage)

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

from swarms import Agent, run_agents_concurrently_async

agents = [
    Agent(
        agent_name=f"Summariser-{n:03d}",
        system_prompt="Summarise the task in one sentence.",
        model_name=["gpt-5.4-mini", "groq/openai/gpt-oss-120b"][n % 2],
        max_loops=1,
        output_type="final",
        print_on=False,
    )
    for n in range(100)
]

async def main():
    # Agent.arun runs Agent.run in a worker thread (asyncio.to_thread), and the
    # event loop's default pool has min(32, CPU cores + 4) threads. Size it to the fan-out.
    asyncio.get_running_loop().set_default_executor(ThreadPoolExecutor(max_workers=len(agents)))

    for cap in (None, 20):
        start = time.perf_counter()
        results = await run_agents_concurrently_async(
            agents, "Summarise today's market news.", max_concurrency=cap
        )
        print(f"max_concurrency={cap}: {len(results)} results in {time.perf_counter() - start:.2f} s")

    print(results[0])

asyncio.run(main())

import asyncio
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor

from swarms import Agent, run_agents_concurrently_async

# How many requests each provider account can take at once. Set these from
# your own rate-limit tier; the numbers here are placeholders.
PROVIDER_CAPS = {
    "gpt-5.4-mini": 40,
    "anthropic/claude-haiku-5-5": 10,
    "groq/openai/gpt-oss-120b": 25,
    "deepseek/deepseek-v4-pro": 15,
}
# A backup model on a different provider for each primary.
FALLBACK = {
    "gpt-5.4-mini": "groq/openai/gpt-oss-120b",
    "anthropic/claude-haiku-5-5": "gpt-5.4-mini",
    "groq/openai/gpt-oss-120b": "gpt-5.4-mini",
    "deepseek/deepseek-v4-pro": "anthropic/claude-haiku-5-5",
}
models = list(PROVIDER_CAPS)

agents = [
    Agent(
        agent_name=f"Grader-{n:03d}",
        system_prompt="Grade the answer from 1 to 10 and give one reason.",
        model_name=models[n % len(models)],
        fallback_model_name=FALLBACK[models[n % len(models)]],
        retry_attempts=3,  # client-side retries per model, with backoff on rate limits
        max_loops=1,
        output_type="final",
        print_on=False,
    )
    for n in range(100)
]

async def main():
    asyncio.get_running_loop().set_default_executor(ThreadPoolExecutor(max_workers=len(agents)))

    by_model = defaultdict(list)
    for agent in agents:
        by_model[agent.model_name].append(agent)

    task = "Answer to grade: 'Water boils at 100 C at sea level.'"
    groups = list(by_model.items())
    results = await asyncio.gather(*(
        run_agents_concurrently_async(group, task, max_concurrency=PROVIDER_CAPS[model])
        for model, group in groups
    ))

    for (model, group), answers in zip(groups, results):
        print(f"{model}: {len(answers)} answers, at most {PROVIDER_CAPS[model]} in flight")

asyncio.run(main())

from swarms import Agent, batched_grid_agent_execution

samples = [
    "The checkout page froze twice before my order went through.",
    "Arrived a day early and the packaging was perfect.",
    "Twelve dollars a month is too much for what it does.",
    "My parcel has been stuck in the depot for a week.",
]
reviews = [f"Review #{n}: {samples[n % len(samples)]}" for n in range(100)]

agents = [
    Agent(
        agent_name=f"Tagger-{n:03d}",
        system_prompt="Tag the review with one of: bug, praise, pricing, shipping. Reply with the tag only.",
        model_name="gpt-5.4-mini" if n % 2 else "anthropic/claude-haiku-5-5",
        max_loops=1,
        output_type="final",
        print_on=False,
    )
    for n in range(100)
]

# One task per agent. max_workers defaults to 95% of CPU cores here, so set it.
tags = batched_grid_agent_execution(agents, reviews, max_workers=len(agents))

failed = [t for t in tags if isinstance(t, Exception)]
print(len(tags), "tagged,", len(failed), "failed")
print(tags[0])

import time

from swarms import Agent, ConcurrentWorkflow, GraphWorkflow

agents = [
    Agent(
        agent_name=f"Worker-{n:03d}",
        model_name="anthropic/claude-haiku-5-5",
        max_loops=1,
        output_type="final",
        print_on=False,
    )
    for n in range(100)
]

# At most 10 requests in flight, for a provider tier that allows little concurrency.
throttled = ConcurrentWorkflow(agents=agents, max_workers=10)
start = time.perf_counter()
throttled.run("Name one prime number.")
print(f"max_workers=10: {time.perf_counter() - start:.2f} s")

graph = GraphWorkflow(max_parallel_nodes=10)  # the same cap for a graph's widest layer
for agent in agents:
    graph.add_node(agent)
start = time.perf_counter()
outputs = graph.run("Name one prime number.")
print(f"max_parallel_nodes=10: {len(outputs)} nodes in {time.perf_counter() - start:.2f} s")

pip install -U swarms