Предметно-ориентированное проектирование для ИИ-агентов: контексты и правила

Viacheslav Dubrov

2025-10-20 · Обновлено 2026-10-08 · 15 мин · ИИ-агенты

Автоматический перевод

Эта статья была автоматически переведена с оригинальной английской версии.

Это руководство для Python-инженеров, которые создают ИИ-агентов, меняющих бизнес-данные. Вы узнаете, как предметно-ориентированное проектирование (Domain-Driven Design, DDD) организует бизнес-словарь и правила и как превратить предложение модели в авторизованную операцию.

Проекты с агентами трудно менять, когда промпты, код и бизнес-процессы используют разные термины. Комплаенс просит «проверку политики», а в реализации есть process_data(). Расплывчатое имя скрывает, какое правило применяется, кто им владеет и куда вносить изменение.

DDD ставит в центр бизнес-язык и владение правилами. Языковая модель интерпретирует запрос и предлагает операцию с именованными типизированными полями. Код приложения добавляет доверенные данные о пользователе, проверяет права и вызывает бизнес-правила перед сохранением изменения.

Проблема в том, кто владеет правилом

В агентных системах одно правило часто распределено между системным промптом, описанием инструмента, API-обработчиком и ограничением в базе данных. Копии расходятся. Лимит возврата меняется, один промпт остаётся старым, и синтаксически корректный tool call попадает под неверную политику.

DDD начинает с других вопросов:

• Какая команда владеет правилом?
• Каким языком о нём говорят эксперты предметной области?
• В каких границах этот термин имеет одно значение?
• Какие изменения состояния должны оставаться согласованными вместе?

Эти вопросы полезны, когда процесс достаточно важен, чтобы иметь политики и жизненный цикл. Простому чат-боту только для чтения паттерны кода из этой статьи могут не понадобиться. Применяйте их, когда агент меняет состояние, у которого есть правила; для функций только для чтения или из одного шага они не нужны.

Задавайте эти четыре вопроса по порядку. Каждое «да» добавляет один паттерн, который разбирается в следующих разделах.

1. Агент меняет бизнес-данные? Если нет, используйте доменные термины в инструментах и промптах и остановитесь.
2. Правила решают, допустимо ли изменение? Добавьте агрегат и сервис приложения.
3. Один термин значит разное для разных команд? Добавьте ограниченные контексты и карту контекстов.
4. Другой сервис должен реагировать на изменение? Добавьте доменное событие и transactional outbox.

Стратегический дизайн до кода

Постройте единый язык

Ограниченный контекст (bounded context) — часть системы, где бизнес-термины и правила имеют одно согласованное значение. Его единый язык (ubiquitous language) — словарь, общий для экспертов предметной области и разработчиков. Если служба поддержки говорит RefundRequest, approval limit и settlement, эти термины должны встречаться в требованиях, коде, контрактах инструментов и эвалах.

Это больше, чем выбор описательных имён методов. Терминам нужны определения и примеры. «Одобрено» означает, что менеджер нажал кнопку, что платёжный процессор принял перевод или и то и другое? Неоднозначность, найденная в глоссарии, обходится дешевле, чем неоднозначность, найденная в трейсе агента.

Проведите границы контекстов по моделям и владению

Доменная модель — это объекты, данные и правила, которыми описывается область бизнеса; её не нужно путать с языковой моделью. Одно и то же существительное может означать разное в разных контекстах. «Продукт» может быть складской единицей в Inventory, строкой с ценой в Billing и обязательством по доставке в Order Management.

Схема сравнивает один общий класс Product с отдельной моделью Product в каждом контексте. В нижней половине каждая стрелка — отображение, которое выполняет код, когда строка заказа переходит в другой контекст.

Каждый контекст владеет своей доменной моделью. Когда информация переходит в другой контекст, код отображает термины и идентификаторы между ними. Фаулер описывает это разделение как способ не давать одной модели охватывать все значения термина (bounded contexts). Контекст не обязан быть микросервисом, репозиторием, агентом или командой, хотя эти границы часто совпадают.

Для проектирования агентов это означает:

• один контекст может использовать внутри несколько вызовов модели или специализированных агентов
• агенту, который охватывает несколько контекстов, нужны явный перевод терминов и полномочия для каждого из них
• оркестрация относится к приложению; она не отменяет владение доменом

