升级到 Swarms v17：破坏性变更与迁移指南

# see what you have
python -c "import swarms; print(swarms.__version__)"

# stay on v16 while you migrate (16.0.4 was the last v16 point release)
pip install "swarms==16.0.4"

# then, on a branch, move to v17
pip install -U "swarms==17.0.0"
# or with RouteHub and rustworkx, the optional speedups
pip install -U "swarms[fast]==17.0.0"

grep -rnE "SwarmRearrange|swarm_arrange|create_agents_from_yaml|CSVAgentLoader|MarkdownAgentLoader|MarkdownAgentConfig|AgentConfigDict|AgentValidator|ModelName|image_batch_processor|ImageAgentBatchProcessor|hierarchical_structured_communication_framework|Transcript|log_agent_data|generate_conversation_id|MAX_CONCURRENT_AGENTS|bulk_run|check_available_tokens|get_available_models|get_current_model|switch_to_next_model|reset_model_index|is_fallback_available|multi_modal=|sop=|sop_list=|output_cleaner|agents_config=|memory_system=|shared_memory=|idle_timeout=|auto_generate_prompts=|metadata_output_path=|output_metadata=|correct_answer=|basemodel|mcp_url=" --include="*.py" .

from swarms import SwarmRearrange

pipeline = SwarmRearrange(swarms=[research_team, review_team], flow="Research-Team -> Review-Team")

from swarms import Agent, AgentRearrange, ConcurrentWorkflow

def make(name, model, prompt):
    return Agent(agent_name=name, model_name=model, system_prompt=prompt, max_loops=1, print_on=False)

planner = make("Planner", "anthropic/claude-sonnet-5-5", "Break the question into two research angles.")
analyst_a = make("Supply-Analyst", "groq/openai/gpt-oss-120b", "Research the supply side.")
analyst_b = make("Demand-Analyst", "gpt-5.4-mini", "Research the demand side.")
editor = make("Editor", "anthropic/claude-sonnet-5-5", "Merge the research into a five-line brief.")

# A whole swarm can be one step. This output_type returns each analyst's answer, without the task.
research_team = ConcurrentWorkflow(
    name="Research-Team", agents=[analyst_a, analyst_b], output_type="str-all-except-first",
)

pipeline = AgentRearrange(
    agents=[planner, research_team, editor],
    flow="Planner -> Research-Team -> Editor",
    output_type="final",
)
print(pipeline.run("What is the HBM memory supply outlook for 2027?"))

from swarms import create_agents_from_yaml
from swarms.structs.agent_loader import CSVAgentLoader

agents = create_agents_from_yaml("team.yaml", return_type="agents")
csv_agents = CSVAgentLoader("team.csv").load_agents()

from pathlib import Path

from swarms import AgentLoader, load_agent_from_markdown

Path("team.yaml").write_text("""
agents:
  - agent_name: Researcher
    model_name: gemini/gemini-2.5-pro
    system_prompt: You research the question and list the key facts.
    max_loops: 1
  - agent_name: Writer
    model_name: anthropic/claude-haiku-5-5
    system_prompt: You turn research notes into a short, plain answer.
    max_loops: 1
swarm_architecture:
  name: Research-Pipeline
  swarm_type: SequentialWorkflow
  output_type: final
  task: Why do bond prices fall when interest rates rise?
""")

Path("reviewer.md").write_text("""---
name: Reviewer
model_name: gpt-5.4
max_loops: 1
---
You review an answer for factual errors and reply with a corrected version.
""")

loader = AgentLoader()

agents = loader.load_agents_from_yaml("team.yaml", return_type="agents")
print([a.agent_name for a in agents])

answer = loader.load_agents_from_yaml("team.yaml", return_type="run_swarm")  # builds the SwarmRouter and runs its task
print(answer)

reviewer = load_agent_from_markdown("reviewer.md")
print(reviewer.agent_name, reviewer.model_name)

tags = agent.bulk_run([{"task": h} for h in headlines])
samples = agent.run("Fed holds rates steady", n=3)
agent.check_available_tokens()

from swarms import Agent

agent = Agent(
    agent_name="Tagger",
    model_name="gpt-5.4-mini",
    system_prompt="Reply with one topic tag for the headline.",
    max_loops=1,
    output_type="final",
    print_on=False,
)

headlines = ["Fed holds rates steady", "Chipmaker beats on datacenter revenue"]

# was agent.bulk_run(...)
tags = agent.run_batched(headlines)
print(tags)

# was agent.run(task, n=3): run it three times yourself, on a fresh conversation each time
samples = []
for _ in range(3):
    agent.reset()
    samples.append(agent.run("Fed holds rates steady"))
print(len(samples), "samples")

# was agent.check_available_tokens()
print("next request would carry about", agent.input_tokens, "input tokens")

from swarms import Agent
from swarms.structs.agent import AgentLLMError

agent = Agent(
    agent_name="Analyst",
    model_name="anthropic/claude-sonnet-5-5",
    fallback_models=["anthropic/claude-sonnet-5-5", "gpt-5.4", "gemini/gemini-2.5-pro"],
    retry_attempts=2,   # client retries per model, then the next model in the list
    max_loops=1,
    output_type="final",
    print_on=False,
)

try:
    print(agent.run("Summarise the risks of buying an ETF on margin in two sentences."))
    print("answered by:", agent.model_name)
except AgentLLMError as error:
    # Every model failed. The provider's own exception is chained underneath.
    print("every model failed:", type(error.__cause__).__name__, error.__cause__)

