Flows:
event-driven оркестрация

Когда Crew мало

Crew хорош, пока работа линейна (sequential) или сводится к делегированию (hierarchical). Но реальные процессы часто ветвятся и зависят от результатов:

• Условия: «если критик одобрил — публикуем, иначе — на доработку».
• Маршрутизация: «классифицируй запрос → запусти разную команду под тип (код / текст / данные)».
• Смесь с обычным кодом: между шагами нужно сходить в БД, дёрнуть API, преобразовать данные — без агентов.
• Несколько Crew: оркестрировать команды, а не только задачи внутри одной.

Всё это — про поток управления, которого у плоского Process нет. Flow добавляет этот слой: он оркеструет шаги (функции, в том числе запускающие целые Crew) по событиям и условиям. Crew остаётся «рабочей лошадкой» внутри шага, а Flow — дирижёром над ними.

Строительные блоки Flow

Flow — это класс, унаследованный от Flow, где методы помечены декораторами, задающими событийные связи.

• @start() — точка входа Flow. С неё начинается выполнение.
• @listen(X) — «запусти этот шаг, когда завершился шаг X». Это событийная связь = ребро.
• @router(X) — после шага X ветвится: возвращает строку-метку, по которой выбирается следующий @listen (= conditional edge).
• self.state — общее состояние Flow, доступное всем шагам (структурированное через Pydantic или словарь).

Пример: генерация с проверкой и ветвлением

Соберём Flow, который генерирует текст, проверяет его и в зависимости от вердикта либо публикует, либо отправляет на доработку. Это то, что плоский Crew не выразит, — есть условие и петля.

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

class ContentState(BaseModel):       # общее состояние Flow
    topic: str = ""
    draft: str = ""
    approved: bool = False
    attempts: int = 0

class ContentFlow(Flow[ContentState]):

    @start()
    def generate(self):
        # тут можно запустить целый Crew: result = ContentCrew().crew().kickoff(...)
        self.state.draft = f"Черновик про {self.state.topic}"
        self.state.attempts += 1

    @router(generate)
    def review(self):
        # проверяем черновик (тоже можно через Crew-критика)
        self.state.approved = check_quality(self.state.draft)
        if self.state.approved:
            return "approved"
        return "rejected" if self.state.attempts < 3 else "approved"  # лимит

    @listen("approved")
    def publish(self):
        return f"ОПУБЛИКОВАНО: {self.state.draft}"

    @listen("rejected")
    def revise(self):
        self.state.draft += " [доработано]"
        self.generate()          # ещё круг (петля, как в Reflection)

# Запуск
flow = ContentFlow()
flow.kickoff(inputs={"topic": "рынок электромобилей 2025"})

Разберём поток: generate (старт) создаёт черновик → router проверяет и возвращает метку → по ней срабатывает publish (для «approved») или revise (для «rejected»), и тот гонит на новый круг. Лимит попыток в роутере не даёт зациклиться — прямой аналог recursion_limit и критерия остановки из Reflection (модуль 03).

Общее состояние и слияние событий

self.state — это память Flow между шагами: один шаг пишет, другой читает (как State в LangGraph). Структурируй его через Pydantic — получишь типизацию и валидацию (привет уроку про state schema из модуля 03).

Помимо @listen на один шаг, Flow умеет слияние событий: шаг может ждать нескольких предшественников через and_(...) (запустись, когда завершились все) или or_(...) (когда любой). Это даёт fan-in для параллельных веток — как сборка в map-reduce.

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

@listen(and_(fetch_news, fetch_prices))   # запустится, когда ОБА завершились
def combine(self):
    return merge(self.state.news, self.state.prices)

@listen(or_(path_a, path_b))              # запустится, когда ЛЮБОЙ завершился
def proceed(self):
    ...

Crew или Flow

Типичные ошибки

Шпаргалка

ИДЕЯ: Flow = событийный граф ПОВЕРХ Crew (ветвления, условия, петли, код)
  Flow ≈ граф (модуль 03) · Crew ≈ узел/шаг

БЛОКИ:
  @start()        — точка входа
  @listen(X)      — запусти, когда завершился X  (= ребро)
  @router(X)      — после X верни метку → выбор ветки  (= conditional edge)
  self.state      — общее состояние (Pydantic), память между шагами
  and_(...) / or_(...) — слияние: ждать всех / любого  (= fan-in)

ВНУТРИ ШАГА: можно запустить целый Crew →
  Flow рулит потоком, Crew делает командную работу

CREW vs FLOW:
  Crew  — линейно/делегирование, проще
  Flow  — ветвления/условия/петли/несколько команд/код, мощнее

ПРАВИЛА:
  • начинай с Crew; Flow — когда нужен реальный поток управления
  • петли → счётчик в state + условие выхода (как Reflection)
  • метки @router == строки @listen
  • состояние только через self.state

Практическое задание

Собери ветвящийся процесс:

Что дальше

На этом раздел CrewAI закрыт: ты знаешь сущности (Crew/Agent/Task/Process), ролевых агентов, два процесса, YAML-конфигурацию и событийные Flows. Дальше — отвлечёмся от конкретных фреймворков и систематизируем паттерны координации, которые встречались по всему модулю: Supervisor, Pipeline, Swarm, Debate.