Router Workflow - Definable AI

Route workflow execution to one or more paths based on a selector function.

from definable.agent import Agent
from definable.agent.workflow import Workflow, Router, Step

tech_agent = Agent(model="gpt-4o", instructions="Handle technical questions.")
billing_agent = Agent(model="gpt-4o", instructions="Handle billing questions.")
general_agent = Agent(model="gpt-4o", instructions="Handle general questions.")

workflow = Workflow(
    name="support",
    steps=[
        Router(
            name="classifier",
            selector=lambda ctx: "technical" if "bug" in (ctx.input or "").lower() else "general",
            routes={
                "technical": Step(name="tech", agent=tech_agent),
                "billing": Step(name="billing", agent=billing_agent),
                "general": Step(name="general", agent=general_agent),
            },
        ),
    ],
)

How It Works

The selector function receives the step context.
It returns a route name (string).
The matching route’s step executes.

Multi-Route Selection

Return a list of route names to execute multiple routes in parallel:

Router(
    name="multi-route",
    selector=lambda ctx: ["technical", "legal"] if "compliance" in (ctx.input or "") else ["general"],
    routes={
        "technical": Step(name="tech", agent=tech_agent),
        "legal": Step(name="legal", agent=legal_agent),
        "general": Step(name="general", agent=general_agent),
    },
)

Async Selectors

async def classify(ctx):
    # Call a classification model, API, etc.
    return "technical"

Router(name="classifier", selector=classify, routes={...})