Sarcini condiționate în CrewAI (ConditionalTask) în CrewAI & Sisteme Multi-Agent

ConditionalTask(description=..., expected_output=..., condition=functie, agent=...)

Ce face Sarcini condiționate în CrewAI (ConditionalTask)

ConditionalTask adaptează fluxul unei echipe după rezultatul pașilor anteriori. Pe lângă câmpurile unui Task obișnuit, primește parametrul condition: o funcție care primește TaskOutput-ul task-ului anterior și întoarce True sau False.

Dacă funcția întoarce True, task-ul condiționat rulează; dacă întoarce False, este sărit. Cu output_pydantic pe task-ul anterior, condiția poate citi câmpuri tipizate în loc de text liber.

Într-un proiect JSON-first, un task condiționat se scrie în crew.jsonc cu "type": "ConditionalTask" și un câmp condition. Pentru ramificări mai complexe, documentația folosește Flows cu @router.

Sintaxă și opțiuni

Opțiune / Parametru	Descriere
ConditionalTask	Clasa din crewai.tasks.conditional_task.
condition	Funcție (TaskOutput) -> bool. False înseamnă că task-ul se sare.
TaskOutput.pydantic	Ieșirea structurată a task-ului anterior, dacă acesta are output_pydantic.

Exemple practice

Exemplul 1: Task rulat doar dacă lipsesc date

Adaptat după exemplul oficial: al doilea task rulează doar dacă primul a găsit mai puțin de 10 evenimente.

from typing import List
from pydantic import BaseModel
from crewai import Agent, Crew, Task
from crewai.tasks.conditional_task import ConditionalTask
from crewai.tasks.task_output import TaskOutput

class EventOutput(BaseModel):
    events: List[str]

def is_data_missing(output: TaskOutput) -> bool:
    return len(output.pydantic.events) < 10

task1 = Task(
    description="Fetch data about events in San Francisco",
    expected_output="List of 10 things to do in SF this week",
    agent=data_fetcher_agent,
    output_pydantic=EventOutput,
)
conditional_task = ConditionalTask(
    description="Fetch more events so that we have a total of 10.",
    expected_output="List of 10 things to do in SF this week",
    condition=is_data_missing,
    agent=data_processor_agent,
)
crew = Crew(agents=[data_fetcher_agent, data_processor_agent], tasks=[task1, conditional_task])

Greșeli frecvente și bune practici

• Scrii condiția pe textul liber al răspunsului. O mică variație de formulare schimbă rezultatul; folosește output_pydantic și citește câmpuri tipizate.
• Numești parametrul condition_func. Parametrul se numește condition.

Comenzi înrudite

• crewai-task-context
• crewai-pydantic-structured-output
• crewai-flows-start-listen-router
• crewai-testing