Гайд · TNWS AI

Как объединить события CrewAI Flow через or_ и and_

26 сентября 2026 4 мин—

Проверяемая разница or_ и and_ в CrewAI Flow: запуск после любого события или после всех, пример состояния и критерии выбора.

Задача и применимость

Актуальный поисковый запрос: «как использовать or_ и and_ в CrewAI Flow». Выбрать правильную семантику синхронизации. Официальная документация определяет or_ как запуск listener после любого из указанных методов, а and_ — только после того, как результаты испустили все источники. В примере последовательной зависимости and_ гарантирует заполнение обоих полей; or_ подходит для общего журнала нескольких событий, но может вызваться несколько раз. Материал сверён 26 сентября 2026 года с официальной документацией CrewAI 1.15.22. Примеры предназначены для Python 3.10+; закрепите версию пакета, потому что сигнатуры и поведение развиваются. Все коды, статусы и даты в примерах вымышлены.

Пошаговая настройка

Перечислите данные, которые действительно обязательны следующему шагу.
Для ожидания всех источников используйте and_.
Для реакции на любое событие используйте or_ и делайте обработчик идемпотентным.
Запишите порядок и количество вызовов в тесте.

Перед запуском установите CrewAI по официальной инструкции. Для примеров с агентом задайте CREWAI_MODEL и учётные данные провайдера через переменные среды; доступность модели зависит от вашего аккаунта. Flow-примеры не требуют LLM и должны сначала проходить как детерминированные тесты. Не включайте действующие ключи и реальные клиентские данные в исходник или prompt.

Готовый шаблон

from crewai.flow.flow import Flow, and_, listen, start

class JoinFlow(Flow):
    @start()
    def first(self):
        self.state["code"] = "R-7"

    @listen(first)
    def second(self):
        self.state["status"] = "готово"

    @listen(and_(first, second))
    def join(self):
        return f'{self.state["code"]}:{self.state["status"]}'

assert JoinFlow().kickoff() == "R-7:готово"

Конкретный вход и ожидаемый результат

first записывает код, second — статус. join должен выполниться после обоих событий и вернуть R-7:готово. Если заменить and_ на or_, обработчик способен запуститься уже после первого события, когда status ещё отсутствует, и затем снова после второго. Это не ошибка библиотеки, а другая требуемая семантика.

Как фактически проверить

Добавьте массив events и фиксируйте вход каждого метода. Для and_ ожидается один вызов join после двух источников. Сделайте отдельный учебный Flow с or_ и подтвердите два возможных вызова на официальном шаблоне. Выбирайте конструкцию по данным, а не по удобству импорта.

Проверяйте не только итоговую строку, но и путь исполнения: входные данные, вызванный метод или Task, промежуточный результат и финальное преобразование. Успешный Python-процесс не доказывает правильность фактов LLM. Для агентных сценариев сделайте не менее трёх прогонов фиксированной модели и укажите долю прохождения критериев, а не выбирайте лучший ответ вручную.

Критерии приёмки

Python-код проходит синтаксическую проверку, а зависимости и версия зафиксированы.
Положительный пример даёт заранее описанный результат без новых фактов.
Отрицательный пример завершается явным отказом либо контролируемой веткой.
Ни один шаг не получает больше данных или полномочий, чем требуется задаче.
После обновления CrewAI или модели тот же набор выполняется повторно.

Протокол воспроизводимости

Сохраните дату, версию CrewAI, версию Python, идентификатор модели для agent-примеров и обезличенный вход. Для Flow зафиксируйте порядок событий и финальный state; для Task — TaskOutput, число попыток и результат guardrail. Не записывайте секреты, системные prompts и персональные данные без необходимости. Если ошибка плавающая, классифицируйте её как неверный маршрут, неверный аргумент, исключение Python или неподтверждённую генерацию — эти случаи исправляются по-разному.

Регрессионный набор

Не ограничивайтесь одним удачным примером. Добавьте стандартный вход, отсутствующее обязательное поле, похожий идентификатор, пустую строку и слишком длинное значение. Для каждого заранее укажите ожидаемое событие, состояние и финальный результат. В агентных задачах отдельно оценивайте правильность фактов и соблюдение формата; красивый текст не компенсирует неверный код. В Flow проверяйте, какой метод действительно вызван и сколько раз: одинаковая финальная строка может скрыть ошибочный маршрут. При изменении модели не переносите прежние показатели автоматически. При изменении схемы state предусмотрите миграцию сохранённых записей. Ошибка одного теста должна блокировать автоматическое внешнее действие, но не обязана останавливать независимые процессы системы. После исправления обязательно повторите весь набор, а не только упавший случай. Такой набор превращает статью в повторяемую инструкцию, а не в демонстрацию единственного счастливого пути.

Типичные ошибки и чего не делать

Не используйте or_, если обработчику нужны результаты всех источников.
Не делайте or_-listener неидемпотентной операцией записи.
Не предполагайте параллельность только из-за нескольких источников событий.

Не увеличивайте retries, max_iter или таймаут после каждой ошибки автоматически. Ограничение останавливает цикл, но не исправляет отсутствующий факт или плохую схему. Не выполняйте платежи, удаление, публикацию и другие необратимые действия только на основании текста модели. Добавляйте детерминированную проверку и, где требуется, человеческое подтверждение.

FAQ

Когда срабатывает or_?

Когда результат испустил любой из перечисленных методов.

Когда срабатывает and_?

Только после результатов всех перечисленных методов.

Может ли or_ вызвать listener несколько раз?

Да, на разные события; обработку проектируют с учётом этого.

Официальные источники

Частые вопросы

Когда срабатывает or_?

Когда результат испустил любой из перечисленных методов.

Когда срабатывает and_?

Только после результатов всех перечисленных методов.

Может ли or_ вызвать listener несколько раз?

Да, на разные события; обработку проектируют с учётом этого.

Как объединить события CrewAI Flow через or_ и and_

Задача и применимость

Пошаговая настройка

Готовый шаблон

Конкретный вход и ожидаемый результат

Как фактически проверить

Критерии приёмки

Протокол воспроизводимости

Регрессионный набор

Типичные ошибки и чего не делать

FAQ

Когда срабатывает or_?

Когда срабатывает and_?

Может ли or_ вызвать listener несколько раз?

Официальные источники

Частые вопросы

Читайте также

Как хранить данные в словарном state CrewAI Flow

Как передать результат одной задачи CrewAI в другую через context

Как получить Pydantic-результат задачи CrewAI через output_pydantic

Комментарии