Следующая схема показывает эти три случая. Сравните вторую и третью колонки: во второй один агент переводит термины для двух контекстов; в третьей оркестратор только отправляет команды, а каждый контекст применяет свои правила.

Прежде чем рисовать граф агентов, начните с карты контекстов (context map): она показывает области бизнеса и то, как они обмениваются информацией. Иначе граф склонен повторять набор доступных инструментов, а не бизнес.

Классифицируйте поддомены

DDD обычно различает:

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

Для ассистента задач управление задачами может быть основным доменом, планирование — вспомогательным, а доставка уведомлений — типовым. Карта помещает каждый контекст в полосу его поддомена. Scheduling передаёт даты в Task Management, а тот сообщает в Notifications о каждой добавленной задаче через событие TaskAdded.

Вкладывайте усилия по моделированию в основной домен и пишите его правила сами. Для типовых поддоменов берите готовый сервис, например почтового провайдера для уведомлений. Классификация не решает, где место LLM.

Тактические паттерны задают границу состояния

Сущности и объекты-значения

У сущности (entity) есть идентичность и жизненный цикл. Задача остаётся той же задачей после изменения описания. Объект-значение (value object) определяется своими значениями и обычно неизменяем: адрес электронной почты, денежная сумма или временное окно.

Состояние агента делится так же. ConversationSession с историей сообщений — сущность, как и каждая Task, которую планирует агент. Записанный результат инструмента, например ToolOutput с аргументами вызова и ответом, — объект-значение: после сохранения он не меняется. Когда Task становится объектом, он может отказаться завершаться, пока не завершены его зависимости, и это правило находится в одном методе, а не в нескольких шаблонах промптов.

Агрегаты и инварианты

Агрегат (aggregate) объединяет связанные объекты, изменения которых должны подчиняться одним правилам (Evans, Domain-Driven Design Reference). Остальной код меняет эти объекты через один входной объект, который называется корнем агрегата (aggregate root).

В примере ниже TaskList — корень, а его объекты Task — члены. Инвариант — правило, которое должно выполняться после каждой операции. Этот список отклоняет задачу с тем же нормализованным описанием и сроком, что и у существующей. В других доменах инварианты другие:

• завершённую задачу нельзя завершить повторно
• у владельца не может быть дублирующихся открытых напоминаний на один день
• возврат не может превышать оставшуюся возвращаемую сумму
• в плане не больше 10 открытых задач, даже если модель предлагает 50

Агрегат не становится безопасным только потому, что Python-список спрятан за методом add_task(). Внешний код не должен получать изменяемую ссылку, которая обходит этот метод. Хранение тоже требует контроля конкурентного доступа, иначе два корректных запроса, сохранённые одновременно, могут нарушить инвариант.

Репозитории и сервисы приложения

Репозиторий загружает и сохраняет агрегаты, не пуская детали работы с базой данных в домен (Fowler, Repository). Сервис приложения координирует один сценарий использования: загружает состояние, вызывает доменную операцию, сохраняет с ожидаемой версией и публикует получившиеся события.

Unit of work группирует сохранения одного сценария в одну транзакцию базы данных: она применяет все сохранения или ни одного (Percival and Gregory, Unit of Work). Сервис приложения открывает его, вносит изменения и вызывает commit(). Выход без вызова commit() отбрасывает незафиксированные изменения.

Домен не должен вызывать LLM, HTTP-клиент или ORM. Всё это относится к адаптерам — коду, который соединяет приложение с внешними системами.

Доменные события — это факты, а не шина сообщений

TaskAdded фиксирует факт: задача добавлена. Приложение может сохранить это событие в таблице outbox в той же транзакции базы данных, что и изменение задачи. Транзакция сохраняет оба изменения вместе или ни одного. Отдельный издатель отправляет сохранённое событие другим сервисам только после успеха транзакции. Это паттерн transactional outbox; одного объекта события для него недостаточно (Richardson, Transactional Outbox). Если отправлять в брокер прямо из сущности, можно опубликовать событие для транзакции, которая потом завершится ошибкой.

Схема сравнивает три прогона: сущность отправляет событие брокеру, после чего транзакция откатывается; outbox с rollback; outbox с commit.

События могут координировать агентов, но сами по себе не делают координацию надёжной. Семантика доставки, идемпотентность, порядок и версионируемые контракты остаются работой инфраструктуры.

