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

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

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

Кратко

Предметно-ориентированное проектирование (DDD) дает командам AI-агентов общий словарь и четкие границы между подсистемами. Используйте его, когда ваши prompts уже оторвались от бизнеса, а правила разбросаны по шаблонам.

---

Почему предметно-ориентированное проектирование важно для AI-агентов

Большинство агентных проектов проваливаются не потому, что код плохой. Они проваливаются потому, что люди, пишущие prompts, и люди, которые реально владеют бизнес-процессом, не могут договориться о значении терминов. Compliance просит выполнить "policy check", а в ответ получает метод process_data(). Никто не понимает, что он делает, поэтому требования расползаются, а система костенеет.

DDD исправляет это, ставя бизнес-домен в центр. Не схему базы данных. Не шаблон prompt. А реальный процесс из предметной области, который вы пытаетесь смоделировать. Практические эффекты:

• Общий язык. Product, ops и engineering используют одни и те же термины. Когда compliance говорит "refund request", именно это и появляется в вашем коде, prompts и документации.
• Сфокусированный охват. Вы строите то, что действительно важно: ключевые workflow и правила, за которые кто-то реально отвечает. Меньше glue code, который ломается при смене требований.
• Адаптивность. Когда политики меняются, вы обновляете один четко определенный срез вместо того, чтобы искать нужное место в монолите.

Это особенно важно в доменах, где правила часто меняются: финансы, здравоохранение, регулируемые операции. DDD дает хотя бы шанс за этим успевать.

---

Стратегические строительные блоки

DDD — это набор паттернов, а не одна идея. Обычно его делят на две части:

1. Strategic Design: "большая картина". Определение границ, команд и способов взаимодействия систем. Это критично для мультиагентных систем.
2. Tactical Design: паттерны уровня кода (Entities, Aggregates), которые поддерживают чистоту внутренней логики агента.

Концепции ниже — это то, чем вы реально будете пользоваться каждый день.

Ubiquitous language

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

Если compliance говорит "policy check", ваш метод называется run_policy_check(), а не process_data(). Если врачи говорят "admit patient", вы пишете admit_patient(), а не add_user().

class PatientRegistry:
    def admit_patient(self, patient_id: str) -> None:
        """Admit a patient to the registry - term used by medical staff."""
        ...

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

Ограниченные контексты

Большим системам нужны явные границы. Почему? Потому что одно и то же слово означает разное в разных частях бизнеса.

Возьмем "product" в e-commerce. В контексте Inventory product — это элемент каталога с SKU и остатками на складе. В контексте Billing — это позиция счета с правилами ценообразования и расчетом налогов. В Order Management — это количество и обещание по доставке.

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

Это сохраняет каждую модель небольшой и не дает появиться одному гигантскому объекту "product", который должен одновременно удовлетворять требованиям трех команд.

Сущности и value objects

Это базовые строительные блоки вашей доменной модели.

Entities имеют идентичность, которая сохраняется во времени. Task с ID 123 — это одна и та же задача, даже если вы меняете ее описание, статус или дедлайн. Две сущности равны, если у них одинаковый ID, независимо от атрибутов.

from pydantic import BaseModel

class SupportTicket(BaseModel):
    ticket_id: str  # This is the identity
    customer: str
    issue: str
    status: str = "OPEN"

    def close(self) -> None:
        if self.status != "OPEN":
            raise ValueError("Ticket already closed")
        self.status = "CLOSED"

Value objects не имеют идентичности. Они полностью определяются своими атрибутами. Два объекта TimeSlot с одинаковым временем начала и окончания взаимозаменяемы. Value objects неизменяемы; вместо изменения существующего объекта вы создаете новый.

from pydantic import BaseModel

class TimeSlot(BaseModel):
    start: str  # e.g., "2025-10-18 09:00"
    end: str    # e.g., "2025-10-18 10:00"

    @property
    def duration(self) -> int:
        # Compute duration from start to end
        ...

Используйте entities для вещей с жизненным циклом (Order, User, AgentSession). Используйте value objects для описаний и измерений (EmailAddress, Priority, Location).

Aggregates

Aggregates — это кластеры связанных entities и value objects, которые рассматриваются как единое целое. Внутри aggregate бизнес-правила всегда должны оставаться истинными. В этом и есть весь смысл.

