10 esempi pratici

21Esempio 3 · Articoli e contenuti

Un flow redazionale con tre agenti: pianifica la scaletta, scrive l'articolo con un guardrail sulla lunghezza e lo fa revisionare, tornando indietro con le note finché non è approvato.

Tempo di lettura: 25 minuti

Il blog di un vivaio pubblica due articoli a settimana per chi ha un piccolo giardino. Il lavoro è sempre lo stesso: qualcuno decide la scaletta, qualcun altro scrive, un terzo rilegge e rimanda indietro la bozza con le correzioni. Spesso l'articolo fa due o tre giri prima di uscire.

In questo esempio riproduci quella redazione con un Flow: 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: tre agenti, un controllo automatico sulla lunghezza e un ciclo di revisione che torna indietro da solo quando serve, con un limite per non girare all'infinito.

Problema: Produrre articoli coerenti e riletti, senza che un testo debole arrivi alla pubblicazione.
La squadra: Un caporedattore che prepara la scaletta, un autore che scrive, un revisore che approva o chiede correzioni.
Strumenti: Nessuno strumento esterno: il lavoro è tutto di scrittura.
Chiavi e costi: Solo il modello. Funziona anche con Ollama sul tuo computer, gratis ma lento. Ogni giro di revisione aggiunge due chiamate al modello.
Cosa impari di nuovo: Agent.kickoff() dentro un passo del flow, il risultato strutturato del revisore con response_format, il guardrail su un agente, un ciclo con or_ e un limite di sicurezza.

Perché è tra i più usati#

È al terzo posto della classifica, con 54,8 punti, e i primi tre posti restano gli stessi con qualunque combinazione di pesi. È il caso più presente nel materiale ufficiale: ha il punteggio massimo (5 su 5). Tra gli esempi di CrewAI ci sono un flow che crea contenuti per blog e LinkedIn, uno che scrive un libro e uno con un ciclo di autovalutazione, simile a quello che costruisci qui. Compare anche in entrambi i corsi di DeepLearning.AI su CrewAI.

Su GitHub, al 15 settembre 2026, i repository con «crewai content» nel nome o nella descrizione sono 611, più 477 con «crewai blog». Tra le storie dei clienti, General Assembly dichiara una riduzione del 90% del tempo per preparare materiali didattici: è un dato fornito da CrewAI, non verificato da terzi.

Lo schema#

Qui non c'è una crew sola che fa tutto in fila: c'è un flow che decide cosa succede dopo ogni passo. Il cuore è il Router: Un passo di un flow che sceglie la strada successiva restituendo un'etichetta, per esempio "approvato" o "da_rivedere". glossario revisiona, che restituisce un'etichetta.

Il flow redazionale. Segui la freccia rossa in basso: se il revisore chiede correzioni, il flow torna a scrivi con le note. La freccia blu è la valvola di sicurezza dopo due revisioni.

Nota due cicli diversi. Il primo, in alto, è il Guardrail: Un controllo automatico sul risultato di un task. Se il controllo fallisce, l'errore torna all'agente che riprova. glossario: un controllo in Python che fa riscrivere subito l'autore se l'articolo è troppo corto o troppo lungo, prima ancora che lo veda il revisore. Il secondo, in basso, è il giudizio del revisore, che riguarda la qualità del testo. Il controllo meccanico costa pochissimo; il giudizio di un agente costa una chiamata in più al modello. Conviene fare prima quello meccanico.

Preparazione#

Crea una cartella redazione con un file .env. Nessun dato da scaricare.

.env

MODEL=openai/gpt-4.1-mini
OPENAI_API_KEY=sk-...la-tua-chiave...
CREWAI_DISABLE_TELEMETRY=true

Se preferisci non spendere nulla, usa il modello locale del capitolo 7: al posto delle prime due righe scrivi MODEL=ollama/qwen2.5:7b e tieni Ollama acceso. Con un modello di queste dimensioni aspettati tempi lunghi (ogni giro di revisione aggiunge due chiamate) e testi meno curati.

Il codice, pezzo per pezzo#

Le due forme dei dati#