agent = Agent(model_name="gpt-5.4", mcp_url="python server.py")

import os

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("rates")


@mcp.tool()
def policy_rate(central_bank: str) -> str:
    """Return the current policy rate for a central bank."""
    # A stdio server in v17 sees only the env you pass it, plus a minimal default.
    source = os.environ.get("RATES_API_KEY", "missing")
    return f"{central_bank}: 4.25% (key {'set' if source != 'missing' else 'missing'})"


if __name__ == "__main__":
    mcp.run("stdio")

import sys

from swarms import Agent, MCPConnection

agent = Agent(
    agent_name="Rates-Desk",
    model_name="gpt-5.4",
    system_prompt="Use the policy_rate tool to answer questions about interest rates.",
    max_loops=1,
    output_type="final",
    print_on=False,
    mcp_config=MCPConnection(
        command=sys.executable,           # was mcp_url="python server.py"
        args=["server.py"],
        env={"RATES_API_KEY": "env:RATES_API_KEY"},  # pass what the server needs, explicitly
    ),
)
print(agent.run("What is the ECB's policy rate?"))

from swarms import Agent
from swarms.prompts.agent_system_prompts import build_agent_system_prompt

# v17: no system_prompt means no system message at all
plain = Agent(agent_name="Plain", model_name="gpt-5.4-mini", max_loops=1, output_type="final")

# Your own prompt is sent as written
analyst = Agent(
    agent_name="Analyst",
    model_name="anthropic/claude-sonnet-5-5",
    system_prompt="You are a credit analyst. Answer in three bullet points.",
    max_loops=1,
    output_type="final",
)

# The v16 built-in persona, if your agents depended on it
legacy = Agent(
    agent_name="Legacy",
    model_name="gemini/gemini-2.5-pro",
    system_prompt=build_agent_system_prompt(),
    max_loops=1,
    output_type="final",
)

for agent in (plain, analyst, legacy):
    print(agent.agent_name, "->", agent.run("What drives a company's interest coverage ratio?"))

from swarms import Agent

default_agent = Agent(agent_name="Explainer", model_name="gpt-5.4-mini", max_loops=1, print_on=False)
final_agent = Agent(
    agent_name="Explainer", model_name="gpt-5.4-mini", max_loops=1, print_on=False,
    output_type="final",  # only the newest reply
)

for agent in (default_agent, final_agent):
    agent.run("In one sentence, what is a write-ahead log?")
    second = agent.run("And in one sentence, what is a checkpoint?")
    print(f"output_type={agent.output_type!r}:")
    print(second)
    print("---")

from swarms import Agent, SwarmRouter

def make(name, model, prompt):
    return Agent(agent_name=name, model_name=model, system_prompt=prompt, max_loops=1, print_on=False)

agents = [
    make("Bull", "deepseek/deepseek-v4-pro", "Argue the bull case in two sentences."),
    make("Bear", "mistral/mistral-large-latest", "Argue the bear case in two sentences."),
    # MixtureOfAgents uses the last agent as its aggregator
    make("Judge", "anthropic/claude-opus-5-5", "Weigh both cases and give a one-line verdict."),
]

router = SwarmRouter(
    name="Equity-Debate",
    agents=agents,
    swarm_type="MixtureOfAgents",
    fallback_swarms=["SequentialWorkflow"],  # tried only if the primary swarm raises
    output_type="final",
)

result = router.run("Should a long-term investor buy Nvidia at 40x earnings?")
print("ran as:", router.active_swarm_type)
for attempt in router.fallback_attempts:
    print("failed first:", attempt["swarm_type"], "->", type(attempt["error"]).__name__)
print(result)

import os
from pathlib import Path

os.environ["WORKSPACE_DIR"] = str(Path("runs").resolve())  # read on every write in v17

from swarms import Agent, SequentialWorkflow

def make(name, model, prompt):
    return Agent(agent_name=name, model_name=model, system_prompt=prompt, max_loops=1, print_on=False)

workflow = SequentialWorkflow(
    name="Release-Notes",
    agents=[
        make("Drafter", "deepseek/deepseek-v4-pro", "Draft release notes from the change list."),
        make("Copyeditor", "anthropic/claude-haiku-5-5", "Tighten the draft. Keep every fact."),
    ],
    autosave=True,  # v17 default is False: nothing is written unless you ask
    output_type="final",
)
workflow.run("Changes: faster import, channels, uncapped worker pools.")

for path in sorted(Path("runs").rglob("*.json")):
    print(path)

from swarms import Agent, ConcurrentWorkflow

tickers = ["AAPL", "MSFT", "NVDA", "AMZN", "GOOGL", "META", "TSLA", "AVGO", "JPM", "V", "XOM", "LLY"]

screeners = [
    Agent(
        agent_name=f"Screener-{ticker}",
        model_name="groq/openai/gpt-oss-120b",  # a cheap, fast model for many parallel workers
        system_prompt=f"You screen {ticker}. Reply with one line: buy, hold or sell, and why.",
        max_loops=1,
        print_on=False,
    )
    for ticker in tickers
]

# v17 runs all 12 at once by default. Cap it if your provider rate-limits you.
workflow = ConcurrentWorkflow(name="Screen", agents=screeners, max_workers=4, output_type="dict")
results = workflow.run("Screen this stock for a 12-month horizon.")
print(len(results), "messages recorded")

export SWARMS_TELEMETRY_ON=False

pip install -U swarms

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