16I flow: passi, condizioni e persone

Una crew è una squadra che lavora su un incarico: tu descrivi agenti e task, e il modello decide come portarli a termine. Molte applicazioni vere però hanno bisogno di qualcosa in più: fare prima un controllo con regole precise, scegliere una strada diversa a seconda del risultato, chiamare una crew solo quando serve, fermarsi ad aspettare l'approvazione di una persona. Per questo esiste il FlowFlow: Il "copione" che orchestra passi, crew e agenti con uno stato condiviso, condizioni e diramazioni. La documentazione lo consiglia come struttura per le applicazioni vere. glossario.

Pensa alla cucina di un ristorante. La comanda (il foglietto con l'ordine) passa da una postazione all'altra e ognuno ci scrive sopra cosa ha fatto. Il capo cucina guarda la comanda e decide: questo piatto va alla griglia, quest'altro alla postazione dei freddi. Le brigate di cuochi sono le tue crew; il percorso della comanda, con le sue scelte, è il flow.

Perché i flow#

La documentazione ufficiale di CrewAI è netta: per un'applicazione destinata all'uso reale, «start with a Flow», cioè comincia da un flow. Anche il progetto di prova della guida rapida ufficiale è un flow che contiene una crew. Il motivo è che un flow ti lascia decidere quali parti sono lavoro del modello e quali sono codice normale, che è veloce, gratuito e fa sempre la stessa cosa.

Non sono alternative: nella pratica un flow contiene una o più crew. La skill ufficiale getting-started di CrewAI suggerisce di partire con un flow anche quando oggi ti basta una crew, così avrai già la struttura quando serviranno controlli e diramazioni.

Stato, @start e @listen#

Un flow si scrive come una Classe e oggettoClasse e oggetto: Una classe è uno stampo (per esempio Agent); un oggetto è una cosa concreta creata con quello stampo (il tuo agente ricercatore). glossario Python con dentro dei metodi, cioè funzioni che appartengono alla classe. Sopra ogni metodo un DecoratoreDecoratore: Una riga che inizia con @ sopra una funzione e le aggiunge un comportamento. In CrewAI: @agent, @task, @start, @listen. glossario dice quando deve partire. Questo esempio minimo ha due passi e uno stato con due campi:

from pydantic import BaseModel

from crewai.flow.flow import Flow, listen, start


# Lo stato: i dati che il flow porta con sé, con nome e tipo di ogni campo.
class StatoContatore(BaseModel):
    contatore: int = 0
    messaggio: str = ""


class PrimoFlow(Flow[StatoContatore]):   # tra quadre: la forma dello stato

    @start()                             # parte per primo
    def primo_passo(self):
        self.state.messaggio = "Ciao dal primo passo"
        self.state.contatore += 1

    @listen(primo_passo)                 # parte quando primo_passo ha finito
    def secondo_passo(self):
        self.state.messaggio += ", aggiornato dal secondo"
        self.state.contatore += 1
        return self.state.messaggio


flusso = PrimoFlow()
risultato = flusso.kickoff()
print(risultato)       # quello che ha restituito l'ultimo passo
print(flusso.state)    # lo stato finale

Ecco i pezzi, uno per uno:

• Lo stato è un modello PydanticPydantic: Libreria Python per descrivere la forma dei dati (campi e tipi) e controllarla. CrewAI la usa per ottenere risposte strutturate e verificate. glossario, come quelli del capitolo 13. Scrivendo Flow[StatoContatore] dici a CrewAI che forma ha. Ogni passo lo legge e lo modifica con self.state. CrewAI aggiunge da solo un campo id, un codice unico per ogni esecuzione. Esiste anche uno stato «libero», un DizionarioDizionario: Una raccolta di coppie nome → valore tra parentesi graffe: {"citta": "Arezzo"}. glossario senza forma fissa (self.state["campo"]), ma la documentazione consiglia quello strutturato: se scrivi male il nome di un campo, te ne accorgi subito.
• @start() segna il punto di partenza. Puoi averne più di uno: partono tutti all'avvio.
• @listen(primo_passo) fa partire il metodo quando primo_passo ha finito; se quel passo restituisce qualcosa con return, il valore arriva come parametro.
• kickoff() avvia il flow e restituisce il valore dell'ultimo passo completato. Con kickoff(inputs={"messaggio": "..."}) riempi i campi dello stato prima di partire: le chiavi devono avere gli stessi nomi dei campi.

Stampando lo stato alla fine vedrai contatore=2 e il messaggio con entrambe le frasi, più l'id.

Scegliere la strada con @router#

Un RouterRouter: Un passo di un flow che sceglie la strada successiva restituendo un'etichetta, per esempio "approvato" o "da_rivedere". glossario è un passo che non produce un risultato ma una decisione: restituisce una StringaStringa: Un testo nel codice, scritto tra virgolette: "Ricercatore". glossario, l'etichetta della strada da prendere. I metodi con @listen("etichetta") scritta tra virgolette aspettano quella strada; parte solo quello che corrisponde.

L'esempio di questo capitolo è un flow che smista i messaggi dei clienti di un negozio: quelli con parole come «guasto» o «rimborso» vanno subito a una persona, lo spam viene archiviato, gli altri vanno in coda. Le regole sono semplici controlli sul testo, senza nessun modello.

Ricongiungere le strade: or_ e and_#

• @listen(or_(passo_a, passo_b)) parte appena finisce uno qualsiasi dei passi indicati. Con CrewAI 1.15.21 abbiamo provato due passi di partenza che finivano entrambi: il metodo con or_ è partito una volta sola. Con un router il problema non si pone, perché parte una strada sola.
• @listen(and_(passo_a, passo_b)) aspetta che finiscano tutti. Serve quando due passi di partenza lavorano in parallelo (per esempio due ricerche) e un terzo deve unire i risultati.

Il codice completo, senza modello#

Questo script si esegue anche senza chiavi e senza Ollama, perché nessun passo usa un modello: è il modo migliore per capire il meccanismo dei flow. Lo stato ha tre campi, il router sceglie tra tre etichette, or_ riunisce le strade.

# /// script
# requires-python = ">=3.10,<3.14"
# dependencies = ["crewai[tools]==1.15.21"]
# ///
"""Un flow che smista i messaggi dei clienti con regole semplici, senza nessun modello."""
from pydantic import BaseModel

from crewai.flow.flow import Flow, listen, or_, router, start


# Lo stato: la "comanda" che il flow si porta dietro da un passo all'altro.
class StatoMessaggio(BaseModel):
    testo: str = ""
    categoria: str = ""
    azione: str = ""


# Parole che fanno scattare le due strade speciali (regole scritte da noi, niente IA).
PAROLE_URGENTI = ["guasto", "bloccato", "rimborso", "urgente", "non funziona"]
PAROLE_SPAM = ["hai vinto", "clicca qui", "offerta imperdibile"]


class SmistaMessaggi(Flow[StatoMessaggio]):

    @start()
    def ricevi(self):
        # Primo passo: ripulisce il testo (spazi in più, maiuscole).
        self.state.testo = self.state.testo.strip().lower()
        print(f"Ricevuto: {self.state.testo}")

    @router(ricevi)
    def classifica(self):
        # Il router restituisce un'etichetta. Il nome del metodo (classifica)
        # è diverso da tutte le etichette: è importante.
        if any(parola in self.state.testo for parola in PAROLE_SPAM):
            return "spam"
        if any(parola in self.state.testo for parola in PAROLE_URGENTI):
            return "urgente"
        return "normale"

    @listen("urgente")
    def passa_a_una_persona(self):
        self.state.categoria = "urgente"
        self.state.azione = "Avviso subito il responsabile dell'assistenza."

    @listen("normale")
    def metti_in_coda(self):
        self.state.categoria = "normale"
        self.state.azione = "Risposta entro 24 ore lavorative."

    @listen("spam")
    def scarta(self):
        self.state.categoria = "spam"
        self.state.azione = "Messaggio archiviato senza risposta."

    @listen(or_(passa_a_una_persona, metti_in_coda, scarta))
    def riepilogo(self):
        # Qualunque strada sia stata presa, si finisce qui.
        return f"[{self.state.categoria}] {self.state.azione}"


if __name__ == "__main__":
    messaggi = [
        "Buongiorno, vorrei sapere gli orari di apertura di sabato.",
        "La lavatrice nuova non funziona e perde acqua!",
        "HAI VINTO un telefono, clicca qui",
    ]
    for messaggio in messaggi:
        flusso = SmistaMessaggi()  # un flow nuovo, con uno stato nuovo, per ogni messaggio
        risultato = flusso.kickoff(inputs={"testo": messaggio})
        print(f"Esito: {risultato}\n")

Il ciclo for in fondo prende i tre messaggi di prova uno alla volta e per ciascuno crea un flow nuovo, lo avvia mettendo il messaggio nel campo testo dello stato e stampa l'esito. Lo esegui come gli altri script, dalla cartella in cui l'hai salvato:

uv run --env-file .env smista_messaggi.py

Per ogni messaggio CrewAI stampa dei riquadri: l'avvio del flow, poi «Flow Method Running» e «Flow Method Completed» per ogni passo che parte. Qui sotto la parte finale dell'esecuzione vera, quella del terzo messaggio: dopo classifica parte solo scarta, poi riepilogo.

Le tre righe «Esito» dell'esecuzione, una per messaggio, sono queste:

Esito: [normale] Risposta entro 24 ore lavorative.
Esito: [urgente] Avviso subito il responsabile dell'assistenza.
Esito: [spam] Messaggio archiviato senza risposta.

Guardando tutto il log noterai che a volte il riquadro «Running» di classifica compare prima del «Completed» di ricevi: i riquadri vengono stampati mentre il flow lavora e il loro ordine sullo schermo può mescolarsi un po'. L'ordine vero dei passi è quello del diagramma.

Un agente o una crew dentro un passo#

Ora il passo in più: quando il messaggio è normale, un Agente AIAgente AI: Programma che usa un LLM per decidere da solo i passi da fare verso un obiettivo: ragiona, usa strumenti, osserva il risultato e riprova finché ha finito. glossario scrive la bozza di risposta. Non serve una crew intera per un solo agente: si usa Agent.kickoff(), che fa lavorare un agente da solo e restituisce un risultato con il testo in .raw. Le regole di smistamento restano codice normale, così il modello lavora (e costa) solo quando serve.

# /// script
# requires-python = ">=3.10,<3.14"
# dependencies = ["crewai[tools]==1.15.21"]
# ///
"""Un flow che smista i messaggi con regole e fa scrivere a un agente la bozza di risposta."""
import os

from pydantic import BaseModel

from crewai import Agent
from crewai.flow.flow import Flow, listen, or_, router, start

# Il modello si sceglie nel file .env (capitolo 7). Se MODEL manca, si usa quello scritto qui.
MODELLO = os.getenv("MODEL", "openai/gpt-4.1-mini")


class StatoMessaggio(BaseModel):
    testo: str = ""
    categoria: str = ""
    bozza: str = ""


PAROLE_URGENTI = ["guasto", "bloccato", "rimborso", "urgente", "non funziona"]
PAROLE_SPAM = ["hai vinto", "clicca qui", "offerta imperdibile"]


class AssistenzaFlow(Flow[StatoMessaggio]):

    @start()
    def ricevi(self):
        self.state.testo = self.state.testo.strip()
        print(f"Ricevuto: {self.state.testo}")

    @router(ricevi)
    def classifica(self):
        # Le regole restano codice normale: veloci, gratuite, prevedibili.
        testo = self.state.testo.lower()
        if any(parola in testo for parola in PAROLE_SPAM):
            return "spam"
        if any(parola in testo for parola in PAROLE_URGENTI):
            return "urgente"
        return "normale"

    @listen("urgente")
    def passa_a_una_persona(self):
        self.state.categoria = "urgente"
        self.state.bozza = "Nessuna bozza automatica: il messaggio va a un responsabile."

    @listen("normale")
    def scrivi_bozza(self):
        # Solo qui entra in gioco il modello: un agente singolo, senza crew.
        self.state.categoria = "normale"
        addetto = Agent(
            role="Addetto all'assistenza clienti",
            goal="Scrivere risposte brevi, cortesi e precise",
            backstory="Rispondi ai clienti di un negozio di elettrodomestici. Non prometti nulla che non sai.",
            llm=MODELLO,
        )
        risultato = addetto.kickoff(
            f"Scrivi in italiano una risposta di massimo 3 frasi a questo messaggio: {self.state.testo}"
        )
        self.state.bozza = risultato.raw  # il testo scritto dall'agente finisce nello stato

    @listen("spam")
    def scarta(self):
        self.state.categoria = "spam"
        self.state.bozza = ""

    @listen(or_(passa_a_una_persona, scrivi_bozza, scarta))
    def riepilogo(self):
        return f"[{self.state.categoria}] {self.state.bozza}"


if __name__ == "__main__":
    flusso = AssistenzaFlow()
    risultato = flusso.kickoff(inputs={"testo": "Buongiorno, fate anche la consegna al piano?"})
    print("\n=== ESITO ===")
    print(risultato)

Questo script usa il modello, quindi serve il file .env del capitolo 7. Lo lanci con uv run --env-file .env assistenza_flow.py: in fondo vedrai [normale] seguito dalla bozza scritta dall'agente. Il testo cambierà a ogni esecuzione e con ogni modello.

Se invece in un passo serve una squadra, al posto dell'agente metti una crew. La catena dei dati è sempre la stessa: gli inputs del flow riempiono lo stato, lo stato riempie gli inputs della crew, questi riempiono i Segnaposto {…}Segnaposto {…}: Una parola tra graffe, come {citta}, dentro i testi di agenti e task. Al kickoff(inputs={...}) viene sostituita con il valore vero. glossario dei task.

    @listen("normale")
    def scrivi_bozza(self):
        # crew è una Crew costruita come nei capitoli 8-11, con {messaggio} nei task
        risultato = crew.kickoff(inputs={"messaggio": self.state.testo})
        self.state.bozza = risultato.raw

Quando usare l'uno o l'altra? Un agente con Agent.kickoff() quando ogni passo ha un solo specialista e vuoi decidere tu la sequenza; una crew quando più agenti devono collaborare sullo stesso pezzo di lavoro.

Altri attrezzi, in breve#

Una persona che approva: @human_feedback#

Il decoratore @human_feedback (serve CrewAI 1.8.0 o successivo) mette in pausa il flow, mostra il risultato di un passo e chiede un commento a una persona: è lo Human in the loopHuman in the loop: Un punto del processo in cui una persona controlla o approva prima che il lavoro prosegua. In CrewAI: human_input=True o @human_feedback. glossario dentro un flow. Con emit elenchi le strade possibili; un modello legge il commento scritto liberamente e lo traduce in una di quelle etichette.

from crewai.flow.human_feedback import human_feedback

    @human_feedback(
        message="Approvi questa bozza?",        # la domanda mostrata alla persona
        emit=["approvata", "da_rivedere"],        # le etichette possibili
        llm=MODELLO,                              # il modello che interpreta la risposta
        default_outcome="da_rivedere",            # la strada se non arriva un commento
    )
    @listen(scrivi_bozza)
    def revisione(self):
        return self.state.bozza                   # quello che la persona vede

    @listen("approvata")
    def invia(self):
        ...

Nel metodo che parte dopo, self.last_human_feedback contiene l'ultimo commento ricevuto. È il modo giusto per mettere un controllo umano prima di azioni come l'invio di un'email.

Riprendere da dove eri: @persist#

@persist() scritto sopra la classe salva lo stato dopo ogni passo in un piccolo database SQLite sul tuo computer. Se il programma si interrompe, puoi ripartire con kickoff(inputs={"id": "..."}) passando l'id dello stato salvato. Nella documentazione il decoratore compare anche senza parentesi (@persist), ma con CrewAI 1.15.21 quella forma, provata su un flow con stato strutturato, dà un TypeError appena Python legge la classe; con le parentesi funziona.

from crewai.flow.persistence import persist

@persist()   # salva lo stato dopo ogni passo
class SmistaMessaggi(Flow[StatoMessaggio]):
    ...

Vedere il flow: plot()#

flusso.plot("smistamento.html") disegna i passi e i collegamenti in una pagina web interattiva. Attenzione a una differenza tra documentazione e codice: la documentazione dice che il file viene salvato nella cartella corrente, ma in CrewAI 1.15.21 finisce in una cartella temporanea (con un nome come crewai_flow_...), si apre da solo nel browser e il metodo restituisce il percorso. Per sapere dov'è: print(flusso.plot("smistamento.html")). Scrivi il nome con .html: senza estensione il file viene creato senza, e il browser non lo apre come pagina.

Il progetto ufficiale: crewai create flow#

Nel capitolo 14 hai creato un progetto crew con la CLICLI: Command Line Interface: un programma che si usa dal terminale con comandi. CrewAI ne ha una: crewai create, crewai run… glossario. Per un flow il comando è:

# il nome diventa anche il nome della cartella del progetto
crewai create flow smistamento
cd smistamento
crewai install
crewai run

Il comando crea una cartella con src/smistamento/main.py (il flow, con uno stato Pydantic e tre passi), una crew di esempio in src/smistamento/crews/content_crew/ con i file YAMLYAML: Formato di file di testo per scrivere configurazioni in modo leggibile, con rientri e coppie chiave: valore. Nei progetti CrewAI classici descrive agenti e task. glossario di agenti e task, la cartella tools, il .env e il .gitignore. Il flow generato pianifica un argomento, fa scrivere un post alla crew e lo salva in output/post.md. Da lì modifichi i passi e le crew per il tuo caso; crewai flow plot disegna il flow del progetto. Se nel nome metti un trattino (mio-flow), in CrewAI 1.15.21 la CLI lo trasforma da sola in trattino basso (mio_flow), perché Python non accetta trattini nei nomi dei moduli. Per non confonderti tra il nome che hai scritto e quello della cartella, conviene usare direttamente il trattino basso.

PAROLE_RECLAMO = ["reclamo", "deluso"]

    @router(ricevi)
    def classifica(self):
        if any(parola in self.state.testo for parola in PAROLE_SPAM):
            return "spam"
        if any(parola in self.state.testo for parola in PAROLE_URGENTI):
            return "urgente"
        if any(parola in self.state.testo for parola in PAROLE_RECLAMO):
            return "reclamo"
        return "normale"

    @listen("reclamo")
    def gestisci_reclamo(self):
        self.state.categoria = "reclamo"
        self.state.azione = "Rispondo entro 4 ore con il responsabile qualità."

    @listen(or_(passa_a_una_persona, metti_in_coda, scarta, gestisci_reclamo))
    def riepilogo(self):
        return f"[{self.state.categoria}] {self.state.azione}"

Nel prossimo capitolo, l'ultimo di questa parte, vediamo come leggere gli errori, quanto costa far lavorare gli agenti e come usarli in sicurezza.

Corso indipendente, non affiliato a CrewAI Inc. Contenuti verificati su CrewAI 1.15.21 il 15 settembre 2026.