Due classi Pydantic: Libreria Python per descrivere la forma dei dati (campi e tipi) e controllarla. CrewAI la usa per ottenere risposte strutturate e verificate. glossario descrivono i dati del flow. Revisione è la scheda che il revisore deve compilare: l'esito può essere solo "approvato" o "da_rivedere", grazie a Literal. StatoArticolo è lo Stato (di un flow): I dati che un flow porta con sé da un passo all'altro, per esempio l'argomento scelto e la bozza scritta. Si leggono e scrivono con self.state. glossario del flow: tutto quello che deve passare da un passo all'altro.

Python

MAX_GIRI = 2  # quante revisioni al massimo: una valvola di sicurezza contro i cicli infiniti


class Revisione(BaseModel):
    esito: Literal["approvato", "da_rivedere"]
    note: list[str] = Field(description="Correzioni concrete da fare, in italiano. Lista vuota se approvato.")


class StatoArticolo(BaseModel):
    argomento: str = ""
    pubblico: str = ""
    scaletta: str = ""
    bozza: str = ""
    note: list[str] = []
    giri: int = 0
    approvato: bool = False

La descrizione nel Field non è un commento per te: finisce nelle istruzioni al modello e gli spiega cosa mettere nelle note.

Il guardrail sulla lunghezza#

Il guardrail è una normale Funzione: Un blocco di codice con un nome che fa un lavoro e restituisce un risultato. Si definisce con def e si chiama con le parentesi: somma(2, 3). glossario, come nel capitolo 13. Conta le parole e restituisce una coppia: True e il testo se va bene, False e il motivo se no. Il motivo torna all'autore come istruzione per riprovare.

Python

def controlla_lunghezza(risultato):
    parole = len(risultato.raw.split())
    if parole < 300:
        return (False, f"L'articolo ha {parole} parole: allungalo fino ad almeno 300.")
    if parole > 600:
        return (False, f"L'articolo ha {parole} parole: accorcialo a non più di 600.")
    return (True, risultato.raw)

I tre agenti#

Ogni agente ha un mestiere e un modo di lavorare diverso. La novità è sull'autore: il guardrail si può attaccare anche a un agente, non solo a un task. Vale per ogni testo che l'agente produce, e guardrail_max_retries=2 dice quante volte può riprovare prima di arrendersi con un errore.

Python

pianificatore = Agent(
    role="Caporedattore di un blog divulgativo",
    goal="Preparare scalette chiare, con un'idea forte e un ordine logico",
    backstory="Da quindici anni decidi come raccontare argomenti difficili a chi non è del mestiere.",
    llm=MODELLO,
    verbose=True,
)

scrittore = Agent(
    role="Autore di articoli divulgativi",
    goal="Scrivere articoli precisi e piacevoli, seguendo la scaletta e le note del revisore",
    backstory="Scrivi frasi brevi, usi esempi concreti e non gonfi i numeri.",
    llm=MODELLO,
    guardrail=controlla_lunghezza,
    guardrail_max_retries=2,
    verbose=True,
)

revisore = Agent(
    role="Revisore editoriale",
    goal="Approvare solo articoli chiari, corretti e adatti al pubblico, indicando correzioni concrete",
    backstory="Sei severo ma giusto: segnali solo problemi reali e spieghi come risolverli.",
    llm=MODELLO,
    verbose=True,
)

Pianificare: un agente senza crew#

Nel primo passo non c'è né Crew né Task. Quando un passo del flow è il lavoro di un solo agente, la skill ufficiale di CrewAI consiglia di chiamare direttamente agente.kickoff("richiesta"): il flow si occupa già dell'ordine e dello stato, quindi una crew intorno a un solo task non aggiunge nulla. Il risultato ha lo stesso campo raw che conosci.

Python

class RedazioneFlow(Flow[StatoArticolo]):

    @start()
    def pianifica(self):
        risultato = pianificatore.kickoff(
            f"Prepara la scaletta di un articolo su: {self.state.argomento}.\n"
            f"Pubblico: {self.state.pubblico}.\n"
            "Scrivi in italiano: un titolo, una frase con l'idea principale e 4 punti in ordine."
        )
        self.state.scaletta = risultato.raw

