Lab 03 — CrewAI Flow Engine

Phase 19 · Lab 03 · Phase README · Warmup

The problem

A Crew is a self-contained task sequence. A Flow is CrewAI's event-driven orchestration layer — the thing you reach for when one crew is not enough: several crews (or plain steps) wired together with dynamic routing, branching, fan-in/fan-out, persisted state, and human-in-the-loop gates. You don't write the call graph by hand; you declare an event graph with decorators, and the engine fires methods as their triggers complete.

The interview probes here are precise:

  • What does @start / @listen / @router actually do, and what does a @listen method receive?
  • How do or_ and and_ gate a step (fire on ANY vs ALL)?
  • Structured vs unstructured state — the dict-with-a-UUID vs the typed model — and why you'd pick each.
  • What does @persist buy you, and where does the state live?

You build the whole event engine, with the methods pure (they read/write self.state and return values), so the flow is deterministic and offline.

What you build

PieceWhat it doesThe lesson
@start / @listen(trigger) / @router(trigger)declare entry, chaining, and branchingthe event graph is the control flow
or_(...) / and_(...) + ConditionANY-of / ALL-of trigger gatesfan-in and "either path" joins
Flow.kickoff(inputs)the round-based event loopfires each method once → always terminates
structured (initial_state dataclass) vs unstructured (dict) statetyped model vs free dict, both id-stampedschema vs flexibility
@persist + InMemoryFlowStoresave state per stepcrash-safe resume (SQLite in real CrewAI)

File map

FileRole
lab.pyyour implementation (fill the # TODOs)
solution.pyreference + worked example (python solution.py)
test_lab.py21 tests: start, listen+payload, router branch, or_/and_, state, persist, termination, determinism
requirements.txtpytest only

Run it

pytest test_lab.py -v
LAB_MODULE=solution pytest test_lab.py -v
python solution.py

Success criteria

  • You can explain what a @listen(m) method receives (m's return value) and what a @router emits (a label that fires @listen("label")).
  • You can state the difference between or_ (ANY) and and_ (ALL) gating and give a use for each (either-path join vs fan-in).
  • You can explain why the engine always terminates (each method fires at most once).
  • You can contrast structured vs unstructured state and name what @persist stores and where (SQLite, keyed by the state id).
  • You can say when you'd use a Flow over a Crew (multi-crew, branching, HITL) — and vice versa.
  • All 21 tests pass under both lab and solution.

How this maps to the real stack

  • This is CrewAI's crewai.flow.flow module. Real Flow, @start(), @listen(...), @router(...), or_, and_, and @persist behave as modeled: @listen fires on completion and receives the trigger's output; @router returns a string that routes to matching @listen; or_/and_ gate on ANY/ALL. self.state is a Pydantic BaseModel for structured flows (class MyFlow(Flow[MyState])) or a dict for unstructured — both get an auto id.
  • Real @persist writes to SQLite (SQLiteFlowPersistence) keyed by the state id, so a flow can be resumed after a crash — the durable-execution idea from Phase 08, at the flow level. Our InMemoryFlowStore is the same interface with a dict backend.
  • The real reason Flows exist: orchestrating multiple Crews with routing and human-in-the-loop — a @router can branch on a human approval, and a @listen can run a whole Crew().kickoff() and pass its CrewOutput downstream. Flow = the conductor; Crew = the section that plays.

Limits of the miniature. We use a stdlib @dataclass where real CrewAI uses Pydantic (so no field validation/coercion), an in-memory store instead of SQLite, and an injected counter id instead of a UUID (for determinism, per the LAB-STANDARD). We also fire each listener at most once and process in deterministic rounds; real CrewAI's dispatch is event-async, and an or_ listener can re-fire per trigger. The event-graph semantics — what triggers what, what a listener receives, how branches gate — are faithful.

Extensions (your own machine)

  • Put a real Crew inside a @listen: have a step call SomeCrew().kickoff(inputs) and route on its CrewOutput — the canonical "Flows orchestrate Crews" pattern.
  • Add human-in-the-loop: a @router that returns "approve" / "reject" based on an injected approval callback, and resume the flow from persisted state after the human answers.
  • Swap InMemoryFlowStore for a SQLite store (stdlib sqlite3) keyed by the state id, then kill the process mid-flow and resume — durable execution (Phase 08) for flows.

Interview / resume signal

"Built CrewAI's Flow engine from scratch — @start/@listen/@router event decorators, or_/and_ fan-in gates, structured vs unstructured id-stamped state, and a @persist store — as a deterministic round-based event loop, and can explain when a Flow (branching, multi-crew, HITL) beats a Crew and vice versa."