În fluxurile simple CrewAI, sarcinile sunt executate într-o secvență liniară (`Process.sequential`) sau ierarhică (`Process.hierarchical`). Cu toate acestea, scenariile din producție impun flexibilitate decizională: dacă un agent de audit detectează o vulnerabilitate critică, fluxul trebuie să declanșeze imediat o sarcină de remediere de urgență; dacă nu se găsesc erori, acea sarcină este omisă complet.
Funcționalitatea de Sarcini Condiționate (ConditionalTask) permite definirea unor funcții predicat în Python care evaluează ieșirea sarcinii precedente. În funcție de valoarea returnată (True/False), sarcina asociată este programată pentru execuție sau complet sărită.
Această arhitectură reduce masiv timpul total de execuție și consumul de tokeni prin eliminarea sarcinilor redundante și conferă echipei de agenți un comportament adaptiv orientat pe cazuri de utilizare reale.
Sintaxă, opțiuni și concepte cheie
| Opțiune / Concept | Descriere practică |
|---|---|
ConditionalTask |
Clasa specializată ce leagă o sarcină de o funcție condițională de validare. |
condition_func(output) |
Funcția Python care inspectează obiectul `TaskOutput` și returnează un boolean. |
TaskOutput.pydantic |
Accesarea datelor structurate validate pentru luarea de decizii tipizate sigure. |
Branching Strategy |
Mecanismul de ramificare prin care un agent ‘Router’ direcționează fluxul către echipe de specialiști. |
Exemple practice și configurații
Exemplul 1: Implementarea unei sarcini de alertă de securitate condiționată în Python
Dacă auditorul găsește probleme severe, agentul de securitate este trezit să trimită alertă.
from crewai import Agent, Crew, Task, Process
from crewai.tasks.conditional_task import ConditionalTask
def contine_vulnerabilitati(output):
# Evaluează rezultatul structurat al sarcinii de scanare
return "CRITICAL" in output.raw.upper()
auditor = Agent(role="Security Scanner", goal="Găsește breșe în cod", llm="claude-3-5-sonnet")
alertor = Agent(role="Incident Responder", goal="Trimite alerte PagerDuty", llm="claude-3-5-haiku")
task_scanare = Task(
description="Scanează dependentele din repo pentru breșe CVE.",
expected_output="Raport de securitate cu severitate indicată.",
agent=auditor
)
task_urgenta = ConditionalTask(
description="Transmite alerta critică pe Slack și declanșează PagerDuty.",
expected_output="Confirmarea expedierii alertei către echipă.",
agent=alertor,
condition=contine_vulnerabilitati
)
echipa = Crew(agents=[auditor, alertor], tasks=[task_scanare, task_urgenta])
echipa.kickoff()Rezultat & Impact
Task-ul de urgență rulează exclusiv dacă au fost detectate vulnerabilități critice reale.Greșeli frecvente și bune practici
- Parsarea ne-robustă a textului liber returnat de agent în predicatul de condiție, generând `False` din cauza unei mici variații sintactice; folosiți întotdeauna output structurat Pydantic.
- Crearea de cicluri infinite de condiții între doi agenți care depind reciproc de starea celuilalt.