Usage Example

Usage Example#

Example 1: Weather Inquiry (Star Topology)#

Planner coordinates, other Agents each perform their respective duties. Users only input natural language and do not need to know how the system is internally divided.

import asyncio
from typing import Annotated, List
from pydantic import BaseModel
from evofabric.core.multi_agent import Swarm
from evofabric.core.agent import AgentNode
from evofabric.core.factory import ComponentFactory
from evofabric.core.tool import ToolManager
from evofabric.core.typing import StateMessage, UserMessage, AssistantMessage

client = ComponentFactory.create("OpenAIChatClient", api_key="...", model="gpt-4o-mini")

class S(BaseModel):
    messages: Annotated[List[StateMessage], "append_messages"] = []

def check_user_name(): return "Zhang San"
def check_user_location(name: str): return "Fuxin" if name == "Zhang San" else "Hong Kong"
def check_weather(city: str): return "Light rain" if city == "Fuxin" else "Cloudy"

planner = AgentNode(
    client=client,
    system_prompt=(
        "Plan the order of agent calls based on the user's question, and output 'FINISHED' when done. "
        "Available agents: user_name, user_conf, weather. Use 'handoff' to delegate tasks."
    ),
    tool_manager=[ToolManager(tools=[])]
)
user_name = AgentNode(
    client=client,
    system_prompt="Query the user's name, then hand off to the planner.",
    tool_manager=[ToolManager(tools=[check_user_name])]
)
user_conf = AgentNode(
    client=client,
    system_prompt="Query the user's city based on their name, then hand off to the planner.",
    tool_manager=[ToolManager(tools=[check_user_location])]
)
weather = AgentNode(
    client=client,
    system_prompt="Query the weather based on the city; you may answer directly and output 'FINISHED' at the end.",
    tool_manager=[ToolManager(tools=[check_weather])]
)

edges = [
    ("planner", "user_name"), ("user_name", "planner"),
    ("planner", "user_conf"), ("user_conf", "planner"),
    ("planner", "weather"), ("weather", "planner"),
]

swarm = Swarm(
    agents={"planner": planner, "user_name": user_name, "user_conf": user_conf, "weather": weather},
    state_schema=S,
    entry_point_agent="planner",
    edges=edges,
    max_turns=15,
)
graph = swarm.build()

async def main():
    state_in = {"messages": [UserMessage(content="What’s the weather like in my city today?")]}
    state_out = await graph.run(state_in)
    last = state_out.messages[-1]
    assert isinstance(last, AssistantMessage)
    print(last.content)  # Expected to contain weather information + FINISHED

asyncio.run(main())

Example 2: Runtime Extension Capability (Dynamic Agent Addition)#

First start with a fully connected configuration, then introduce the new time_agent at runtime, and re-build() to take effect.

from evofabric.core.agent import AgentNode
from evofabric.core.tool import ToolManager

def time_checker(): return "23:50"

time_agent = AgentNode(
    client=client,
    system_prompt="Query the current time and hand off to the planner.",
    tool_manager=[ToolManager(tools=[time_checker])]
)

swarm.add_agent("time_agent", time_agent)
graph = swarm.build()  # Important: rebuild after adding the new agent

# Now the planner's handoff can point to time_agent (under the default fully-connected mode)
# If edges were configured earlier, make sure to extend the edges before rebuilding

Example 3: Capture route logs for debugging#

At runtime, Swarm will print routing jump logs. These can be captured and analyzed to verify whether they follow your topology constraints and expected planned paths.

import io, contextlib

f = io.StringIO()
with contextlib.redirect_stdout(f):
    state_out = await graph.run({"messages": [UserMessage(content="Ask about the weather")]})
logs = f.getvalue()
print(logs)
# Example output (may vary depending on model behavior):
# Router: Detected handoff from 'user_name' to 'planner'.
# Router: Detected handoff from 'planner' to 'weather'.
Contents