Skip to main content

Documentation Index

Fetch the complete documentation index at: https://launchpad.datalumina.com/llms.txt

Use this file to discover all available pages before exploring further.

This node enables dynamic routing by selecting the next node based on routing rules. Each rule is a RouterNode. If no rule matches, the router uses a fallback node. Router components:
  • BaseRouter - Orchestrates routing decisions and manages the routing process
  • RouterNode - Individual routing rules implementing conditional logic

​
Router Classes

​
BaseRouter

class BaseRouter(Node):
    async def process(self, task_context: TaskContext) -> TaskContext:
        pass

    def route(self, task_context: TaskContext) -> Node:
        for route_node in self.routes:
            route_node.task_context = task_context
            next_node = route_node.determine_next_node(task_context)
            if next_node:
                return next_node
        return self.fallback if self.fallback else None

​
RouterNode

class RouterNode(ABC):
    @abstractmethod
    def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
        pass

    @property
    def node_name(self):
        return self.__class__.__name__

​
How routing works

  1. Router Node Reached - The workflow detects is_router=True on the current NodeConfig
  2. Route Evaluation - The workflow calls BaseRouter.route(), which iterates through each RouterNode and calls determine_next_node()
  3. First Match Wins - The first router rule that returns a node determines the next step
  4. Fallback Handling - If no rules match, the router uses the fallback node
  5. Class Resolution - The selected node instance is converted back to its class and used as the next workflow node

​
Implementation example

​
Main Router

class TicketRouterNode(BaseRouter):
    def __init__(self):
        self.routes = [
            CloseTicketRouter(),
            EscalationRouter(),
            InvoiceRouter(),
        ]
        self.fallback = GenerateResponseNode()

​
Router rule examples

class CloseTicketRouter(RouterNode):
    def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
        output: FilterSpamNode.OutputType = task_context.nodes["FilterSpamNode"][
            "result"
        ].output
        if not output.is_human and output.confidence > 0.8:
            return CloseTicketNode()
        return None

class EscalationRouter(RouterNode):
    def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
        analysis = task_context.nodes["DetermineTicketIntentNode"]["result"].output
        if analysis.intent.escalate or analysis.escalate:
            return EscalateTicketNode()
        return None

​
Best practices

  • Order Matters - Arrange router nodes in order of priority; first match wins
  • Clear Conditions - Keep routing conditions explicit and easy to reason about
  • Fallback Strategy - Provide a meaningful fallback node
  • Test Coverage - Test all routing paths

​
Configuration in WorkflowSchema

NodeConfig(
    node=TicketRouterNode,
    connections=[CloseTicketNode, EscalateTicketNode, ProcessInvoiceNode, GenerateResponseNode],
    is_router=True,
    description="Route tickets based on analysis results",
)