evofabric.core.multi_agent

evofabric.core.multi_agent#

Swarm#

class evofabric.core.multi_agent.Swarm(BaseComponent)[source]#

automated builder for creating ‘Swarm-style’ multi-agent collaboration graph

The Swarm component receives an agent dictionary, an entry point, and an optional topology definition to build an immediately runnable graph structure. It dynamically injects a special handoff tool into each agent, enabling them to delegate tasks to other agents along the defined communication paths.

Parameters:

  • agents (Dict[str, Union[AgentNode, Dict, LazyInstance]]) – A mapping from the unique agent name to an AgentNode instance or its configuration. The internally accepted type is AgentNodeOrConfig = InstanceOrConfig[AgentNode], and it can contain LazyInstance entries.

  • state_schema (Type[BaseModel]) – Pydantic model defining the graph shared state structure.

  • entry_point_agent (str) – Entry Point Agent Name. Must be one of the keys in agents.

  • edges (Optional[List[Tuple[str, str]]]) – Optional directed edges for specifying allowed task handoff candidates, formatted as [(source_agent_name, target_agent_name), ...]. If None, each agent can hand off tasks to any other agent by default. Note: This is a soft constraint used to adjust the function signature and documentation of the handoff tool.

  • max_turns (int, default 20) – Maximum iteration limit for the compiled graph, used to prevent infinite loops.

add_agent(name: str, agent: AgentNodeOrConfig) → None[source]#

Dynamically add an agent. It will take effect upon the next call to build().

Parameters:

  • name (str) – The unique agent name to be registered.

  • agent (AgentNodeOrConfig) – an AgentNode instance or configuration (such as LazyInstance).

Raises:

ValueError – If name already exists.

remove_agent(name: str) → None[source]#

Dynamically remove an agent. Will take effect on the next call to build().

Parameters:

name (str) – Agent name to be removed

Raises:

ValueError – Thrown when the agent does not exist or when attempting to remove the entry agent.

build()[source]#

Build and compile the Swarm graph based on the current configuration.

When the configuration of Swarm (e.g., adding/removing agents) changes, this method should be called.

Returns:

A compiled GraphEngine or GraphEngineDebugger instance, ready to run.

Example code:

from typing import List, Tuple, Dict
from pydantic import BaseModel
from evofabric.core.multi_agent import Swarm
from evofabric.core.agent import AgentNode
from evofabric.core.factory import LazyInstance
from evofabric.core.typing import StateMessage

# Define the state schema with a messages field
class MyState(BaseModel):
    messages: List[StateMessage] = []

# Define agents (instances or LazyInstance configs)
# You can supply real AgentNode instances...
planner = AgentNode(...)     # configure per your framework
writer = AgentNode(...)

# ...or defer construction using LazyInstance
# planner = LazyInstance(class_name="AgentNode", kwargs={...})
# writer = LazyInstance(class_name="AgentNode", kwargs={...})

swarm = Swarm(
    agents={
        "planner": planner,
        "writer": writer,
    },
    state_schema=MyState,
    entry_point_agent="planner",
    # If omitted, defaults to all-to-all (excluding self) suggestions
    edges=[("planner", "writer")],
    max_turns=15,
)

graph = swarm.build()
# Use the returned graph as per your Graph runtime API.

Topology Customization via Edges:

# Allow planner to hand off to researcher or writer; researcher can hand off to writer
edges = [
    ("planner", "researcher"),
    ("planner", "writer"),
    ("researcher", "writer"),
    # writer has no outgoing edges -> no handoff tool injected for writer
]

swarm = Swarm(
    agents={"planner": planner, "researcher": AgentNode(...), "writer": writer},
    state_schema=MyState,
    entry_point_agent="planner",
    edges=edges,
)

graph = swarm.build()
# The "handoff" tool on each agent will expose only the allowed target names.
Contents