У каждого aggregate есть один aggregate root — сущность, которая контролирует доступ ко всему внутри. Хотите что-то изменить в aggregate? Делайте это через root. Root обеспечивает инварианты, чтобы aggregate не оказался в неконсистентном состоянии.

from datetime import date
from pydantic import BaseModel, Field

class Task(BaseModel):
    id: str
    description: str
    completed: bool = False

class Plan(BaseModel):  # This is the aggregate root
    id: str
    tasks: list[Task] = Field(default_factory=list)

    def add_task(self, task: Task) -> None:
        # Business rule enforced here: no duplicate task IDs
        if any(t.id == task.id for t in self.tasks):
            raise ValueError("Task ID already exists")
        self.tasks.append(task)

Внешний код никогда не работает с списком tasks напрямую. Он всегда вызывает add_task(). Именно это гарантирует, что правило "никаких дублирующихся ID" не будет нарушено. При сохранении в базу данных вы обычно сохраняете весь aggregate целиком.

Репозитории

Репозитории скрывают слой хранения данных. С точки зрения домена вы вызываете save(plan) и get(plan_id). То, что эти вызовы в итоге идут в Postgres или Redis, — чужая зона ответственности.

Это дает два выигрыша. В тестах можно использовать in-memory репозиторий вместо mock-ов вызовов базы. А когда вы в итоге замените SQLite на что-то тяжелее, бизнес-правила останутся на месте.

from abc import ABC, abstractmethod
from pydantic import BaseModel, Field

class Task(BaseModel):
    id: str
    description: str
    completed: bool = False

class Plan(BaseModel):
    id: str
    tasks: list[Task] = Field(default_factory=list)

class PlanRepository(ABC):
    """Domain layer defines the interface."""
    @abstractmethod
    def save(self, plan: Plan) -> None:
        ...

    @abstractmethod
    def get(self, plan_id: str) -> Plan | None:
        ...

class InMemoryPlanRepository(PlanRepository):
    """Infrastructure layer provides the implementation."""
    def __init__(self) -> None:
        self.storage: dict[str, Plan] = {}

    def save(self, plan: Plan) -> None:
        self.storage[plan.id] = plan

    def get(self, plan_id: str) -> Plan | None:
        return self.storage.get(plan_id)

Ваш доменный код знает только про PlanRepository (интерфейс). Infrastructure layer подставляет реальную реализацию.

Доменные события

Доменные события фиксируют важные факты, произошедшие в системе. Имена даются в прошедшем времени (OrderPlaced, TaskCompleted, PaymentFailed), потому что они описывают факты, а не команды.

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

from datetime import datetime
from pydantic import BaseModel

class TaskCompleted(BaseModel):
    task_id: str
    completed_at: datetime

Когда задача завершена, вы эмитите TaskCompleted. Сервис уведомлений может слушать это событие и отправлять email. Сервис отчетности может записывать его для аналитики. Важный момент: aggregate задачи не должен знать ни про email, ни про аналитику. Он просто объявляет, что произошло.

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

---

Перенос DDD в архитектуры агентов

У реальных агентных систем есть многошаговые workflow, ошибки в ответах LLM с некоторой вероятностью и требования, которые меняются каждый квартал. Паттерны DDD неожиданно хорошо ложатся на эти проблемы.

Ограниченные контексты становятся агентами или навыками

Каждый агент (или крупная capability) — это ограниченный контекст. Исследовательский orchestrator может координировать трех специализированных агентов:

• Trends Agent — собирает рыночные данные, используя собственный словарь и инструменты
• Compliance Agent — выполняет policy checks в регуляторной терминологии
• Cost Agent — оценивает затраты по финансовым правилам

У каждого — собственная модель, терминология и инварианты. Они общаются через четко определенные интерфейсы или события.

Даже в системе с одним агентом вы можете определить внутренние контексты. Например, модуль Planning и модуль Execution, каждый со своей доменной моделью.

Prompts соблюдают ubiquitous language

Используйте доменные термины в system prompts, описаниях tools и сигнатурах функций. Если эксперты по compliance говорят "policy check", именно эта фраза должна быть в prompts и коде. Польза очень приземленная: когда в трассировке агента видно run_policy_check, команда compliance может прочитать это без переводчика.

Состояние становится явными entities

LLM часто stateless, но реальные агенты отслеживают много состояния: сессии диалога, цели, промежуточные результаты, результаты вызовов tools. Моделируйте это как entities или value objects:

