Гайд · TNWS AI
Как сделать типизированный state CrewAI Flow через Pydantic
Типизированное состояние CrewAI Flow на BaseModel: code, attempts, изменение между шагами, проверка типов и отличие от словаря.
Задача и применимость
Актуальный поисковый запрос: «как создать structured state CrewAI Flow Pydantic». Зафиксировать поля и типы состояния для поддерживаемого workflow. Официальное руководство предлагает Flow[StateModel], где StateModel наследуется от Pydantic BaseModel и имеет значения по умолчанию. Доступ идёт через атрибуты self.state.code, что помогает IDE и даёт runtime-валидацию при создании модели, но не подтверждает фактическую правильность строк. Материал сверён 26 сентября 2026 года с официальной документацией CrewAI 1.15.22. Примеры предназначены для Python 3.10+; закрепите версию пакета, потому что сигнатуры и поведение развиваются. Все коды, статусы и даты в примерах вымышлены.
Пошаговая настройка
- Опишите BaseModel с безопасными значениями по умолчанию.
- Параметризуйте Flow моделью состояния.
- Изменяйте поля через атрибуты в start и listener.
- Проверьте финальное состояние и отдельно попытку передать неверный тип.
Перед запуском установите CrewAI по официальной инструкции. Для примеров с агентом задайте CREWAI_MODEL и учётные данные провайдера через переменные среды; доступность модели зависит от вашего аккаунта. Flow-примеры не требуют LLM и должны сначала проходить как детерминированные тесты. Не включайте действующие ключи и реальные клиентские данные в исходник или prompt.
Готовый шаблон
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
class CheckState(BaseModel):
code: str = ""
attempts: int = 0
class TypedFlow(Flow[CheckState]):
@start()
def initialize(self):
self.state.code = "R-7"
self.state.attempts = 1
@listen(initialize)
def finish(self):
self.state.attempts += 1
return self.state.model_dump()
flow = TypedFlow()
result = flow.kickoff()
assert result["code"] == "R-7"
assert result["attempts"] == 2
Конкретный вход и ожидаемый результат
После start состояние равно code='R-7', attempts=1; listener увеличивает attempts. Ожидаемый словарь — {'code':'R-7','attempts':2}. Строка attempts='два' нарушает предметный тип, но точное поведение присваивания зависит от настроек Pydantic; надёжнее создавать и валидировать модели на границах входа.
Как фактически проверить
Создайте отдельный CheckState(code='R-7', attempts=0) и негативный вход с недопустимым значением. Проверьте model_dump после каждого важного перехода. Если state восстанавливается из хранения, повторно валидируйте схему после обновления версии. Не переименовывайте поле без миграции сохранённых данных.
Проверяйте не только итоговую строку, но и путь исполнения: входные данные, вызванный метод или Task, промежуточный результат и финальное преобразование. Успешный Python-процесс не доказывает правильность фактов LLM. Для агентных сценариев сделайте не менее трёх прогонов фиксированной модели и укажите долю прохождения критериев, а не выбирайте лучший ответ вручную.
Критерии приёмки
- Python-код проходит синтаксическую проверку, а зависимости и версия зафиксированы.
- Положительный пример даёт заранее описанный результат без новых фактов.
- Отрицательный пример завершается явным отказом либо контролируемой веткой.
- Ни один шаг не получает больше данных или полномочий, чем требуется задаче.
- После обновления CrewAI или модели тот же набор выполняется повторно.
Протокол воспроизводимости
Сохраните дату, версию CrewAI, версию Python, идентификатор модели для agent-примеров и обезличенный вход. Для Flow зафиксируйте порядок событий и финальный state; для Task — TaskOutput, число попыток и результат guardrail. Не записывайте секреты, системные prompts и персональные данные без необходимости. Если ошибка плавающая, классифицируйте её как неверный маршрут, неверный аргумент, исключение Python или неподтверждённую генерацию — эти случаи исправляются по-разному.
Регрессионный набор
Не ограничивайтесь одним удачным примером. Добавьте стандартный вход, отсутствующее обязательное поле, похожий идентификатор, пустую строку и слишком длинное значение. Для каждого заранее укажите ожидаемое событие, состояние и финальный результат. В агентных задачах отдельно оценивайте правильность фактов и соблюдение формата; красивый текст не компенсирует неверный код. В Flow проверяйте, какой метод действительно вызван и сколько раз: одинаковая финальная строка может скрыть ошибочный маршрут. При изменении модели не переносите прежние показатели автоматически. При изменении схемы state предусмотрите миграцию сохранённых записей. Ошибка одного теста должна блокировать автоматическое внешнее действие, но не обязана останавливать независимые процессы системы. После исправления обязательно повторите весь набор, а не только упавший случай. Такой набор превращает статью в повторяемую инструкцию, а не в демонстрацию единственного счастливого пути.
Типичные ошибки и чего не делать
- Не путайте типобезопасность с истинностью code.
- Не добавляйте обязательное поле без default в существующий persist flow без миграции.
- Не изменяйте state из параллельных шагов без согласованной стратегии.
Не увеличивайте retries, max_iter или таймаут после каждой ошибки автоматически. Ограничение останавливает цикл, но не исправляет отсутствующий факт или плохую схему. Не выполняйте платежи, удаление, публикацию и другие необратимые действия только на основании текста модели. Добавляйте детерминированную проверку и, где требуется, человеческое подтверждение.
FAQ
Как объявить typed flow?
Указать Flow[ВашBaseModel].
Как читать поля?
Через атрибуты self.state.
Зачем defaults?
Они определяют корректное начальное состояние и упрощают восстановление.
Официальные источники
Частые вопросы
Как объявить typed flow?
Указать `Flow[ВашBaseModel]`.
Как читать поля?
Через атрибуты self.state.
Зачем defaults?
Они определяют корректное начальное состояние и упрощают восстановление.
Читайте также
Как хранить данные в словарном state CrewAI Flow
Словарное состояние CrewAI Flow: запись кода и счётчика, чтение в следующем шаге, точный итог и защита от отсутствующего ключа.
Как объединить события CrewAI Flow через or_ и and_
Проверяемая разница or_ и and_ в CrewAI Flow: запуск после любого события или после всех, пример состояния и критерии выбора.
Как передать результат одной задачи CrewAI в другую через context
Две последовательные задачи CrewAI с context: извлечение кода, формирование ответа, конкретный эталон и проверка отсутствия посторонних данных.
Комментарии
Пока тихо. Скажите первое слово