CrewAI Flows: orchestrazione deterministica degli agenti

Perché esistono i Flows

Un Crew delega la strategia agli agenti: tu dichiari obiettivo e ruoli, loro decidono i passaggi. È la scelta giusta quando il percorso non è noto in anticipo, ma ha un costo che in produzione diventa pesante: due esecuzioni con lo stesso input possono seguire strade diverse, consumare token diversi e produrre output diversi.

Il problema che i Flows risolvono è esattamente questo: la riproducibilità. Se devi spiegare a un cliente perché il sistema ha risposto così, se devi scrivere un test che passa in modo affidabile, se devi tenere il costo per esecuzione dentro un intervallo prevedibile, l'autonomia degli agenti smette di essere un vantaggio e diventa un rischio.

Un Flow toglie agli agenti il controllo della struttura. È una classe che estende Flow, con metodi Python normali collegati da eventi: il completamento di un metodo innesca il successivo, una condizione scritta da te sceglie il ramo, lo stato passa di mano in modo esplicito. Dentro un singolo passo puoi ancora chiamare un agente o un intero Crew, ma la mappa del percorso resta nel codice, leggibile e testabile.

I decoratori: start, listen, router

L'intera API dei Flows sta in una manciata di decoratori, importati da crewai.flow.flow.

• @start() marca i punti d'ingresso. Possono essere più di uno: all'avvio partono tutti i metodi @start() le cui condizioni sono soddisfatte, anche in parallelo.
• @listen() aggancia un metodo al completamento di un altro. Accetta il riferimento al metodo, @listen(genera), oppure il suo nome come stringa, @listen("genera"). L'output del metodo ascoltato arriva come argomento.
• @router() trasforma un metodo in un bivio: il valore restituito è un'etichetta, e proseguono solo i metodi in ascolto su quell'etichetta.

Quando un passo deve aspettare più sorgenti, le funzioni or_() e and_() combinano i trigger: la prima fa partire il metodo appena uno degli eventi arriva, la seconda solo quando sono arrivati tutti.

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

class Pipeline(Flow):

    @start()
    def scarica_dati(self):
        self.state["origine"] = "api"

    @start()
    def leggi_archivio(self):
        self.state["archivio"] = "locale"

    @listen(or_(scarica_dati, leggi_archivio))
    def appena_possibile(self):
        # parte quando UNO dei due sopra ha finito
        print("primo evento ricevuto")

    @listen(and_(scarica_dati, leggi_archivio))
    def quando_tutto_pronto(self):
        # parte solo quando ENTRAMBI hanno finito
        print(self.state)

Lo stato: libero o tipizzato

Ogni Flow porta con sé un oggetto self.state condiviso fra tutti i passi. Puoi usarlo in due modi, e la scelta va fatta presto perché cambia il modo in cui scrivi ogni metodo.

Con lo stato non strutturato dichiari la classe come Flow senza parametro di tipo e accedi ai campi come in un dizionario: self.state["contatore"]. Nessuno schema da definire, massima velocità di prototipazione. Con lo stato strutturato definisci un modello Pydantic e lo passi come parametro generico, Flow[TuoStato]: i campi diventano attributi tipizzati, self.state.contatore, con validazione e completamento automatico nell'editor. In entrambi i casi il framework aggiunge allo stato un campo id, un UUID che identifica l'esecuzione e che serve alla persistenza.

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

class Libero(Flow):                    # stato non strutturato
    @start()
    def apri(self):
        self.state["contatore"] = 0     # accesso da dizionario
        print(self.state["id"])        # UUID aggiunto dal framework

class StatoReport(BaseModel):         # schema dello stato
    argomento: str = ""
    revisioni: int = 0

class Tipizzato(Flow[StatoReport]):  # stato strutturato
    @start()
    def apri(self):
        self.state.argomento = "agenti AI"
        self.state.revisioni += 1
        print(self.state.id)

Persistenza e ripresa con @persist

Il decoratore @persist, importato da crewai.flow.persistence, salva lo stato dopo ogni passo. Applicato alla classe copre tutti i metodi; applicato a un singolo metodo copre solo quello. Il backend predefinito è SQLiteFlowPersistence, un database SQLite locale che non richiede configurazione.

La ripresa passa dal campo id dello stato: chiamando kickoff(inputs={"id": ...}) con un identificativo già salvato, il Flow ricarica lo stato e continua sotto lo stesso UUID invece di ripartire da zero. Esiste anche la variante kickoff(restore_from_state_id=...), che idrata un nuovo flusso dai dati salvati ma gli assegna un id nuovo: utile per clonare un'esecuzione senza toccare l'originale. Per un processo che può fermarsi a metà, per un errore di rete o per un limite di API, questa è la differenza fra ripetere tutto e ripartire dal punto giusto.

Un Flow completo: il router in azione

L'esempio che segue è il pattern più utile da imparare: una richiesta in arrivo viene classificata da una regola deterministica, e solo il ramo che richiede analisi vera attiva un Crew. Il ramo semplice risponde senza toccare un modello, quindi senza costi e senza latenza.

from pydantic import BaseModel
from crewai import Agent, Crew, Process, Task
from crewai.flow.flow import Flow, listen, router, start

class StatoTriage(BaseModel):
    richiesta: str = ""
    risposta: str = ""

