TopolearnFlows give you deterministic, testable control over multi-step AI apps, with crews only where needed.
A Flow is a Python class whose methods are steps. Decorators (the @name lines above a method) say when each step runs:
@start()marks an entry point. It runs when the flow starts.@listen(step)runs a method afterstepfinishes. It receives that step's return value as an argument.
class Pipeline(Flow):
@start()
def fetch(self):
return "raw text"
@listen(fetch)
def clean(self, text):
return text.strip()Pipeline().kickoff() runs the flow and returns the output of the last step that finished.
State is data shared by all steps of one run. Two styles:
- Unstructured:
self.statebehaves like a dictionary:self.state["count"] = 1. - Structured: subclass
Flow[MyState], whereMyStateis a Pydantic model (a class with typed, validated fields). Then useself.state.count. This catches typos and documents the data. Prefer it beyond toys.
Every run's state has a unique id, used later for saving and resuming.
You can pass values in with kickoff(inputs={...}); matching state fields are filled before the first step.
Steps can contain anything: plain code, a direct LLM call, agent.kickoff(...) for a single agent, or a whole crew. That is the point of flows: AI where you need it, normal code everywhere else.
flow.plot("my_flow") writes an HTML diagram of the steps, useful for reviews.
Example
from pydantic import BaseModel
from crewai.flow import Flow, listen, start
class TicketState(BaseModel):
text: str = ""
words: int = 0
summary: str = ""
class TicketFlow(Flow[TicketState]):
@start()
def receive(self):
self.state.words = len(self.state.text.split())
return self.state.text
@listen(receive)
def summarize(self, text):
# Replace with an agent or crew call in a real app
self.state.summary = text[:40] + ("..." if len(text) > 40 else "")
return self.state.summary
flow = TicketFlow()
out = flow.kickoff(inputs={"text": "My invoice for March shows a double charge, please fix."})
print(out) # My invoice for March shows a double char...
print(flow.state.words) # 10Remember
@start()steps run first;@listen(step)runs afterstep, receiving its result- Structured state:
Flow[PydanticModel]andself.state.field - Steps mix plain code, LLM calls, single agents and crews
Watch out for
- Using untyped dictionary state in a large flow, where a typo in a key silently creates a new value.