• entity ConversationSession с ID и историей сообщений
• entity Task, представляющая единицы работы
• value object ToolOutput для неизменяемых результатов

Когда эти объекты становятся явными, к ним можно привязывать валидацию и бизнес-правила. Сущность Task может отказываться переходить в completed, пока не завершены зависимости, и это правило не будет жить в трех разных шаблонах prompt.

Aggregates выражают планы агента

Aggregate root Plan управляет списком задач и обеспечивает те ограничения, которые важны для бизнеса. Когда LLM предлагает добавить 50 задач, а ваша политика разрешает 10, aggregate отклоняет лишние. Когда она предлагает дублирующуюся работу, aggregate тоже ее отклоняет. Модель может быть чрезмерно инициативной; домен остается в здравом состоянии.

Доменные события управляют оркестрацией

Агенты поднимают события вроде ResearchCompleted, ThresholdExceeded или PolicyViolationDetected. Другие агенты или сервисы подписываются и реагируют. Ничего не зашито жестко, поэтому добавить нового слушателя (или нового агента) дешево.

Бизнес-правила оборачивают действия AI

Выходы LLM проходят через доменные сервисы или методы сущностей, а не сразу в базу данных. Если модель предлагает возврат за пределами policy limits, ваш RefundRequest валидирует и отклоняет это. LLM может импровизировать; последнее слово остается за бизнес-правилами.

Слой Anti-Corruption Layer (ACL)

LLM вероятностна и иногда ошибается неожиданными способами. Ваша доменная модель должна оставаться детерминированной. Напрямую встречаться они не должны.

Для этого и нужен Anti-Corruption Layer (ACL).

ACL находится между моделью и доменом. Он переводит сырой вывод LLM в строгие типы, которые ожидает ваш домен.

1. Ingest сырой текст или JSON от LLM.
2. Validate структуру и типы с помощью моделей Pydantic.
3. Sanitize значения (никаких отрицательных цен, транзакций из будущего и т. д.).
4. Translate DTO (Data Transfer Objects) в доменные сущности.

Если валидация не проходит, ACL отклоняет данные и часто отправляет ошибку обратно в LLM, чтобы та попробовала еще раз. Смысл прост: к вашей основной бизнес-логике попадают только валидные данные.

---

Пример: task assistant, смоделированный с помощью DDD

Построим персонального task assistant, который обрабатывает запросы вроде "Remind me to buy milk tomorrow" или "What's on my to-do list?". Ниже по шагам применяются элементы DDD, описанные выше.

1. Определите контексты

Начните с разбиения задачи на поддомены:

• Task Management — обработка to-do items и напоминаний (core domain)
• Scheduling — события календаря и встречи
• Notifications — отправка алертов и email

Сначала сосредоточимся на Task Management. Остальные части могут развиваться как отдельные ограниченные контексты или companion agents.

2. Говорите на одном языке

Выберите словарь вместе с людьми, которые реально владеют процессом (или просто используйте здравый смысл для персонального приложения): "task", "deadline", "reminder", "priority". Затем используйте именно эти термины в шаблонах prompt, именах методов и UI labels. Никакого отдельного "бизнес-перевода" нет.

3. Зафиксируйте entities, value objects и события

Теперь смоделируем основные понятия:

• Entity: Task с идентичностью (id) и изменяемым состоянием (completed)
• Value object: enum Priority (неизменяемый, определяется своим значением)
• Domain event: TaskCompletedEvent, сигнализирующий о завершении работы

from datetime import datetime, date, timezone
from enum import Enum
from pydantic import BaseModel, Field

class Priority(Enum):
    """Value object: priority is defined by its value alone."""
    LOW = 1
    NORMAL = 2
    HIGH = 3

class TaskCompletedEvent(BaseModel):
    """Domain event: announces a task was completed."""
    task_id: str
    time: datetime

class Task(BaseModel):
    """Entity: identity persists even as attributes change."""
    id: str
    description: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    due_date: date | None = None
    priority: Priority = Priority.NORMAL
    completed: bool = False

    def mark_completed(self) -> TaskCompletedEvent:
        """Business rule: can't complete an already-completed task."""
        if self.completed:
            raise ValueError("Task is already completed.")
        self.completed = True
        return TaskCompletedEvent(task_id=self.id, time=datetime.now(timezone.utc))

Бизнес-правило (нельзя завершить уже завершенную задачу) находится в методе сущности, а не в шаблоне prompt.

