> ## 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.

# Router Node

> Implement dynamic workflow routing with BaseRouter and RouterNode classes to create conditional branching based on processing results.

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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
class TicketRouterNode(BaseRouter):
    def __init__(self):
        self.routes = [
            CloseTicketRouter(),
            EscalationRouter(),
            InvoiceRouter(),
        ]
        self.fallback = GenerateResponseNode()
```

### Router rule examples

```python theme={null}
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

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