Run Support Swarm Example

This page embeds the full source for examples/run_support_swarm.py.

"""Example: run a single user turn through the swarm orchestrator with mocked agent iterations and variable extraction."""

import asyncio
import json

from swarmforge.authoring import build_swarm_definition
from swarmforge.swarm import AgentTurnResult, InMemorySessionStore, SwarmSession, process_swarm_stream


swarm = build_swarm_definition(
    {
        "nodes": [
            {
                "node_key": "triage",
                "name": "Triage",
                "persona": "",
                "system_prompt": "You triage support requests and transfer them when the route is clear.",
                "is_entry_node": True,
                "sub_agents": [
                    {
                        "sub_agent": "billing",
                        "handoff_description": "Transfer only after confirming the request is billing-related.",
                        "required_variables": ["account_id"],
                    }
                ],
            },
            {
                "node_key": "billing",
                "name": "Billing",
                "persona": "",
                "system_prompt": "You handle billing issues.",
                "is_entry_node": False,
            },
        ],
    },
    swarm_id="support",
    name="Support Swarm",
)


class SupportTurnRunner:
    async def run_turn(self, *, agent_node, contents, config):
        del contents, config
        if agent_node.node_key == "triage":
            return AgentTurnResult(tool_calls=[{"id": "handoff-1", "name": "transfer_to_agent", "args": {"target_node_key": "billing"}}])
        return AgentTurnResult(response_text="I can help with that billing issue for account ACME-991.")


async def extract_required_variables(**_kwargs):
    return {"account_id": "ACME-991"}


async def main():
    session = SwarmSession(id="session-1", swarm=swarm)
    store = InMemorySessionStore()
    async for event in process_swarm_stream(
        session,
        "I need help with a charge on my account.",
        store=store,
        turn_runner=SupportTurnRunner(),
        extract_required_variables=extract_required_variables,
    ):
        print(json.dumps(event, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

Run:

python examples/run_support_swarm.py