Получателя событий для издателя выбирайте по тому, кто потребляет событие. Если слушатели работают в том же процессе, их может вызвать после коммита библиотека внутрипроцессных сигналов, например blinker. Если событие потребляют другие сервисы, отправляйте его в брокер, например RabbitMQ, или в Redis Streams, если команда уже запускает Redis. Не используйте для этих событий Redis Pub/Sub: он доставляет каждое сообщение не более одного раза, и отключённый подписчик его теряет (Redis Pub/Sub delivery semantics).

Считайте вывод модели недоверенным предложением

Интеграция с LLM похожа на слой защиты от повреждения (anti-corruption layer): она переводит внешнее вероятностное представление в термины, понятные домену. Слой перевода проверяет только форму и значения предложения. Права проверяет сервис приложения, а бизнес-правила остаются в агрегате.

Порт (port) — интерфейс, который перечисляет операции, нужные приложению. Порт репозитория перечисляет, как загружать и сохранять списки задач; адаптер базы данных реализует эти операции. Схема рисует четыре пакета как кольца и показывает зависимости кода, а не порядок выполнения. Приложение использует агрегат и порт репозитория; адаптер инфраструктуры реализует этот порт. Домен не зависит ни от адаптера базы данных, ни от модели.

В Python-проекте четыре слоя могут быть четырьмя пакетами: domain/, application/, infrastructure/ и interface/. Тогда замена базы данных затрагивает infrastructure/, а смена провайдера модели — interface/.

Следующая схема прогоняет четыре вызова из разобранного ниже примера через пять шагов списка под ней. Каждая строка показывает, где вызов останавливается и что возвращает инструмент.

Код между моделью и доменом выполняет такие шаги:

1. Получить предложение: модель возвращает AddTaskProposal; считайте каждое значение в нём недоверенным.
2. Разобрать и нормализовать: проверьте его по типизированному контракту, затем определите даты, единицы, идентификаторы и локаль по доверенному контексту.
3. Авторизовать: решите, может ли этот актор запрашивать операцию.
4. Обеспечить инвариант: вызовите метод агрегата, который проверяет правило и создаёт TaskAdded.
5. Сохранить вместе: запишите список задач и событие outbox в одной транзакции.

Три инструмента помогают получить от модели типизированный вывод и проверить его (шаги 1 и 2):

• Pydantic AI, если нужен фреймворк для агентов. Задайте тип вывода агента как модель Pydantic; если вывод не проходит валидацию, фреймворк возвращает ошибку модели и просит повторить.
• Instructor, если вы остаётесь на SDK своего провайдера и вам нужны только разобранные объекты Pydantic. Он оборачивает клиент, а max_retries повторно спрашивает модель, передавая ей ошибку валидации.
• Собственный режим structured output у провайдера, например OpenAI Structured Outputs, если вы хотите, чтобы провайдер ограничивал генерацию по JSON Schema.

Возвращайте ошибку валидации модели один-два раза. Если предложение всё равно не проходит, спросите пользователя, а не повторяйте дальше.

Pydantic может отклонить пропущенное поле или приоритет вне low, normal, high с помощью типизированной модели валидации (Pydantic documentation). Он не может решить, что «завтра» соответствует правильной дате, что пользователь владеет этим списком задач или что похожая задача уже открыта.

Рабочий пример: добавление задачи

Рассмотрим запрос добавить «Send the invoice» в список задач. Актор — вошедший в систему автор запроса (actor_id); владелец — человек, чей список изменится (owner_id). Это могут быть разные люди, поэтому приложение проверяет права для этого актора и этого списка до изменения.

Адаптер модели передаёт детали задачи как AddTaskProposal. Сервис приложения проверяет права, просит TaskList добавить задачу и фиксирует изменение вместе с событием TaskAdded через unit of work. Сохраните семь блоков Python от этого раздела до «Запустите пример» включительно, по порядку, в task_example.py. Они образуют один модуль для Python 3.12 с Pydantic 2; последний блок его запускает.

1. Опишите предложение, обращённое к модели

Держите предложение близким к тому, что модель может вывести сама. Не просите её выдумывать ID базы данных или доверенные идентификаторы владельца.

from datetime import date
from typing import Literal

from pydantic import BaseModel, Field, field_validator

