agents.multi.experiments.routing_patterns¶

Routing patterns for multi-agent systems.

from typing import Any Experiments with conditional routing, branching, and dynamic paths. Uses BaseGraph’s add_conditional_edges for sophisticated routing.

Classes¶

BranchingMultiAgent

Multi-agent with branching and merging capabilities.

RoutingMultiAgent

Multi-agent with conditional routing capabilities.

Functions¶

category_router(state)

Route based on category field.

confidence_router(state)

Route based on confidence level.

error_router(state)

Route based on error presence.

has_tool_calls_router(state)

Check if there are tool calls in the last message.

Module Contents¶

class agents.multi.experiments.routing_patterns.BranchingMultiAgent¶

Bases: RoutingMultiAgent

Multi-agent with branching and merging capabilities.

Supports parallel branches that merge back together.

Examples

multi = BranchingMultiAgent("branching")

# Main path
multi.append(InputProcessor())

# Branch based on input type
multi.branch_on(
"InputProcessor",
lambda s: s.get("input_type"),
branches={
"text": [TextAnalyzer(), TextSummarizer()],
"image": [ImageAnalyzer(), ImageCaptioner()],
"audio": [AudioTranscriber(), AudioAnalyzer()]
},
merge_to=OutputFormatter()
)
branch_on(source, condition, branches, merge_to=None)¶

Create branching paths that merge back.

Parameters:

  • source (str | haive.agents.base.agent.Agent) – Agent to branch from

  • condition (collections.abc.Callable[[Any], str]) – Function returning branch key

  • branches (dict[str, list[haive.agents.base.agent.Agent]]) – Map of keys to agent sequences

  • merge_to (haive.agents.base.agent.Agent | None) – Agent where branches merge (optional)

Return type:

BranchingMultiAgent

build_graph()¶

Build graph with branching support.

Return type:

haive.core.graph.state_graph.base_graph2.BaseGraph

class agents.multi.experiments.routing_patterns.RoutingMultiAgent¶

Bases: haive.agents.multi.experiments.list_multi_agent.ListMultiAgent

Multi-agent with conditional routing capabilities.

Extends ListMultiAgent with routing rules that determine which agent executes next based on state conditions.

Examples

multi = RoutingMultiAgent("router")

# Add agents
multi.append(ClassifierAgent())
multi.append(TechnicalAgent())
multi.append(BusinessAgent())
multi.append(GeneralAgent())

# Add routing from classifier
multi.add_route(
source="ClassifierAgent",
condition=lambda state: state.get("category", "general"),
routes={
"technical": "TechnicalAgent",
"business": "BusinessAgent",
"general": "GeneralAgent"
}
)
add_boolean_route(source, condition, true_dest, false_dest=END)¶

Add simple boolean routing.

Parameters:

  • source (str | haive.agents.base.agent.Agent) – Source agent

  • condition (collections.abc.Callable[[Any], bool]) – Boolean condition function

  • true_dest (str | haive.agents.base.agent.Agent) – Destination when condition is True

  • false_dest (str | haive.agents.base.agent.Agent) – Destination when condition is False (default: END)

Return type:

RoutingMultiAgent

add_multi_route(source, condition, **routes)¶

Add multi-way routing with keyword arguments.

Examples

multi.add_multi_route(

“classifier”, lambda s: s[“category”], technical=”tech_agent”, business=”biz_agent”, creative=”creative_agent”

)

Parameters:
Return type:

RoutingMultiAgent

add_route(source, condition, routes, default=None)¶

Add routing rule for an agent.

Parameters:

  • source (str | haive.agents.base.agent.Agent) – Source agent (name or instance)

  • condition (collections.abc.Callable[[Any], str | bool]) – Function that returns route key based on state

  • routes (dict[str | bool, str | haive.agents.base.agent.Agent]) – Map of condition results to destination agents

  • default (str | None) – Default route if no match (uses self.default_route if None)

Return type:

RoutingMultiAgent

build_graph()¶

Build graph with conditional routing.

Return type:

haive.core.graph.state_graph.base_graph2.BaseGraph

agents.multi.experiments.routing_patterns.category_router(state)¶

Route based on category field.

Parameters:

state (dict[str, Any])

Return type:

str

agents.multi.experiments.routing_patterns.confidence_router(state)¶

Route based on confidence level.

Parameters:

state (dict[str, Any])

Return type:

str

agents.multi.experiments.routing_patterns.error_router(state)¶

Route based on error presence.

Parameters:

state (dict[str, Any])

Return type:

str

agents.multi.experiments.routing_patterns.has_tool_calls_router(state)¶

Check if there are tool calls in the last message.

Parameters:

state (dict[str, Any])

Return type:

bool