La richiesta è una Stringa: Un testo nel codice, scritto tra virgolette: "Ricercatore". glossario costruita con i valori dello stato: qui non si usano i segnaposto tra graffe dei task, ma le f-string di Python (la f prima delle virgolette), che inseriscono subito i valori.

Scrivere, la prima volta e dopo le note#

Il passo scrivi parte in due casi: dopo pianifica oppure quando il router restituisce "da_rivedere". È quello che dice or_(pianifica, "da_rivedere"): «ascolta questo passo oppure questa etichetta». Se nello stato ci sono note, la richiesta include la bozza precedente e le correzioni da applicare.

Python

    @listen(or_(pianifica, "da_rivedere"))
    def scrivi(self):
        richiesta = (
            f"Scrivi in italiano un articolo di 300-600 parole per {self.state.pubblico}, "
            f"seguendo questa scaletta:\n{self.state.scaletta}"
        )
        if self.state.note:  # dal secondo giro in poi: correggi la bozza precedente
            richiesta += (
                f"\n\nEcco la bozza precedente:\n{self.state.bozza}\n\n"
                "Riscrivila applicando queste note del revisore:\n- " + "\n- ".join(self.state.note)
            )
        risultato = scrittore.kickoff(richiesta)
        self.state.bozza = risultato.raw

Revisionare e scegliere la strada#

Il router chiede al revisore un risultato strutturato con response_format=Revisione: la versione per Agent.kickoff() dell'output_pydantic dei task. Il valore arriva in risultato.pydantic. Poi il codice, non il modello, sceglie l'etichetta.

Python

    @router(scrivi)
    def revisiona(self):
        self.state.giri += 1
        risultato = revisore.kickoff(
            f"Rivedi questo articolo destinato a {self.state.pubblico}.\n\n{self.state.bozza}\n\n"
            "Rispondi con esito 'approvato' o 'da_rivedere' e, se serve, massimo 3 note in italiano.",
            response_format=Revisione,
        )
        revisione = risultato.pydantic
        if revisione is None:  # il modello non ha rispettato il formato: meglio fermarsi
            return "limite_raggiunto"
        self.state.note = revisione.note
        print(f"\n--- Revisione n. {self.state.giri}: {revisione.esito} ---")
        if revisione.esito == "approvato":
            self.state.approvato = True
            return "approvato"
        if self.state.giri >= MAX_GIRI:
            return "limite_raggiunto"
        return "da_rivedere"

Leggi l'ordine dei controlli: prima si guarda se il risultato esiste, poi se è approvato, poi se si è superato il limite. Senza MAX_GIRI un revisore mai soddisfatto farebbe girare il flow (e spendere token) per sempre.

Salvare#

L'ultimo passo ascolta due etichette. Se l'articolo non è stato approvato, lo salva lo stesso ma mette le note in cima, dentro un commento markdown, così chi lo rilegge sa cosa sistemare.

Python

    @listen(or_("approvato", "limite_raggiunto"))
    def salva(self):
        testo = self.state.bozza
        if not self.state.approvato:  # non approvato: lo segnaliamo in cima, per chi lo rilegge
            testo = "<!-- NON APPROVATO: controlla queste note -->\n- " + "\n- ".join(self.state.note) + "\n\n" + testo
        with open("articolo.md", "w", encoding="utf-8") as f:
            f.write(testo)
        return f"Articolo salvato in articolo.md dopo {self.state.giri} revisioni (approvato: {self.state.approvato})."

Al kickoff del flow, le chiavi di inputs riempiono i campi dello stato con lo stesso nome: argomento e pubblico.

Il codice completo#

redazione_articoli.py

# /// script
# requires-python = ">=3.10,<3.14"
# dependencies = ["crewai[tools]==1.15.21"]
# ///
"""Flow redazionale: pianifica, scrive e revisiona un articolo, con giri di correzione se servono."""
import os
from typing import Literal

from pydantic import BaseModel, Field

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")
MAX_GIRI = 2  # quante revisioni al massimo: una valvola di sicurezza contro i cicli infiniti


# Il giudizio del revisore, con due soli esiti possibili.
class Revisione(BaseModel):
    esito: Literal["approvato", "da_rivedere"]
    note: list[str] = Field(description="Correzioni concrete da fare, in italiano. Lista vuota se approvato.")


