Single Agent Swarm Example

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

"""Example: use the swarm runtime with a single entry agent and no handoffs for a one-assistant workflow."""

import asyncio
import json

from swarmforge.swarm import (
    AgentTurnResult,
    InMemorySessionStore,
    SwarmDefinition,
    SwarmNode,
    SwarmSession,
    process_swarm_stream,
)


class SingleAgentTurnRunner:
    async def run_turn(self, *, agent_node, contents, config):
        del agent_node, contents, config
        return AgentTurnResult(response_text="Here is a direct answer from the only assistant in the swarm.")


swarm = SwarmDefinition(
    id="single-agent",
    name="Single Agent",
    nodes=[
        SwarmNode(
            id="assistant",
            node_key="assistant",
            name="Assistant",
            system_prompt="You are a helpful assistant that handles the full request directly.",
            is_entry_node=True,
        )
    ],
)


async def main():
    session = SwarmSession(id="session-1", swarm=swarm)
    store = InMemorySessionStore()
    async for event in process_swarm_stream(
        session,
        "Give me a concise summary.",
        store=store,
        turn_runner=SingleAgentTurnRunner(),
    ):
        print(json.dumps(event, indent=2))


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

Run:

python examples/single_agent_swarm.py