Flow state with a Pydantic model

Flow[Ticket] gives a flow a typed state: a Pydantic model with named fields and defaults. Every method reads and writes the same object.

Lesson 20's state was a dictionary, and a typo in a key would only show up when that line ran. A model class fixes the fields up front.

import re

import shop_llm
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel


class Ticket(BaseModel):
    message: str = ""
    order_id: str = ""
    reply: str = ""

Ticket holds the three things the desk tracks. Every field needs a default, because the state exists before inputs are applied.

class Desk(Flow[Ticket]):
    @start()
    def read_ticket(self):
        found = re.findall(r"\b[A-Z]\d+\b", self.state.message)
        self.state.order_id = found[0] if found else ""

    @listen(read_ticket)
    def answer(self):
        self.state.reply = f"Looking into {self.state.order_id}."

Flow[Ticket] makes self.state a Ticket. The methods pass nothing to each other; they share the state.

desk = Desk(suppress_flow_events=True)
desk.kickoff(inputs={"message": "Where is my order A17?"})
print(desk.state.order_id, "|", desk.state.reply)

inputs filled message. After the run, the state holds what each step wrote, for your code to read.

desk = Desk(suppress_flow_events=True)
desk.kickoff(inputs={"message": None})

An input of the wrong type is refused before any method runs, where a dictionary state would have stored it and failed later inside re.findall. A key the model has no field for is dropped without a word, so a misspelled input name still needs care.