class AddTaskProposal(BaseModel):
    description: str = Field(min_length=1, max_length=200)
    due_date: date | None = None
    priority: Literal["low", "normal", "high"] = "normal"

    @field_validator("description")
    @classmethod
    def description_must_contain_text(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("description must contain non-whitespace characters")
        return value

Если пользователь говорит «завтра», приложение должно передать модели явную локальную дату или разрешить относительное выражение протестированным парсером дат. Никогда не используйте часы сервера инференса как неявный бизнес-контекст.

2. Поместите инвариант в агрегат

TaskList.add_task() отклоняет пустые описания и дубликаты до изменения списка. Эти проверки здесь работают независимо от того, пришёл запрос от LLM или из обычной формы.

from dataclasses import dataclass, field
from datetime import date
from uuid import UUID, uuid4

class DomainError(Exception):
    """A business rule rejected the operation."""

class DuplicateTask(DomainError):
    pass

@dataclass(frozen=True)
class TaskAdded:
    task_id: UUID
    owner_id: UUID

@dataclass(frozen=True)
class Task:
    task_id: UUID
    description: str
    due_date: date | None
    priority: str

@dataclass
class TaskList:
    owner_id: UUID
    version: int
    _tasks: tuple[Task, ...] = ()
    _events: list[TaskAdded] = field(default_factory=list)

    @property
    def tasks(self) -> tuple[Task, ...]:
        return self._tasks

    def pull_events(self) -> tuple[TaskAdded, ...]:
        events = tuple(self._events)
        self._events.clear()
        return events

    def add_task(
        self,
        description: str,
        due_date: date | None,
        priority: str,
    ) -> Task:
        description = description.strip()
        if not description:
            raise DomainError("Task description must contain non-whitespace characters")

        normalized = " ".join(description.casefold().split())
        duplicate = any(
            " ".join(task.description.casefold().split()) == normalized
            and task.due_date == due_date
            for task in self._tasks
        )
        if duplicate:
            raise DuplicateTask("A matching task already exists for that date")

        task = Task(uuid4(), description, due_date, priority)
        self._tasks += (task,)
        self._events.append(TaskAdded(task.task_id, self.owner_id))
        return task

Домен поднимает собственные типы исключений. pydantic.ValidationError — подкласс ValueError, поэтому инструмент, который ловил бы ValueError, чтобы сообщить о доменном конфликте, поймал бы и некорректные предложения.

TaskAdded — замороженный dataclass: объект-значение, фиксирующий факт. Агрегат хранит задачи в неизменяемом кортеже, поэтому вызывающий код не получает изменяемую коллекцию, в которую можно добавлять или править элементы в обход add_task(). Его изменяющие методы заменяют этот кортеж только после проверки правила.

Обнаружение дубликатов здесь намеренно простое. Реальным правилам могут понадобиться нормализация с учётом локали, семантика повторений или ограничение уникальности в базе данных как последняя защита от гонок.

3. Определите порты

Эти протоколы Python перечисляют, что приложению нужно от хранилища и от проверки прав. TaskListRepository загружает и сохраняет один список задач. UnitOfWork держит этот репозиторий и ожидающие события и коммитит оба. TaskAuthorizer поднимает Unauthorized, когда актору нельзя менять список.

from typing import Protocol, Self
from uuid import UUID

class ConcurrentUpdate(Exception):
    pass

class Unauthorized(Exception):
    pass

class TaskListRepository(Protocol):
    def get(self, owner_id: UUID) -> TaskList: ...

    def save(self, task_list: TaskList, expected_version: int) -> None: ...

class UnitOfWork(Protocol):
    task_lists: TaskListRepository

    def __enter__(self) -> Self: ...

    def __exit__(self, *exc_info: object) -> None: ...

    def add_events(self, events: tuple[TaskAdded, ...]) -> None: ...

    def commit(self) -> None: ...

class TaskAuthorizer(Protocol):
    def require_add_permission(self, actor_id: UUID, owner_id: UUID) -> None: ...

Адаптер базы данных должен сохранять, только если сохранённая версия всё ещё совпадает с expected_version, то есть с версией, загруженной раньше. Успешное сохранение увеличивает эту версию. Если другой запрос уже изменил список, адаптер поднимает ConcurrentUpdate, а не перезаписывает работу того запроса. Это оптимистический контроль конкурентного доступа; сравнение, лежащее в его основе, показывает документация SQLAlchemy о счётчике версий. Протокол формулирует требование; конкретный адаптер должен выполнять проверку и обновление вместе.

4. Скоординируйте сценарий

AddTaskService выстраивает по порядку проверку прав, загрузку, доменную операцию и сохранение. Идентификаторы актора и владельца приходят из доверенного контекста приложения, а не из предложения модели.

from uuid import UUID

class AddTaskService:
    def __init__(self, uow: UnitOfWork, authorizer: TaskAuthorizer) -> None:
        self.uow = uow
        self.authorizer = authorizer

    def execute(
        self,
        actor_id: UUID,
        owner_id: UUID,
        proposal: AddTaskProposal,
    ) -> Task:
        self.authorizer.require_add_permission(actor_id, owner_id)

        with self.uow:
            task_list = self.uow.task_lists.get(owner_id)
            expected_version = task_list.version
            task = task_list.add_task(
                description=proposal.description,
                due_date=proposal.due_date,
                priority=proposal.priority,
            )
            self.uow.task_lists.save(task_list, expected_version)
            self.uow.add_events(task_list.pull_events())
            self.uow.commit()
        return task

Если строка до commit() вызывает исключение, unit of work отбрасывает незафиксированное изменение списка задач и событие. Ошибка соединения или таймаут во время commit() может оставить результат неизвестным: база данных могла зафиксировать транзакцию до потери ответа. Выход из блока не может отменить уже зафиксированную транзакцию (Amazon Builders’ Library, Making retries safe with idempotent APIs).

Модели в этом сервисе нет. Один адаптер может получать AddTaskProposal от LLM, другой — из HTTP-формы, а тесты могут создавать его напрямую. Бизнес-поведение остаётся одним и тем же.

5. Добавьте адаптеры в памяти

Эти адаптеры позволяют запустить пример без базы данных. InMemoryUnitOfWork.commit() сначала проверяет все версии и только потом записывает списки задач и события, поэтому сохраняет либо всё, либо ничего. AllowList — поддельный авторизатор для тестов и демо.

import copy
from uuid import UUID

class InMemoryDatabase:
    def __init__(self) -> None:
        self.task_lists: dict[UUID, TaskList] = {}
        self.outbox: list[TaskAdded] = []

class InMemoryTaskLists:
    def __init__(self, db: InMemoryDatabase) -> None:
        self.db = db
        self.staged: dict[UUID, tuple[TaskList, int]] = {}

    def get(self, owner_id: UUID) -> TaskList:
        stored = self.db.task_lists.get(owner_id, TaskList(owner_id, version=0))
        return copy.deepcopy(stored)

    def save(self, task_list: TaskList, expected_version: int) -> None:
        self.staged[task_list.owner_id] = (task_list, expected_version)

class InMemoryUnitOfWork:
    def __init__(self, db: InMemoryDatabase) -> None:
        self.db = db

    def __enter__(self) -> "InMemoryUnitOfWork":
        self.task_lists = InMemoryTaskLists(self.db)
        self.events: list[TaskAdded] = []
        return self

    def __exit__(self, *exc_info: object) -> None:
        # Leaving the block without commit() discards every staged change.
        self.task_lists.staged.clear()
        self.events.clear()

    def add_events(self, events: tuple[TaskAdded, ...]) -> None:
        self.events.extend(events)

    def commit(self) -> None:
        staged = self.task_lists.staged
        for owner_id, (_, expected) in staged.items():
            stored = self.db.task_lists.get(owner_id)
            if (stored.version if stored else 0) != expected:
                raise ConcurrentUpdate(f"task list {owner_id} changed")
        for owner_id, (task_list, expected) in staged.items():
            task_list.version = expected + 1
            self.db.task_lists[owner_id] = task_list
        self.db.outbox.extend(self.events)
        staged.clear()
        self.events.clear()

class AllowList:
    def __init__(self, allowed: set[tuple[UUID, UUID]]) -> None:
        self.allowed = allowed

    def require_add_permission(self, actor_id: UUID, owner_id: UUID) -> None:
        if (actor_id, owner_id) not in self.allowed:
            raise Unauthorized(f"{actor_id} may not add tasks for {owner_id}")

Адаптер базы данных реализует UnitOfWork через одну SQL-транзакцию; версия в памяти показывает контракт.

Сопоставьте инструменты с командами приложения

Инструменты агента должны открывать сценарии использования, а не примитивы базы данных. Предпочитайте:

add_task(description, due_date, priority)
complete_task(task_id)
reschedule_task(task_id, due_date)

вместо:

insert_row(table, values)
update_record(table, id, patch)

Первый набор говорит на языке домена и даёт приложению место, где можно проверить права и применить правила. Второй позволяет модели описывать произвольные изменения хранилища. Схема проводит один и тот же дублирующий запрос через оба подхода: инструмент базы данных сохраняет вторую копию, а инструмент доменной команды отказывает.

Код фреймворка агентов, например узел LangGraph или инструмент Pydantic AI, относится к слою интерфейса и вызывает сервисы приложения, как это делает add_task_tool() ниже. Если оркестрация в слое приложения использует фреймворк, поместите его за порт, которым владеет приложение. Домен никогда не импортирует фреймворк.

Результат инструмента должен различать сбои, на которые агент может отреагировать. Не сводите все сбои к строке, которая приглашает к слепым повторам. У каждого сбоя своё следующее действие:

Здесь unavailable сообщает об исключении инфраструктуры; оно не доказывает, что ничего не сохранилось. Повторяйте с нарастающей задержкой только после подтверждения приложением, что запись не произошла, либо если приложение атомарно сохраняет запись и её результат под ID запроса и возвращает этот результат при повторах. Используйте тот же ID и те же аргументы в течение срока их хранения (Amazon Builders’ Library, Making retries safe with idempotent APIs). В остальных случаях выясните результат по записям приложения. Если эти записи не позволяют его определить, сообщите пользователю, что результат неизвестен.

Демо отклоняет задачи с одинаковым содержимым, но не реализует идемпотентность запросов: повтор после зафиксированной записи возвращает domain_conflict, а не исходный успешный результат.

Эта функция инструмента оборачивает AddTaskService.execute() и возвращает эти коды. Фреймворк передаёт актора и владельца из сессии вошедшего пользователя, а не от модели.

from uuid import UUID

from pydantic import ValidationError

def add_task_tool(
    service: AddTaskService,
    actor_id: UUID,
    owner_id: UUID,
    arguments: dict,
) -> dict:
    try:
        proposal = AddTaskProposal.model_validate(arguments)
        task = service.execute(actor_id, owner_id, proposal)
    except ValidationError as exc:
        return {"error": "invalid_proposal", "detail": str(exc)}
    except Unauthorized:
        return {"error": "unauthorized"}
    except DomainError as exc:
        return {"error": "domain_conflict", "detail": str(exc)}
    except ConcurrentUpdate:
        return {"error": "concurrent_update"}
    except (ConnectionError, TimeoutError):
        return {"error": "unavailable"}
    return {"ok": True, "task_id": str(task.task_id)}

Запустите пример

Последний блок вызывает инструмент четыре раза: одна корректная задача, один дубликат с другим регистром и пробелами, один приоритет вне допустимых значений и один актор без прав.

from uuid import uuid4

if __name__ == "__main__":
    alice, bob = uuid4(), uuid4()
    db = InMemoryDatabase()
    service = AddTaskService(InMemoryUnitOfWork(db), AllowList({(alice, alice)}))

    def call(actor, arguments):
        result = add_task_tool(service, actor, alice, arguments)
        print(result.get("error", "ok"))

    invoice = {"description": "Send the invoice", "due_date": "2026-10-05"}
    call(alice, invoice)  # ok
    call(alice, {**invoice, "description": "send the  INVOICE"})  # domain_conflict
    call(alice, {**invoice, "priority": "urgent"})  # invalid_proposal
    call(bob, invoice)  # unauthorized

    stored = db.task_lists[alice]
    print(f"version={stored.version} tasks={len(stored.tasks)} outbox={len(db.outbox)}")

Если Python 3.12 установлен, выполните эти команды в каталоге с task_example.py:

python3.12 -m venv .venv
.venv/bin/python -m pip install "pydantic>=2,<3"
.venv/bin/python task_example.py

Модуль выводит:

ok
domain_conflict
invalid_proposal
unauthorized
version=1 tasks=1 outbox=1

Состояние изменил только первый вызов. Сохранённый список имеет версию 1 с одной задачей, а в outbox лежит одно событие TaskAdded.

Тесты для каждого слоя

Тесты домена

Тестируйте агрегаты с помощью pytest, без модели, сети и базы данных:

• дубликаты задач отклоняются
• корректные задачи создают ожидаемое событие
• открытые наружу коллекции не могут изменить внутреннее состояние
• правила переходов выполняются при повторных операциях

Второй вызов в запуске выше уже проверяет первый пункт через инструмент. Доменный тест делает тот же вызов прямо на TaskList и ожидает DuplicateTask.

Тесты приложения

Используйте поддельные репозитории и авторизаторы, чтобы проверить загрузку, порядок авторизации, сохранение с ожидаемой версией, поведение outbox и сопоставление ошибок. Unit of work в памяти выше и есть такая подделка. Сохраните следующий блок отдельно в test_task_example.py, рядом с task_example.py. Он открывает два unit of work на одних данных, чтобы смоделировать два запроса, загрузивших одну версию:

from uuid import uuid4

import pytest

from task_example import ConcurrentUpdate, InMemoryDatabase, InMemoryUnitOfWork

def test_stale_save_raises_concurrent_update():
    db, owner = InMemoryDatabase(), uuid4()
    first, second = InMemoryUnitOfWork(db), InMemoryUnitOfWork(db)

    with first, second:
        mine = first.task_lists.get(owner)  # both requests load version 0
        theirs = second.task_lists.get(owner)

        mine.add_task("Call the bank", None, "normal")
        first.task_lists.save(mine, expected_version=0)
        first.commit()  # stored version is now 1

        theirs.add_task("Book travel", None, "normal")
        second.task_lists.save(theirs, expected_version=0)
        with pytest.raises(ConcurrentUpdate):
            second.commit()

    assert [t.description for t in db.task_lists[owner].tasks] == ["Call the bank"]

Установите pytest в то же окружение и запустите тест:

.venv/bin/python -m pip install pytest
.venv/bin/python -m pytest -q test_task_example.py

Эвалы контракта модели

Оценивайте вероятностный адаптер отдельно:

• точность определения намерения и извлечения полей
• разрешение относительных дат при заданном контексте часового пояса
• отказ или уточняющий вопрос, когда не хватает нужной информации
• устойчивость к промпт-инъекции внутри цитируемого текста задачи
• доля предложений, которые валидны по схеме, но семантически непригодны

Для тестов, которые не должны вызывать настоящую модель, Pydantic AI предоставляет TestModel, который вызывает инструменты агента и возвращает вывод, валидный по схеме, и FunctionModel, где тест сам решает, каким будет каждый ответ (Pydantic AI testing). Описанные выше эвалы оставьте для настоящей модели.

Сквозной тест после этого должен подтвердить, что плохие предложения никогда не обходят те же доменные методы, которыми пользуются доверенные интерфейсы.

Порядок работы над новым проектом с агентами

1. Опросите экспертов предметной области и составьте глоссарий с определением и примером для каждого термина.
2. Нарисуйте карту контекстов. Начните с одного основного контекста.
3. Смоделируйте сущности и объекты-значения. Поместите каждое правило в метод, который меняет состояние.
4. Выберите корни агрегатов и запишите их инварианты.
5. Определите порты репозитория, unit of work и авторизатора. Используйте адаптеры в памяти, пока не пройдут доменные тесты.
6. Создавайте доменные события для изменений, которые нужны другим контекстам, и записывайте их в outbox в той же транзакции.
7. Определите типы предложений, обращённых к модели, и парсер, который их проверяет.
8. Откройте каждый сценарий как инструмент, который вызывает сервис приложения и сопоставляет его сбои.
9. Оркестрацию добавляйте последней, отправляя команды между контекстами.

Когда дизайн работает

Вы должны уметь менять провайдера модели, не меняя доменный тест. Изменение политики должно затрагивать один агрегат или доменный сервис, а не несколько промптов. Трейсы должны использовать термины, которые узнаёт команда-владелец. Некорректные или неавторизованные предложения должны падать до сохранения, а параллельное сохранение должно завершаться ошибкой, а не молча перезаписывать состояние.

Модель всё равно будет делать неверные предложения; агрегат отклоняет те, что нарушают правило.

Источники

• Eric Evans, Domain-Driven Design Reference — определения стратегических и тактических паттернов
• Martin Fowler, Bounded Context — почему одна модель не должна охватывать все значения термина
• Martin Fowler, Repository — абстракция хранилища в виде коллекции
• Harry Percival and Bob Gregory, Architecture Patterns with Python: Unit of Work — паттерн unit of work в Python
• Chris Richardson, Transactional Outbox — публикация после транзакции базы данных без потери событий
• Pydantic documentation — валидация типизированных предложений
• Pydantic AI output — structured output с повторами при ошибке валидации
• Instructor retrying — валидированный вывод Pydantic поверх SDK провайдера