4. Сформируйте aggregate

TaskList — это наш aggregate root. Он содержит несколько сущностей Task и обеспечивает правила консистентности между ними. Все изменения проходят через методы root.

from datetime import date
from pydantic import BaseModel, Field

class Task(BaseModel):
    id: str
    description: str
    due_date: date | None = None
    completed: bool = False

class TaskList(BaseModel):
    """Aggregate root: enforces invariants across all tasks."""
    owner: str
    tasks: list[Task] = Field(default_factory=list)

    def add_task(self, task: Task) -> None:
        """Business rule: no duplicate tasks on the same day."""
        if any(
            existing.description == task.description
            and existing.due_date == task.due_date
            for existing in self.tasks
        ):
            raise ValueError("A similar task on that date already exists.")
        self.tasks.append(task)

    def get_pending(self) -> list[Task]:
        """Query helper: find tasks that aren't done yet."""
        return [task for task in self.tasks if not task.completed]


TaskList.model_rebuild()  # Resolve forward references for Pydantic.

Внешний код никогда не трогает tasks напрямую. Он всегда идет через add_task() или другой метод root, и именно это удерживает правило "без дубликатов" в силе.

5. Оберните хранение в репозиторий

Репозиторий абстрагирует storage. Domain layer не знает, живут ли задачи в памяти или в Postgres.

from pydantic import BaseModel, Field

class Task(BaseModel):
    id: str
    description: str
    completed: bool = False

class TaskList(BaseModel):
    owner: str
    tasks: list[Task] = Field(default_factory=list)

TaskList.model_rebuild()  # Resolve forward references for Pydantic.

class TaskRepository:
    """Abstracts task storage - in-memory implementation for simplicity."""

    def __init__(self) -> None:
        self._data: dict[str, TaskList] = {}

    def get_task_list(self, owner: str) -> TaskList:
        """Retrieve a user's task list, or create a new empty one."""
        return self._data.get(owner, TaskList(owner=owner))

    def save_task_list(self, task_list: TaskList) -> None:
        """Persist changes to the task list."""
        self._data[task_list.owner] = task_list

В production вы замените это на реализацию поверх базы данных (с использованием SQLAlchemy или напрямую Postgres), не трогая доменный код.

6. Выполните flow

Когда пользователь делает запрос, поток выглядит так:

from datetime import date, timedelta
from uuid import uuid4
from pydantic import BaseModel, Field

class Task(BaseModel):
    id: str
    description: str
    due_date: date | None = None
    completed: bool = False

class TaskList(BaseModel):
    owner: str
    tasks: list[Task] = Field(default_factory=list)

    def add_task(self, task: Task) -> None:
        if any(
            existing.description == task.description
            and existing.due_date == task.due_date
            for existing in self.tasks
        ):
            raise ValueError("A similar task on that date already exists.")
        self.tasks.append(task)

class TaskRepository:
    def __init__(self) -> None:
        self._data: dict[str, TaskList] = {}

    def get_task_list(self, owner: str) -> TaskList:
        return self._data.get(owner, TaskList(owner=owner))

    def save_task_list(self, task_list: TaskList) -> None:
        self._data[task_list.owner] = task_list


TaskList.model_rebuild()  # Resolve forward references for Pydantic.


# User says: "Remind me to buy milk tomorrow"
# (In reality, an LLM would parse this into structured data)
user_input = "Remind me to buy milk tomorrow"
intent = "add_task"

# Initialize repository
repo = TaskRepository()

if intent == "add_task":
    # 1. Load the user's task list
    task_list = repo.get_task_list(owner="User123")

    # 2. Create a new task entity
    task = Task(
        id=str(uuid4()),
        description="buy milk",
        due_date=date.today() + timedelta(days=1),
    )

    # 3. Domain layer enforces business rules
    try:
        task_list.add_task(task)
        repo.save_task_list(task_list)
        print(f"Task '{task.description}' added for {task.due_date}.")
    except Exception as exc:
        print(f"Sorry, I couldn't add that task: {exc}")

Слои остаются разделенными:

• LLM layer разбирает естественный язык в структурированные данные (intent + parameters)
• Domain layer обеспечивает бизнес-правила через методы сущностей
• Repository layer отвечает за persistence, не протекая в доменную логику

LLM может быть креативной при парсинге, но домен решает, что консистентно. Если она попытается добавить дублирующуюся задачу, aggregate root ее отклонит. Вам не нужен специальный пункт в prompt для этого случая.

