多智能体系统的失败模式：7 种静默失败及其修复

pip install -U swarms
# or
uv pip install -U swarms

export OPENAI_API_KEY="sk-..."

python -c "import swarms; print(swarms.__version__)"

from swarms import Agent, AgentLLMError

TASK = "In one sentence: why should a failed LLM call raise an error?"

# No fallback: a model that cannot be reached now raises.
broken = Agent(
    agent_name="Broken-Agent",
    model_name="gpt-5.4-typo",
    max_loops=1,
    retry_attempts=1,
    print_on=False,
)

try:
    broken.run(TASK)
except AgentLLMError as e:
    print(f"Raised {type(e).__name__}: {str(e)[:90]}...")

# With fallbacks: the first entry is the primary, the rest are tried in order.
resilient = Agent(
    agent_name="Resilient-Agent",
    fallback_models=["gpt-5.4-typo", "gpt-5.4-mini"],
    max_loops=1,
    retry_attempts=1,
    output_type="final",
    print_on=False,
)

answer = resilient.run(TASK)
print("Served by:", resilient.get_current_model())
print("Answer:", answer)

from swarms import Agent


def get_exchange_rate(base: str, quote: str) -> str:
    """Get the current exchange rate between two currencies.

    Args:
        base: Base currency code, e.g. 'USD'.
        quote: Quote currency code, e.g. 'EUR'.

    Returns:
        The exchange rate as a string.
    """
    raise ConnectionError("rates API returned HTTP 503")


agent = Agent(
    agent_name="FX-Agent",
    system_prompt="Answer currency questions. If a tool fails, say so plainly and do not guess a number.",
    model_name="gpt-5.4-mini",
    tools=[get_exchange_rate],
    tool_retry_attempts=1,
    max_loops=2,
    output_type="final",
    print_on=False,
)

print(agent.run("What is the USD to EUR exchange rate right now?"))

from swarms import Agent, HierarchicalSwarm

workers = [
    Agent(agent_name="Researcher", model_name="gpt-5.4-mini", max_loops=1, print_on=False),
    Agent(agent_name="Writer", model_name="gpt-5.4-mini", max_loops=1, print_on=False),
]

# The director's model is unreachable, which simulates a provider outage.
swarm = HierarchicalSwarm(
    agents=workers,
    director_model_name="gpt-5.4-typo",
    director_settings={"retry_attempts": 1},
    max_loops=2,
    print_on=False,
)

try:
    swarm.run("Write a two-line product update.")
    print("Reported success")
except Exception as e:
    print(f"Run failed loudly: {type(e).__name__}")

try:
    from mcp.server.mcpserver import MCPServer  # mcp 2.x
except ImportError:
    from mcp.server.fastmcp import FastMCP as MCPServer  # mcp 1.x

server = MCPServer(name="inventory")


@server.tool()
def reserve_stock(sku: str, quantity: int) -> str:
    """Reserve stock for an order."""
    raise RuntimeError(f"warehouse API timed out reserving {quantity} x {sku}")


if __name__ == "__main__":
    server.run()

import sys

from swarms import MCPConnection, MCPManager

manager = MCPManager(
    mcp_config=MCPConnection(
        transport="stdio",
        command=sys.executable,
        args=["flaky_server.py"],
    )
)

result = manager.call_tool("reserve_stock", {"sku": "SKU-42", "quantity": 3})

print("is_error:", result["is_error"])
print("result:", result.get("result"))
assert result["is_error"], "A failed MCP tool call was reported as a success"

from swarms import RoundRobinSwarm


class RecordingAgent:
    """A stub agent that records every message it is shown. No API calls."""

    def __init__(self, agent_name):
        self.agent_name = agent_name
        self.seen = []

    def run(self, task, *args, messages=None, **kwargs):
        self.seen.append(str(task) + str(messages or ""))
        return f"{self.agent_name} done"


alice, bob = RecordingAgent("Alice"), RecordingAgent("Bob")
swarm = RoundRobinSwarm(agents=[alice, bob], max_loops=1)

swarm.run("Task A. Customer secret: CANARY-7F3A")
alice.seen.clear()
bob.seen.clear()

swarm.run("Task B. Summarise the release notes.")

leaked = any("CANARY-7F3A" in m for m in alice.seen + bob.seen)
print("Task A leaked into task B:", leaked)
assert not leaked

import time

from swarms import ConcurrentWorkflow


class SlowAgent:
    """A stub agent that sleeps, then answers. No API calls."""

    def __init__(self, agent_name, delay):
        self.agent_name = agent_name
        self.delay = delay

    def run(self, task, *args, **kwargs):
        time.sleep(self.delay)
        return f"{self.agent_name} answer"


agents = [
    SlowAgent("First", 0.30),
    SlowAgent("Second", 0.20),
    SlowAgent("Third", 0.05),
]

workflow = ConcurrentWorkflow(agents=agents, output_type="dict", autosave=False)
result = workflow.run("Same task for everyone")

order = [m["role"] for m in result if m["role"] != "User"]
print("Returned order:", order)
assert order == ["First", "Second", "Third"]

import time
from datetime import datetime

# The bug: an f-string at module level is evaluated once, at import.
PROMPT = f"You are a helpful assistant. Time: {datetime.now():%H:%M:%S}"


# The fix: build the prompt when it is used.
def build_prompt() -> str:
    return f"You are a helpful assistant. Time: {datetime.now():%H:%M:%S}"


time.sleep(2)
print("Constant:", PROMPT)
print("Function:", build_prompt())

from datetime import datetime

from swarms import Agent

agent = Agent(agent_name="Clock-Agent", model_name="gpt-5.4-mini", max_loops=1, print_on=False)

# The default system prompt is built when the agent is constructed.
print([line for line in agent.system_prompt.splitlines() if line.startswith("Time:")])

# For an agent object you keep alive for days, send the time with each task.
task = f"Current time: {datetime.now().astimezone():%Y-%m-%d %H:%M %Z}. Is it a weekday?"
print(task)

from swarms import Agent, SwarmRouter

agents = [
    Agent(
        agent_name="Researcher",
        system_prompt="List three facts about the topic. Be brief.",
        model_name="gpt-5.4-mini",
        max_loops=1,
        print_on=False,
    ),
    Agent(
        agent_name="Writer",
        system_prompt="Turn the facts you are given into one short paragraph.",
        model_name="gpt-5.4-mini",
        max_loops=1,
        print_on=False,
    ),
]

router = SwarmRouter(
    agents=agents,
    swarm_type="HierarchicalSwarm",
    director_model_name="gpt-5.4-typo",  # simulate a director outage
    director_settings={"retry_attempts": 1},
    fallback_swarms=["SequentialWorkflow"],
    output_type="final",
)

result = router.run("Why do multi-agent systems need fallbacks?")

print("Served by:", router.active_swarm_type)
for attempt in router.fallback_attempts:
    print("Failed:", attempt["swarm_type"], "-", type(attempt["error"]).__name__)
print(result)