# Lo stato del flow: tutto quello che passa da un passo all'altro.
class StatoArticolo(BaseModel):
    argomento: str = ""
    pubblico: str = ""
    scaletta: str = ""
    bozza: str = ""
    note: list[str] = []
    giri: int = 0
    approvato: bool = False


# Guardrail: un controllo scritto in Python, senza modello. Se fallisce, lo scrittore riprova.
def controlla_lunghezza(risultato):
    parole = len(risultato.raw.split())
    if parole < 300:
        return (False, f"L'articolo ha {parole} parole: allungalo fino ad almeno 300.")
    if parole > 600:
        return (False, f"L'articolo ha {parole} parole: accorcialo a non più di 600.")
    return (True, risultato.raw)


pianificatore = Agent(
    role="Caporedattore di un blog divulgativo",
    goal="Preparare scalette chiare, con un'idea forte e un ordine logico",
    backstory="Da quindici anni decidi come raccontare argomenti difficili a chi non è del mestiere.",
    llm=MODELLO,
    verbose=True,
)

scrittore = Agent(
    role="Autore di articoli divulgativi",
    goal="Scrivere articoli precisi e piacevoli, seguendo la scaletta e le note del revisore",
    backstory="Scrivi frasi brevi, usi esempi concreti e non gonfi i numeri.",
    llm=MODELLO,
    guardrail=controlla_lunghezza,
    guardrail_max_retries=2,
    verbose=True,
)

revisore = Agent(
    role="Revisore editoriale",
    goal="Approvare solo articoli chiari, corretti e adatti al pubblico, indicando correzioni concrete",
    backstory="Sei severo ma giusto: segnali solo problemi reali e spieghi come risolverli.",
    llm=MODELLO,
    verbose=True,
)


class RedazioneFlow(Flow[StatoArticolo]):

    @start()
    def pianifica(self):
        risultato = pianificatore.kickoff(
            f"Prepara la scaletta di un articolo su: {self.state.argomento}.\n"
            f"Pubblico: {self.state.pubblico}.\n"
            "Scrivi in italiano: un titolo, una frase con l'idea principale e 4 punti in ordine."
        )
        self.state.scaletta = risultato.raw

    @listen(or_(pianifica, "da_rivedere"))
    def scrivi(self):
        richiesta = (
            f"Scrivi in italiano un articolo di 300-600 parole per {self.state.pubblico}, "
            f"seguendo questa scaletta:\n{self.state.scaletta}"
        )
        if self.state.note:  # dal secondo giro in poi: correggi la bozza precedente
            richiesta += (
                f"\n\nEcco la bozza precedente:\n{self.state.bozza}\n\n"
                "Riscrivila applicando queste note del revisore:\n- " + "\n- ".join(self.state.note)
            )
        risultato = scrittore.kickoff(richiesta)
        self.state.bozza = risultato.raw

    @router(scrivi)
    def revisiona(self):
        self.state.giri += 1
        risultato = revisore.kickoff(
            f"Rivedi questo articolo destinato a {self.state.pubblico}.\n\n{self.state.bozza}\n\n"
            "Rispondi con esito 'approvato' o 'da_rivedere' e, se serve, massimo 3 note in italiano.",
            response_format=Revisione,
        )
        revisione = risultato.pydantic
        if revisione is None:  # il modello non ha rispettato il formato: meglio fermarsi
            return "limite_raggiunto"
        self.state.note = revisione.note
        print(f"\n--- Revisione n. {self.state.giri}: {revisione.esito} ---")
        if revisione.esito == "approvato":
            self.state.approvato = True
            return "approvato"
        if self.state.giri >= MAX_GIRI:
            return "limite_raggiunto"
        return "da_rivedere"

    @listen(or_("approvato", "limite_raggiunto"))
    def salva(self):
        testo = self.state.bozza
        if not self.state.approvato:  # non approvato: lo segnaliamo in cima, per chi lo rilegge
            testo = "<!-- NON APPROVATO: controlla queste note -->\n- " + "\n- ".join(self.state.note) + "\n\n" + testo
        with open("articolo.md", "w", encoding="utf-8") as f:
            f.write(testo)
        return f"Articolo salvato in articolo.md dopo {self.state.giri} revisioni (approvato: {self.state.approvato})."