---

Инструменты, чтобы оживить модель

DDD не требует специальных фреймворков. Но несколько инструментов делают реализацию проще, особенно для AI-агентов.

FastAPI

FastAPI хорошо ложится на слои DDD. Используйте routers для разделения ограниченных контекстов (/tasks, /schedule), модели Pydantic для валидации запросов и ответов, а dependency injection — для подключения репозиториев.

Структурируйте проект по слоям:

project/
├── domain/          # Pure business logic (entities, aggregates, value objects)
├── application/     # Use cases and command handlers
├── infrastructure/  # Repositories, databases, external APIs
└── interface/       # FastAPI routers and HTTP contracts

Такая слоистость (иногда ее называют "onion architecture") не дает изменениям распространяться по всей кодовой базе. Замена базы данных означает изменения только в infrastructure/ и больше нигде. Изменение UI означает изменения только в interface/ и больше нигде.

Pydantic и Pydantic AI

Pydantic обеспечивает инварианты и валидирует данные во время выполнения. Используйте его для entities, value objects и особенно для валидации выходов LLM.

Pydantic AI идет дальше: он гарантирует, что ответы LLM соответствуют схемам вашего домена. Определите AddTaskCommand с обязательными полями, и Pydantic AI провалидирует JSON-ответ модели до того, как ваш код к нему прикоснется.

Instructor — еще один вариант. Он патчит клиенты OpenAI (и других провайдеров), чтобы те сразу возвращали модели Pydantic, что дает легковесный способ реализовать Anti-Corruption Layer.

Вспомогательные библиотеки для DDD

• DDDesign — предоставляет базовые классы для entities, репозиториев и value objects на основе Pydantic
• Protean — полноценный фреймворк для DDD, CQRS и event sourcing, если вам нужно решение с большим количеством готовых возможностей из коробки

Большинство Python-разработчиков обходятся обычными классами и Pydantic, но для крупных проектов на эти библиотеки стоит посмотреть.

Инструменты для event-driven архитектуры

Для доменных событий рассмотрите:

• blinker — легковесный in-process диспетчер событий
• redis-py Pub/Sub или RabbitMQ — для распределенных событий между сервисами или агентами
• asyncio event patterns — если у вас уже async-стек

События критичны для мультиагентной оркестрации. Один агент эмитит ResearchCompleted; остальные подписываются и реагируют. Ни одному агенту не нужно знать, кто слушает.

Фреймворки для агентов

LangChain, LangGraph, Haystack, Semantic Kernel, LlamaIndex, AutoGen, Google ADK, smolagents и CrewAI — все они дают структуру для agent workflows. Используйте их на уровне application или infrastructure и оборачивайте в интерфейсы, которыми владеет domain layer. Тогда замена фреймворка станет локальным изменением.

Тестирование

Один из практических плюсов DDD: domain layer можно тестировать без запуска всего стека.

• PyTest для unit-тестов сущностей и aggregates
• Fake repositories (in-memory) для интеграционных тестов
• LLM stubs, которые возвращают заранее заданные результаты

Ваш доменный код никогда не должен требовать живую LLM для запуска тестов. LLM — это деталь реализации. Тесты проверяют бизнес-правила.

---

Чеклист для старта

Практический порядок действий при запуске нового агентного проекта:

1. Поговорите с domain experts. Сформируйте ubiquitous language. Зафиксируйте его письменно.
2. Определите bounded contexts. Нарисуйте поддомены и отметьте, где им нужно взаимодействовать. Начните с одного core context.
3. Смоделируйте entities и value objects. У каких объектов есть идентичность? Какие являются просто значениями? Встраивайте инварианты в их методы.
4. Определите aggregate roots. Сгруппируйте связанные сущности под одним root, который обеспечивает правила консистентности.
5. Создайте интерфейсы репозиториев. Пока не реализуйте storage. Просто определите save() и get(). Домен не должен знать, где живут данные.
6. Эмитите доменные события. Для значимых изменений (order placed, task completed) поднимайте события. Подключите listeners позже, когда понадобится.
7. Оберните выходы LLM в схемы. Используйте модели Pydantic для обеспечения контрактов. Свободный текст не должен протекать в домен.
8. Добавьте оркестрацию. Постройте application services, которые координируют агентов через структурированные команды или события.

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