CrewAI Flows: @start(), @listen(), @router() și Stare Tipizată în CrewAI & Sisteme Multi-Agent

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

class StareFlux(BaseModel):
    scor_calitate: int = 0

class PipelineArticol(Flow[StareFlux]):
    @start()
    def genereaza(self):
        self.state.scor_calitate = 85

    @router(genereaza)
    def verifica_scor(self):
        return "publica" if self.state.scor_calitate >= 80 else "rescrie"

Descriere & Utilizare

În aplicațiile complexe de producție, ai nevoie atât de autonomia agenților (Crews), cât și de controlul strict al codului clasic (apeluri API directe, interogări de baze de date, bucle `if/else` deterministe). CrewAI Flows este stratul de deasupra care le unește pe amândouă.

Un Flow moștenește Flow[ModelStarePydantic], menținând o stare puternic tipizată în self.state pe tot parcursul execuției. Fluxul este condus prin trei decoratori intuitivi: @start() (punctul de intrare), @listen(metoda_anterioara) (se declanșează automat când metoda ascultată s-a terminat) și @router(metoda) (ramificare condițională care returnează o etichetă string).

În plus, poți combina condiții logice complexe folosind funcțiile or_(metoda_a, metoda_b) și and_(metoda_a, metoda_b).

Cum să o reții ușor

💡 Dacă un Crew este o echipă creativă de brainstorming, un Flow este calea ferată cu macazuri automate (@router) și gări (@listen) care transportă rezultatul echipei în siguranță până la destinație.

Opțiuni și parametri tehnici

• @start(): marchează metoda (sau metodele paralele) care pornesc la apelul flow.kickoff().
• @listen(sursa): execută metoda decorată imediat ce sursa (o altă metodă sau o etichetă returnată de un @router) s-a finalizat.
• @router(sursa): evaluează self.state și returnează un string (ex: "aprobat" sau "respins") pe care îl pot asculta alte metode @listen("aprobat").
• @persist(): salvează automat starea self.state pe disc/DB la fiecare pas pentru reluarea fluxurilor întrerupte.

Exemple practice și cazuri de utilizare

• Ascultarea unei ramuri din router: @listen("publica") def trimite_in_wordpress(self): ...
• Așteptarea finalizării a două echipe paralele înainte de sinteză: @listen(and_(echipa_juridica, echipa_financiara)) def raport_final(self): ...
• Vizualizarea grafului Flow într-un fișier HTML interactiv: flow.plot("arhitectura_flow.html")

Sfaturi de configurare & capcane

Folosește întotdeauna o clasă Pydantic BaseModel în Flow[StareaMea] în locul stării nestructurate — vei avea autocompletare în IDE pe self.state și validare garantată a tipurilor între pași.

Comenzi și ghiduri înrudite

• crewai-process-sequential-hierarchical: Resursă asociată din acest modul.
• crewai-async-batch-kickoff: Resursă asociată din acest modul.
• crewai-pydantic-structured-output: Resursă asociată din acest modul.