if __name__ == "__main__":
    flow = RedazioneFlow()
    esito = flow.kickoff(inputs={
        "argomento": "perché conviene raccogliere l'acqua piovana in giardino",
        "pubblico": "chi ha un piccolo giardino e nessuna esperienza",
    })
    print("\n=== ESITO ===")
    print(esito)

Eseguirlo#

Terminale

uv run --env-file .env redazione_articoli.py

Vedrai scorrere i riquadri dei tre agenti, uno dopo l'altro. Se il guardrail scatta, l'autore riceve il messaggio con il numero di parole e riscrive. Dopo ogni revisione compare la riga --- Revisione n. 1: da_rivedere --- (o approvato). Alla fine il flow stampa l'esito dell'ultimo passo e trovi il testo in articolo.md.

Come migliorarlo#

L'ultima parola a una persona. Invece di salvare quando si raggiunge il limite, fermati e chiedi a chi gestisce il blog: nel capitolo 16 hai visto come un flow può coinvolgere una persona (Human 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).
Un revisore più rigoroso. Aggiungi a Revisione un campo voto: int da 1 a 10 e approva nel router solo sopra una soglia decisa da te: la regola resta nel codice, non nel modello.
Un ricercatore prima del caporedattore. Per articoli con numeri e fatti, aggiungi in testa un passo con un agente che usa SerperDevTool, come nell'esempio 1, e passa i suoi appunti alla scaletta.
Modelli diversi per ruoli diversi. La scaletta e la revisione sono brevi e richiedono giudizio; la scrittura produce molto testo. Puoi dare un modello più potente al revisore e uno più economico all'autore, cambiando llm solo dove serve.
Più canali. Aggiungi un passo che, dall'articolo approvato, scrive un post breve per i social: è la variante «contenuti per più canali» degli esempi ufficiali.

Attenzione

Revisore e autore usano lo stesso tipo di modello: tendono ad avere gli stessi punti ciechi. Un revisore AI non sostituisce il controllo dei fatti. Se l'articolo contiene numeri, leggi e verifica prima di pubblicare.

I cicli costano. Ogni giro di revisione rilegge l'intera bozza. Tieni MAX_GIRI basso e guarda i token consumati quando provi un argomento nuovo.

Contenuti e responsabilità. Un testo generato può riprendere frasi viste altrove o dare consigli sbagliati. Chi pubblica ne risponde: dichiara l'uso dell'AI se il tuo contesto lo richiede.

Prova tu: un secondo controllo senza modello

Aggiungi un guardrail che rifiuta l'articolo se non comincia con un titolo markdown, cioè una riga che inizia con #. Tienilo insieme al controllo della lunghezza.

Soluzione. Il modo più semplice è una seconda funzione e un controllo che le chiama in fila: se la prima fallisce, restituisci subito il suo motivo.

Python

def controlla_titolo(risultato):
    if not risultato.raw.lstrip().startswith("#"):
        return (False, "Manca il titolo: la prima riga deve iniziare con # seguito dal titolo.")
    return (True, risultato.raw)


def controlla_articolo(risultato):
    for controllo in (controlla_lunghezza, controlla_titolo):
        ok, valore = controllo(risultato)
        if not ok:
            return (False, valore)
    return (True, risultato.raw)

Poi nell'autore scrivi guardrail=controlla_articolo. Anche i task accettano una lista di controlli con guardrails=[...], come hai visto nel capitolo 13.

In breve

I contenuti sono terzi in classifica e hanno il punteggio massimo nel materiale ufficiale di CrewAI.
Un flow con router permette di tornare indietro: or_(pianifica, "da_rivedere") riattiva la scrittura con le note.
Quando un passo è il lavoro di un solo agente, agente.kickoff() basta: niente crew intorno.
Prima il controllo meccanico (guardrail sulla lunghezza), poi il giudizio del revisore con response_format.
Un limite di giri è obbligatorio in ogni ciclo: evita loop infiniti e costi fuori controllo.

Nel prossimo esempio il flow smista le richieste dei clienti: una parte riceve una risposta basata sulle FAQ, i casi delicati passano a una persona.

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