PhasedFanOutWorkflow Examples

pip install -U swarms
export OPENAI_API_KEY="sk-..."

from swarms.structs.phased_fan_out_workflow import PhasedFanOutWorkflow

CODE = {
    "stats.py": "def mean(xs):\n    return sum(xs) / len(xs) - 1\n",
    "users.py": "def find(users, name):\n    for u in users:\n        if u['name'] = name:\n            return u\n",
    "cache.py": "def get(cache, key, default=[]):\n    default.append(key)\n    return cache.get(key, default)\n",
}

task = (
    "Find every bug in these files. Plan one agent per file to find bugs, "
    "then one independent agent per reported bug to verify it, and report "
    "only verified bugs.\n\n"
    + "\n\n".join(f"# {path}\n{code}" for path, code in CODE.items())
)

workflow = PhasedFanOutWorkflow(
    model_name="gpt-5.4",              # Lead: plans and combines
    worker_model_name="gpt-5.4-mini",  # Inline agents the plan defines
    size_guideline="small",
)
print(workflow.run(task))
print(workflow.usage)

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

print("Workers started:", workflow.last_run["agent_count"])
print("Failures:", workflow.last_run["failures"])

for step_id, output in workflow.last_run["outputs"].items():
    print(step_id, "->", str(output)[:120])

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

CONTRACTS = {
    "acme_supply.txt": "If Supplier undergoes a change of control, Buyer may terminate this Agreement on 30 days' notice.",
    "globex_license.txt": "This license is non-transferable. Licensee may not assign this license without prior written consent.",
    "initech_services.txt": "Either party may terminate for convenience with 90 days' notice. Fees are payable monthly in arrears.",
    "umbrella_lease.txt": "A merger, acquisition or sale of a majority of Tenant's shares is deemed an assignment requiring Landlord's consent.",
}

reader = Agent(
    agent_name="Clause-Reader",
    agent_description="Reads one contract and quotes any clause triggered by a change of ownership or control.",
    system_prompt="You read a single contract and quote, word for word, every clause that is triggered by a change of control, merger, acquisition or sale of shares. If there is none, say so.",
    model_name="gpt-5.4-mini",
    max_loops=1,
)

checker = Agent(
    agent_name="Clause-Checker",
    agent_description="Independently checks whether a quoted clause really is a change-of-control clause.",
    system_prompt="You are a skeptical contracts lawyer. Given a contract and a quoted clause, decide whether the clause is triggered by a change of control: a merger, an acquisition, or a sale of a controlling stake. A clause that treats such an event as an assignment counts. A plain ban on assignment that never mentions a change of ownership does not. Give a verdict and one sentence of reasoning.",
    model_name="gpt-5.4-mini",
    max_loops=1,
)

task = (
    "Which of these contracts contain a change-of-control clause? "
    "Read each contract separately, then have every candidate clause checked independently.\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))

workflow = PhasedFanOutWorkflow(
    model_name="gpt-5.4",
    worker_model_name="gpt-5.4-mini",
    size_guideline="large",      # Guidance: aim for fewer than 50 agents
    max_agents=60,               # Enforced: a fan-out past 60 workers stops before it starts
    max_concurrent_agents=8,     # At most 8 agents at once within a phase
    max_retries=1,               # One retry per failed agent
    max_plan_attempts=2,         # The lead gets one repair after a rejected plan
)

from swarms.structs.phased_fan_out_workflow import PhasedFanOutWorkflowError

try:
    answer = workflow.run(task)
except PhasedFanOutWorkflowError as error:
    print("Run stopped:", error)
    if workflow.last_plan:
        print("Planned phases:", [p.title for p in workflow.last_plan.phases])
    print("Failures so far:", workflow.last_run["failures"])

workflow = PhasedFanOutWorkflow(model_name="gpt-5.4", output_type="dict")
for row in workflow.run(task):
    print(f"[{row['role']}] {str(row['content'])[:100]}")