class SupportoFlow(Flow[StatoTriage]):

    @start()
    def ricevi(self):
        # in produzione arriva da un form o da una coda
        self.state.richiesta = "Il sito è irraggiungibile da un'ora"

    @router(ricevi)
    def classifica(self):
        # regola deterministica: niente modello, niente token
        testo = self.state.richiesta.lower()
        if "irraggiungibile" in testo or "errore" in testo:
            return "incidente"
        return "informazione"

    @listen("informazione")
    def risposta_diretta(self):
        # ramo senza agenti: costo zero, esito identico ogni volta
        self.state.risposta = "Trovi la procedura nella guida online."
        return self.state.risposta

    @listen("incidente")
    def analisi_con_crew(self):
        # solo qui serve autonomia: entra un Crew
        tecnico = Agent(
            role="Tecnico di supporto",
            goal="Diagnosticare la segnalazione: {richiesta}",
            backstory="Isola la causa prima di proporre soluzioni.",
            llm="gpt-5.4-mini",
        )
        diagnosi = Task(
            description="Analizza la segnalazione: {richiesta}",
            expected_output="Causa probabile e primo passo operativo.",
            agent=tecnico,
        )
        crew = Crew(agents=[tecnico], tasks=[diagnosi],
                    process=Process.sequential)
        self.state.risposta = crew.kickoff(
            inputs={"richiesta": self.state.richiesta}
        ).raw
        return self.state.risposta

flow = SupportoFlow()
esito = flow.kickoff()   # restituisce l'output dell'ultimo metodo completato
print(esito)
print(flow.state)        # lo stato resta leggibile dopo l'esecuzione

Il punto da notare è dove sta la decisione. Il metodo classifica è un router, ma la logica dentro è una condizione Python su parole chiave: identica a ogni esecuzione, verificabile con un test unitario, gratuita. Nulla vieta di far classificare un modello, quando le regole non bastano: il router resta deterministico nella struttura, perché i rami possibili sono comunque solo quelli che hai dichiarato con @listen.

Come si combinano Crew e Flow

Questo è il pattern che conta, e l'esempio sopra lo mostra già: il Flow governa, il Crew lavora. Ogni passo che può essere codice puro resta codice puro; Crew(...).kickoff() si chiama solo dentro i metodi dove serve davvero un agente. Nei progetti generati con lo scaffolding ufficiale il crew vive in una classe separata e si richiama con la forma PoemCrew().crew().kickoff(inputs=...), ma la sostanza non cambia: il Crew è un ingranaggio dentro un passo, non il motore del processo.

Il criterio di progetto è economico prima che tecnico. Ogni chiamata a un Crew costa token e introduce varianza; ogni passo deterministico costa zero e si testa in un millisecondo. Un buon Flow comprime l'autonomia nei punti dove rende, recupero di informazioni, sintesi, valutazione di testi, e tiene tutto il resto sotto controllo diretto: validazione degli input, instradamento, formattazione, invio del risultato.

Quando i Flows non servono

Se il tuo processo è una sequenza fissa di compiti, ognuno che passa l'output al successivo, un Crew con Process.sequential fa già esattamente questo, con meno codice e meno concetti. Un Flow fatto solo di @listen in fila indiana è un modo elaborato di scrivere una sequenza che il framework sapeva già eseguire.

I segnali che invece giustificano un Flow sono concreti:

• Rami condizionali: il percorso dipende da un valore calcolato a runtime, e vuoi dichiarare i rami possibili.
• Passi senza modello: metà del processo è lettura di dati, chiamate ad API, regole di business che non hanno bisogno di un LLM.
• Stato da riprendere: il processo può interrompersi e deve ripartire dal punto giusto, non dall'inizio.
• Tracciabilità: devi poter dire quale ramo è stato preso e perché, con lo stato alla mano.

Se nessuno di questi punti ti riguarda, la complessità in più non compra niente. Parti dal Crew, e passa ai Flow quando il bisogno di controllo si presenta da solo.

Il confronto onesto con LangGraph

I Flows occupano lo stesso territorio di LangGraph: orchestrazione esplicita, stato gestito, rami dichiarati nel codice. La differenza vera è il modello mentale. Un Flow si scrive per eventi: decori i metodi, e il grafo emerge dai collegamenti fra decoratori. LangGraph si scrive per grafo: dichiari nodi e archi in modo esplicito, e la struttura è un oggetto che costruisci pezzo per pezzo.

Per chi usa già CrewAI, i Flows hanno il vantaggio di restare nello stesso framework e nella stessa testa: un decoratore sopra un metodo si legge in un secondo. Per processi con cicli complessi, checkpoint granulari e controllo fine su ogni transizione, il grafo esplicito di LangGraph resta lo strumento più preciso, al prezzo di più codice di struttura. Il confronto dedicato mette i due approcci fianco a fianco sullo stesso caso d'uso.

Come si esegue e si visualizza

Da Python, un Flow si istanzia e si avvia con kickoff(), che restituisce l'output dell'ultimo metodo completato; dopo l'esecuzione lo stato resta accessibile su flow.state. Il metodo flow.plot("nome") genera un file nome.html con il grafo interattivo dei passi: prima di eseguire un Flow nuovo, guardarlo disegnato è il modo più rapido per accorgersi di un collegamento mancante.

Da terminale, i comandi sono questi:

crewai create flow assistente   # scaffolding di un progetto flow
crewai run                      # esegue: riconosce il flow dal pyproject.toml
crewai flow kickoff             # funziona ancora, ma è deprecato
crewai flow plot                # grafo interattivo dei passi in HTML

Un dettaglio da sapere: crewai flow kickoff esiste ed è il comando storico, ma la documentazione lo marca come legacy e raccomanda crewai run, che dalla versione 0.103.0 riconosce da solo se il progetto è un crew o un flow. Se trovi tutorial che usano ancora il comando vecchio, funzionano, ma non è la strada indicata oggi.