TopolearnReal processes branch on data and merge parallel work; routers make that explicit and visible.
A router is a step that decides where the flow goes next. Mark it @router(step) and return a label, a short string. Steps that listen to that label run next:
@router(classify)
def route(self):
return "urgent" if self.state.priority > 3 else "normal"
@listen("urgent")
def page_on_call(self): ...Annotate the router's return type with Literal["urgent", "normal"] so plot() can draw both branches.
To combine several steps, use:
or_(a, b): run when any of the listed steps finishes. Useful when two branches end in the same follow-up.and_(a, b): run only after all listed steps have finished. Useful to join work that ran in parallel.
Several methods can listen to the same step. They run concurrently, which is how a flow fans out.
A flow can have more than one @start() method; all of them run at the beginning.
Loops are possible: a router can return a label that leads back to an earlier step, for example "retry" after a failed check. Always keep a counter in state so a loop cannot run forever.
Routing on facts from code (scores, flags, lengths) is predictable. When a model must make the decision, have it return structured output with a fixed set of choices, then route on that field.
Example
from typing import Literal
from pydantic import BaseModel
from crewai.flow import Flow, and_, listen, or_, router, start
class S(BaseModel):
amount: float = 0
log: list[str] = []
class RefundFlow(Flow[S]):
@start()
def check_fraud(self): self.state.log.append("fraud ok")
@start()
def check_stock(self): self.state.log.append("stock ok")
@router(and_(check_fraud, check_stock)) # wait for both checks
def decide(self) -> Literal["auto", "review"]:
return "auto" if self.state.amount < 100 else "review"
@listen("auto")
def approve(self): self.state.log.append("approved")
@listen("review")
def escalate(self): self.state.log.append("sent to human")
@listen(or_(approve, escalate)) # either branch ends here
def notify(self): return " | ".join(self.state.log)
print(RefundFlow().kickoff(inputs={"amount": 250})) # fraud ok | stock ok | sent to humanRemember
@router(step)returns a label;@listen("label")handles that branchand_waits for all listed steps;or_fires on any- Annotate routers with Literal so plots show every branch
Watch out for
- Building a retry loop with a router but no attempt counter in state, so a stubborn failure loops forever.