Workflow.route
from loomflow import Workflow
wf = Workflow.route(classifier, {"a": fn_a, "b": fn_b})Classify-then-dispatch. The classifier step’s output is converted
to str and used as a key into routes. The matching step runs with
the original input (not the classifier’s output), so handlers see
what the user asked, not the routing label.
Signature
@classmethod
def route(
cls,
classifier: StepLike,
routes: Mapping[str, StepLike],
*,
default: StepLike | None = None,
name: str = "route",
telemetry: Telemetry | None = None,
audit_log: AuditLog | None = None,
max_steps: int = 100,
max_visits_per_node: int = 25,
) -> Workflow: ...| Parameter | Type | Default | Description |
|---|---|---|---|
classifier | StepLike | required | Returns a string used as a key into routes. The output is str()-cast and .strip()-ed before lookup, so a classifier that returns “billing\n” still routes to "billing". Can be a function, Agent, or nested Workflow. |
routes | Mapping[str, StepLike] | required | Keys are routing labels; values are handlers. Must be non-empty. Each handler receives the original input (the classifier’s output is only used for routing). |
default | StepLike | None | None | Fallback for unmatched routing keys. When None, an unmatched key raises RuntimeError. |
name | str | "route" | Workflow name. |
telemetry / audit_log | . | None | Same as Workflow.chain. |
max_steps / max_visits_per_node | . | 100 / 25 | Same as Workflow.chain. |
Why handlers see the original input
In a hand-rolled router, you’d typically write:
label = await classifier(input)
result = await {"billing": billing_h, "tech": tech_h}[label](input) # ← input, not labelWorkflow.route matches that shape: the routing label is consumed by
the router only; handlers see the original message. This avoids
the common “I have to remember to pass the original input through”
bug.
Example. Agent classifier + Agent handlers
from loomflow import Agent, Workflow
billing = Agent("Handle billing.", model="claude-opus-4-7", tools=[...])
tech = Agent("Handle tech.", model="claude-opus-4-7", tools=[...])
sales = Agent("Handle sales.", model="claude-opus-4-7")
classifier = Agent(
"Classify the request as 'billing', 'tech', or 'sales'. "
"Reply with exactly one of those words.",
model="claude-haiku-4-5", # cheap classifier; expensive specialists
)
support = Workflow.route(
classifier,
{"billing": billing, "tech": tech, "sales": sales},
default=tech, # ambiguous → tech
name="customer_support",
)
result = await support.run(
"My credit card was charged twice for last month's invoice.",
user_id="user_42",
)
print(result.visited) # ['classify', 'route_billing']
print(result.output) # ← billing agent's responseExample. Function classifier
The classifier doesn’t need to be an Agent. A pure function works:
async def classify_by_keyword(text: str) -> str:
text = text.lower()
if any(w in text for w in ("invoice", "charge", "refund")):
return "billing"
if any(w in text for w in ("error", "bug", "broken")):
return "tech"
return "general"
wf = Workflow.route(
classify_by_keyword,
{"billing": billing_handler, "tech": tech_handler},
default=general_handler,
)For RAG / structured-tag classifiers, use whatever’s appropriate for your latency budget. An LLM classifier is overkill if a regex works.
What the resulting graph looks like
Workflow.route builds:
┌─► route_billing ─► END
classify ────►│
├─► route_tech ─► END
│
└─► route_default ─► END (only when default= is set)Each route_<key> is a node; the classifier is followed by an
add_router edge that maps classifier output → next node. The
underlying graph is exposed via the same add_node / add_edge /
add_router API documented in the explicit graph builder.
Routing-key normalization
The classifier’s output is normalized via str(value).strip()
before lookup. A classifier that emits:
| Output | Lookup key |
|---|---|
"billing" | "billing" |
" billing\n" | "billing" |
"BILLING" | "BILLING" (case-sensitive) |
42 | "42" |
Case sensitivity is intentional. Your route keys should match the classifier’s emission convention, not assume normalization. If your classifier sometimes returns mixed case, lowercase inside the classifier:
async def normalized(text: str) -> str:
return (await classifier.run(text)).output.strip().lower()Unmatched keys
wf = Workflow.route(classifier, {"a": fn_a}, default=None)
result = await wf.run("hello") # classifier returns 'b' → no match, no default
# → RuntimeError: router on 'classify' produced key 'b' with no matching route
# and no defaultWith a default=, the same call routes to default instead of
raising:
wf = Workflow.route(classifier, {"a": fn_a}, default=fn_default)
result = await wf.run("hello") # routes to route_defaultWhen to reach for route
- Customer support. Billing / tech / sales triage.
- Coding agent gateway. Language-specific specialists.
- Search front-end. Multiple retrieval strategies, picked by query shape.
For multi-domain tasks (research + write + review), use
Supervisor instead. Router /
Workflow.route commits to one specialist; Supervisor can delegate
to many.
Workflow.route vs Router architecture. Both classify-and-dispatch.
The Router architecture lives inside an Agent (its base loop sees
the routes); Workflow.route lives at the workflow layer. Use the
architecture when you want the routing wrapped in agent semantics
(retries, structured outputs, full ReAct-style tool access at the
Agent level); use the workflow when you want a deterministic graph
that other workflow nodes can route into / out of.