agents.supervisor.utils.routing

Dynamic Routing Engine for Haive Supervisor System.

Handles intelligent routing decisions using DynamicChoiceModel and LLM-based analysis. Provides context-aware agent selection with validation and fallback mechanisms.

Classes

BaseRoutingStrategy

Abstract base for routing strategies.

DynamicRoutingEngine

Main routing engine that orchestrates routing decisions.

LLMRoutingStrategy

LLM-based routing strategy using structured output.

RoutingContext

Context extracted from state for routing decisions.

RoutingDecision

Model for routing decisions.

RuleBasedRoutingStrategy

Rule-based routing strategy for deterministic routing.

TaskClassifier

Classifies tasks for better routing decisions.

Module Contents

class agents.supervisor.utils.routing.BaseRoutingStrategy

Bases: abc.ABC

Abstract base for routing strategies.

abstractmethod make_routing_decision(context, available_agents, agent_capabilities, config=None)
Async:

Parameters:
Return type:

RoutingDecision

Make routing decision based on context.

class agents.supervisor.utils.routing.DynamicRoutingEngine(routing_model, routing_engine=None, routing_strategy=None, enable_context_analysis=True)

Main routing engine that orchestrates routing decisions.

Handles context extraction, strategy selection, and routing execution with comprehensive error handling and fallback mechanisms.

Initialize routing engine.

Parameters:

  • routing_model (haive.core.common.models.dynamic_choice_model.DynamicChoiceModel[str]) – DynamicChoiceModel for available choices

  • routing_engine (haive.core.engine.base.InvokableEngine | None) – Engine for LLM-based routing (optional)

  • routing_strategy (BaseRoutingStrategy | None) – Custom routing strategy (optional)

  • enable_context_analysis (bool) – Whether to perform context analysis

print_routing_stats()

Print routing engine statistics.

Return type:

None

async route_request(state, agent_capabilities, config=None)

Main routing method.

Parameters:

  • state (Any) – Current graph state

  • agent_capabilities (dict[str, str]) – Dict of agent capabilities

  • config (langchain_core.runnables.RunnableConfig | None) – Runnable configuration

Returns:

Command object with routing decision

Return type:

langgraph.types.Command

class agents.supervisor.utils.routing.LLMRoutingStrategy(routing_engine, routing_model)

Bases: BaseRoutingStrategy

LLM-based routing strategy using structured output.

Init .

Parameters:

  • routing_engine (haive.core.engine.base.InvokableEngine) – [TODO: Add description]

  • routing_model (haive.core.common.models.dynamic_choice_model.DynamicChoiceModel[str]) – [TODO: Add description]

async make_routing_decision(context, available_agents, agent_capabilities, config=None)

Make LLM-based routing decision.

Parameters:
Return type:

RoutingDecision

class agents.supervisor.utils.routing.RoutingContext(/, **data)

Bases: pydantic.BaseModel

Context extracted from state for routing decisions.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class agents.supervisor.utils.routing.RoutingDecision(/, **data)

Bases: pydantic.BaseModel

Model for routing decisions.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class agents.supervisor.utils.routing.RuleBasedRoutingStrategy(routing_rules)

Bases: BaseRoutingStrategy

Rule-based routing strategy for deterministic routing.

Initialize with routing rules.

Parameters:

routing_rules (dict[str, str]) – Dict mapping keywords to agent names

async make_routing_decision(context, available_agents, agent_capabilities, config=None)

Make rule-based routing decision.

Parameters:
Return type:

RoutingDecision

class agents.supervisor.utils.routing.TaskClassifier

Classifies tasks for better routing decisions.

classmethod classify_task(message)

Classify task based on message content.

Parameters:

message (str) – Message to classify

Returns:

List of detected task types

Return type:

list[str]

classmethod estimate_complexity(message, conversation_length)

Estimate task complexity.

Parameters:

  • message (str) – Message content

  • conversation_length (int) – Length of conversation

Returns:

Simple/Medium/Complex

Return type:

Complexity level