LLM 智能体的编排者-工作者模式：扇出到数百个工作者

pip install -U swarms
# optional: RouteHub for lower per-call overhead, rustworkx for graphs
pip install -U "swarms[fast]"

from swarms.structs.phased_fan_out_workflow import PhasedFanOutWorkflow

FILES = {
    "billing.py": """
def apply_discount(price, percent):
    # percent is 0-100, e.g. 15 for 15%
    return price - price * percent
""",
    "dates.py": """
from datetime import datetime

def days_between(start, end):
    delta = datetime.fromisoformat(end) - datetime.fromisoformat(start)
    return delta.seconds // 86400
""",
    "cache.py": """
def get_user(cache, user_id, load):
    if user_id in cache:
        return cache[user_id]
    user = load(user_id)
    return user
""",
}

task = (
    "Find every bug in these files, verify each one independently, "
    "and report only the confirmed bugs with a one-line fix.\n\n"
    + "\n\n".join(f"=== {name} ===\n{code}" for name, code in FILES.items())
)

workflow = PhasedFanOutWorkflow(
    model_name="anthropic/claude-opus-5-5",  # lead: plans and combines
    worker_model_name="gpt-5.4-mini",        # every inline worker agent
    size_guideline="small",
)

print(workflow.run(task))

plan = workflow.last_plan
for phase in plan.phases:
    for step in phase.steps:
        fan = f" (one agent per item of {step.for_each})" if step.for_each else ""
        print(f"{phase.title}: {step.id} -> {step.agent}{fan}")

run = workflow.last_run
print("worker agents:", run["agent_count"])
print("failed slots:", run["failures"])
print("verdicts:", [v["verdict"] for v in run["outputs"]["verify"]])
print("tokens:", workflow.usage)

from swarms import Agent
from swarms.structs.phased_fan_out_workflow import PhasedFanOutWorkflow

reader = Agent(
    agent_name="Clause-Reader",
    agent_description="Reads one contract and quotes every change-of-control clause.",
    system_prompt=(
        "You read a single contract and quote, word for word, every clause "
        "triggered by a change of control. If there is none, return an empty list."
    ),
    model_name="deepseek/deepseek-v4-pro",
    max_loops=1,
)
checker = Agent(
    agent_name="Clause-Checker",
    agent_description="Independently checks whether a quoted clause is a change-of-control clause.",
    system_prompt=(
        "You are a skeptical contracts lawyer. Decide whether the quoted clause "
        "is really triggered by a change of control, and explain why."
    ),
    model_name="mistral/mistral-large-latest",
    max_loops=1,
)

contracts = {
    "acme_supply.txt": (
        "12.1 Either party may terminate this Agreement on 30 days notice. "
        "12.4 If Supplier undergoes a change of control, Buyer may terminate immediately."
    ),
    "globex_license.txt": (
        "8.2 This license may not be assigned without consent, including by merger "
        "or sale of a majority of the Licensee's shares."
    ),
}
task = "Quote every change-of-control clause in these contracts and confirm each one.\n\n" + "\n\n".join(
    f"=== {name} ===\n{text}" for name, text in contracts.items()
)

workflow = PhasedFanOutWorkflow(
    model_name="gpt-5.4",
    agents=[reader, checker],
    allow_inline_agents=False,
    size_guideline="small",
)
print(workflow.run(task))
print(sorted({agent.model_name for agent in workflow.last_run["agents"]}))

import random

from swarms.structs.phased_fan_out_workflow import (
    PhasedFanOutWorkflow,
    PhasedFanOutWorkflowError,
)

# A stand-in for a support-queue export: 200 one-line tickets.
random.seed(7)
issues = [
    "Card was charged twice this month",
    "Cannot log in after password reset",
    "CSV export times out on large projects",
    "Please add a dark mode",
    "Invoice shows the wrong VAT rate",
    "App crashes when uploading a photo",
]
tickets = [f"T-{n:03d}: {random.choice(issues)}" for n in range(1, 201)]

task = (
    "Triage every ticket below. Give each one a category (billing, auth, bug, "
    "feature) and an urgency (low, medium, high), then summarise the queue.\n\n"
    + "\n".join(tickets)
)

workflow = PhasedFanOutWorkflow(
    model_name="gemini/gemini-2.5-pro",           # lead
    worker_model_name="groq/openai/gpt-oss-120b",  # 200 cheap, fast workers
    size_guideline="unrestricted",
    max_agents=250,             # hard ceiling on workers per run
    max_concurrent_agents=64,   # workers calling the model at once (1 to 256)
    max_retries=1,              # one retry per worker, with a fresh agent
    max_plan_attempts=3,
)

try:
    print(workflow.run(task))
except PhasedFanOutWorkflowError as error:
    print("run stopped:", error)

print("workers:", workflow.last_run["agent_count"])
print("failed slots:", len(workflow.last_run["failures"]))

from swarms.structs.phased_fan_out_workflow import (
    PhasedFanOutWorkflow,
    WorkflowPlan,
)

workflow = PhasedFanOutWorkflow(model_name="gpt-5.4", worker_model_name="gpt-5.4-mini")

bad_plan = WorkflowPlan.model_validate({
    "name": "broken",
    "description": "A plan with several mistakes.",
    "inline_agents": [{"name": "worker", "system_prompt": "You do one unit of work."}],
    "phases": [
        {"title": "List", "steps": [
            {"id": "list", "agent": "worker", "prompt": "List the sections of {task}.",
             "output_schema": {"type": "object", "properties": {"sections": {"type": "array"}}}},
        ]},
        {"title": "Work", "steps": [
            {"id": "summarise", "agent": "worker", "prompt": "Summarise {item}."},
            {"id": "check", "agent": "worker", "for_each": "list.pages", "prompt": "Check {item}."},
            {"id": "polish", "agent": "editor", "prompt": "Polish the summary."},
        ]},
    ],
    "return_steps": ["summarise", "check", "polish"],
})

for problem in workflow.validate_plan(bad_plan):
    print("-", problem)

pip install -U swarms