Projektowanie sterowane dziedziną dla agentów AI: konteksty i reguły

Viacheslav Dubrov

2025-10-20 · Zaktualizowano 2026-10-08 · 15 min · Agenci AI

Tłumaczenie automatyczne

Ten artykuł został automatycznie przetłumaczony z angielskiego oryginału.

Ten przewodnik jest przeznaczony dla inżynierów Pythona, którzy budują agentów AI zmieniających dane biznesowe. Dowiesz się, jak Domain-Driven Design (DDD) porządkuje słownictwo i reguły biznesowe oraz jak zamienić propozycję modelu w autoryzowaną operację.

Projekty agentów trudno zmieniać, gdy teksty typu prompt, kod i procesy biznesowe używają różnych terminów. Dział zgodności pyta o „kontrolę polityki”, a implementacja udostępnia process_data(). Nieprecyzyjna nazwa ukrywa, jaka reguła jest stosowana, kto za nią odpowiada i gdzie wprowadzić zmianę.

DDD stawia w centrum ten język biznesowy i odpowiedzialność za reguły. Model językowy interpretuje żądanie i proponuje operację z nazwanymi, typowanymi polami. Kod aplikacji dostarcza zaufane informacje o użytkowniku, sprawdza uprawnienia i wywołuje reguły biznesowe, zanim zapisze zmianę.

Problemem jest odpowiedzialność za reguły

Systemy agentów często rozpraszają jedną regułę między prompt systemowy, opis narzędzia, handler API i ograniczenie w bazie danych. Kopie się rozjeżdżają. Limit zwrotu się zmienia, jeden prompt pozostaje stary, a poprawne składniowo wywołanie narzędzia trafia na niewłaściwą politykę.

DDD zaczyna od innych pytań:

• Który zespół odpowiada za regułę?
• Jakiego języka używają dla niej eksperci domenowi?
• W jakich granicach ten termin ma jedno znaczenie?
• Które zmiany stanu muszą pozostać spójne razem?

Te pytania są przydatne, gdy przepływ pracy jest na tyle ważny, że ma polityki i cykl życia. Prosty chatbot tylko do odczytu może nie potrzebować wzorców kodu opisanych dalej w tym artykule. Stosuj te wzorce, gdy agent zmienia stan podlegający regułom; pomiń je w funkcjach tylko do odczytu lub jednoetapowych.

Zadaj te cztery pytania po kolei. Każde „tak” dodaje jeden wzorzec, który wyjaśniają kolejne sekcje.

1. Czy agent zmienia dane biznesowe? Jeśli nie, używaj terminów domenowych w narzędziach i tekstach typu prompt i na tym poprzestań.
2. Czy reguły decydują, czy zmiana jest dozwolona? Dodaj agregat i serwis aplikacyjny.
3. Czy jeden termin znaczy co innego dla różnych zespołów? Dodaj konteksty ograniczone i mapę kontekstów.
4. Czy inny serwis musi zareagować na zmianę? Dodaj zdarzenie domenowe i transakcyjny outbox.

Projektowanie strategiczne przed kodem

Zbuduj wszechobecny język

Kontekst ograniczony (bounded context) to część systemu, w której terminy i reguły biznesowe mają jedno uzgodnione znaczenie. Jego wszechobecny język (ubiquitous language) to słownictwo wspólne dla ekspertów domenowych i programistów. Jeśli obsługa klienta mówi RefundRequest, approval limit i settlement, te terminy powinny pojawiać się w wymaganiach, kodzie, kontraktach narzędzi i ewaluacjach.

To coś więcej niż wybór opisowych nazw metod. Terminy potrzebują definicji i przykładów. Czy „zatwierdzone” oznacza, że kierownik kliknął przycisk, że operator płatności przyjął przelew, czy jedno i drugie? Niejednoznaczność odkryta w glosariuszu jest tańsza niż ta odkryta w śladzie agenta.

Wyznacz konteksty ograniczone wokół modeli i odpowiedzialności

Model domeny to obiekty, dane i reguły służące do przedstawienia obszaru biznesowego; nie jest to model językowy. Ten sam rzeczownik może znaczyć różne rzeczy w różnych kontekstach. „Produkt” może być jednostką magazynową w Inventory, wycenioną pozycją w Billing i zobowiązaniem dostawy w Order Management.

Rysunek porównuje jedną wspólną klasę Product z jednym modelem Product na kontekst. W dolnej połowie każda strzałka to mapowanie, które wykonuje kod, gdy pozycja zamówienia trafia do innego kontekstu.

Każdy kontekst jest właścicielem swojego modelu domeny. Gdy informacje trafiają do innego kontekstu, kod mapuje terminy i identyfikatory między nimi. Fowler opisuje to rozdzielenie jako sposób, by jeden model nie obejmował wszystkich znaczeń danego terminu (bounded contexts). Kontekst nie jest automatycznie mikroserwisem, repozytorium, agentem ani zespołem, choć te granice często się pokrywają.

W projektowaniu agentów oznacza to, że:

• jeden kontekst może wewnętrznie używać kilku wywołań modelu lub wyspecjalizowanych agentów
• jeden agent obejmujący kilka kontekstów potrzebuje jawnego tłumaczenia i uprawnień dla każdego z nich
• orkiestracja jest sprawą aplikacji; nie znosi odpowiedzialności domeny za dane reguły

Następny rysunek przedstawia te trzy przypadki. Porównaj drugą i trzecią kolumnę: w drugim jeden agent tłumaczy terminy dla dwóch kontekstów; w trzecim orkiestrator tylko wysyła komendy, a każdy kontekst stosuje własne reguły.

Zanim narysujesz graf agentów, zacznij od mapy kontekstów, która pokazuje obszary biznesowe i sposób wymiany informacji między nimi. W przeciwnym razie graf zwykle odtwarza dostępność narzędzi, a nie biznes.

Sklasyfikuj poddomeny

DDD zwykle rozróżnia:

• Domena podstawowa (core domain): zdolność, która tworzy wyróżniającą wartość
• Poddomena wspierająca: konieczna praca specyficzna dla biznesu, która nie jest wyróżnikiem
• Poddomena generyczna: rozwiązana już zdolność, taka jak tożsamość czy dostarczanie e-maili

W asystencie zadań zarządzanie zadaniami może być domeną podstawową, planowanie wspierającą, a dostarczanie powiadomień generyczną. Mapa umieszcza każdy kontekst w pasie jego poddomeny. Scheduling dostarcza daty do Task Management, który zgłasza każde dodane zadanie do Notifications przez zdarzenie TaskAdded.

Wysiłek modelowania poświęć domenie podstawowej i jej reguły napisz samodzielnie. Dla poddomen generycznych użyj istniejącej usługi, na przykład dostawcy e-maili do powiadomień. Klasyfikacja nie rozstrzyga, gdzie ma trafić LLM.

Wzorce taktyczne wyznaczają granicę stanu

Encje i obiekty wartości

Encja ma tożsamość i cykl życia. Zadanie pozostaje tym samym zadaniem po zmianie opisu. Obiekt wartości jest określony przez swoje wartości i zwykle jest niezmienny: adres e-mail, kwota pieniężna lub przedział czasu.

Stan agenta pasuje do tego samego podziału. ConversationSession z historią wiadomości jest encją, podobnie jak każde Task, które agent planuje. Zapisany wynik narzędzia, na przykład ToolOutput z argumentami wywołania i odpowiedzią, jest obiektem wartości: po zapisaniu się nie zmienia. Gdy Task jest obiektem, może odmówić zakończenia, dopóki jego zależności nie zostaną ukończone, a ta reguła znajduje się w jednej metodzie zamiast w kilku szablonach tekstów typu prompt.

Agregaty i niezmienniki

Agregat grupuje powiązane obiekty, których zmiany muszą podlegać tym samym regułom (Evans, Domain-Driven Design Reference). Inny kod zmienia te obiekty przez jeden obiekt wejściowy, zwany korzeniem agregatu (aggregate root).

W poniższym przykładzie korzeniem jest TaskList, a jego członkami są obiekty Task. Niezmiennik to reguła, która musi pozostać prawdziwa po każdej operacji. Ta lista odrzuca zadanie z tym samym znormalizowanym opisem i terminem co istniejące zadanie. Inne domeny mają inne niezmienniki:

• ukończonego zadania nie można ukończyć ponownie
• właściciel nie może mieć zduplikowanych otwartych przypomnień na ten sam dzień
• zwrot nie może przekroczyć pozostałej kwoty do zwrotu
• plan przyjmuje co najwyżej 10 otwartych zadań, nawet gdy model proponuje 50

Agregat nie staje się bezpieczny tylko dlatego, że lista Pythona znajduje się za metodą add_task(). Kod zewnętrzny nie może otrzymać modyfikowalnej referencji, która omija tę metodę. Trwałe przechowywanie wymaga też kontroli współbieżności, bo inaczej dwa poprawne żądania mogą naruszyć niezmiennik, gdy zapisują się jednocześnie.

Repozytoria i usługi aplikacyjne

Repozytorium ładuje i zapisuje agregaty bez przenoszenia spraw bazy danych do domeny (Fowler, Repository). Usługa aplikacyjna koordynuje jeden przypadek użycia: ładuje stan, wywołuje operację domenową, zapisuje z oczekiwaną wersją i publikuje wynikowe zdarzenia.

Jednostka pracy (unit of work) grupuje zapisy jednego przypadku użycia w jedną transakcję bazy danych: zatwierdza wszystkie albo żaden (Percival i Gregory, Unit of Work). Usługa aplikacyjna otwiera ją, wprowadza zmiany i wywołuje commit(). Wyjście bez wywołania commit() odrzuca niezatwierdzone zmiany.

Domena nie powinna wywoływać LLM, klienta HTTP ani ORM. Te elementy należą do adapterów, czyli kodu łączącego aplikację z systemami zewnętrznymi.

Zdarzenia domenowe są faktami, a nie szyną komunikatów

TaskAdded zapisuje fakt: zadanie zostało dodane. Aplikacja może zapisać to zdarzenie w tabeli outbox w tej samej transakcji bazy danych co zmiana zadania. Transakcja zapisuje obie zmiany razem albo żadnej. Osobny publikator wysyła zapisane zdarzenie do innych usług dopiero po powodzeniu transakcji. To wzorzec transactional outbox; sam obiekt zdarzenia go nie zapewnia (Richardson, Transactional Outbox). Wysyłanie prosto do brokera z encji grozi opublikowaniem zdarzenia dla transakcji, która później się nie powiedzie.

Rysunek porównuje trzy przebiegi: encja wysyła zdarzenie do brokera, a transakcja potem się wycofuje; outbox z wycofaniem; outbox z zatwierdzeniem.

Zdarzenia mogą koordynować agentów, ale same nie zapewniają niezawodnej koordynacji. Semantyka dostarczania, idempotentność, kolejność i wersjonowane kontrakty pozostają pracą infrastrukturalną.

Cel publikatora wybierz według tego, kto konsumuje zdarzenie. Gdy odbiorcy działają w tym samym procesie, biblioteka sygnałów w procesie, taka jak blinker, może wywołać ich po zatwierdzeniu. Gdy zdarzenie konsumują inne usługi, wyślij je do brokera, na przykład RabbitMQ, lub do Redis Streams, jeśli zespół już uruchamia Redis. Nie używaj Redis Pub/Sub dla tych zdarzeń: dostarcza każdą wiadomość co najwyżej raz, a rozłączony subskrybent ją traci (semantyka dostarczania Redis Pub/Sub).

Traktuj dane wyjściowe modelu jako niezaufaną propozycję

Integracja z LLM przypomina warstwę antykorupcyjną (anti-corruption layer): tłumaczy zewnętrzną, probabilistyczną reprezentację na terminy, które rozumie domena. Warstwa tłumacząca sprawdza tylko kształt i wartości propozycji. Uprawnienia pozostają w usłudze aplikacyjnej, a reguły biznesowe w agregacie.

Port to interfejs wymieniający operacje, których potrzebuje aplikacja. Port repozytorium wymienia, jak ładować i zapisywać listy zadań; adapter bazy danych implementuje te operacje. Rysunek przedstawia cztery pakiety jako pierścienie i pokazuje zależności kodu, a nie kolejność wykonania. Aplikacja używa agregatu i portu repozytorium; adapter infrastruktury implementuje ten port. Domena nie zależy ani od adaptera bazy danych, ani od modelu.

W projekcie Pythona te cztery warstwy mogą być czterema pakietami: domain/, application/, infrastructure/ i interface/. Wymiana bazy danych dotyka wtedy infrastructure/; zmiana dostawcy modelu dotyka interface/.

Następny diagram przeprowadza cztery wywołania z przykładu poniżej przez pięć kroków z listy pod nim. Każdy wiersz pokazuje, gdzie wywołanie się zatrzymuje i co zwraca narzędzie.

Kod między modelem a domeną wykonuje te kroki:

1. Odbierz propozycję: model zwraca AddTaskProposal; traktuj każdą wartość w niej jako niezaufaną.
2. Sparsuj i znormalizuj: zwaliduj ją względem typowanego kontraktu, a potem rozstrzygnij daty, jednostki, identyfikatory i ustawienia regionalne na podstawie zaufanego kontekstu.
3. Autoryzuj: zdecyduj, czy ten aktor może zażądać operacji.
4. Wymuś niezmiennik: wywołaj metodę agregatu, która sprawdza regułę i zgłasza TaskAdded.
5. Zatwierdź razem: zapisz listę zadań i zdarzenie outbox w jednej transakcji.

Trzy narzędzia mogą sprawić, że model zwróci typowane dane wyjściowe, i je zwalidować (kroki 1 i 2):

• Pydantic AI, gdy chcesz użyć frameworka agentów. Ustaw typ wyjścia agenta na model Pydantic; gdy dane wyjściowe nie przejdą walidacji, framework odsyła błąd do modelu i prosi o ponowną odpowiedź.
• Instructor, gdy zostajesz przy SDK dostawcy i potrzebujesz tylko sparsowanych obiektów Pydantic. Opakowuje klienta, a max_retries ponawia zapytanie do modelu wraz z błędem walidacji.
• Własny tryb ustrukturyzowanych danych wyjściowych dostawcy, na przykład OpenAI Structured Outputs, gdy chcesz, by dostawca ograniczał generowanie do JSON Schema.

Odsyłaj błąd walidacji do modelu raz lub dwa razy. Jeśli propozycja nadal nie przechodzi, zapytaj użytkownika zamiast ponawiać.

Pydantic może odrzucić brakujące pole lub priorytet spoza low, normal, high dzięki swojemu typowanemu modelowi walidacji (dokumentacja Pydantic). Nie rozstrzygnie, że „jutro” oznacza właściwą datę, że użytkownik jest właścicielem listy zadań ani że podobne zadanie jest już otwarte.

Przykład krok po kroku: dodanie zadania

Rozważ żądanie dodania „Send the invoice” do listy zadań. Aktor to zalogowany zgłaszający (actor_id); właściciel to osoba, której lista ulegnie zmianie (owner_id). Mogą to być różne osoby, więc aplikacja sprawdza uprawnienia tego aktora do tej listy, zanim wprowadzi zmianę.

Adapter modelu dostarcza szczegóły zadania jako AddTaskProposal. Usługa aplikacyjna sprawdza uprawnienia, prosi TaskList o dodanie zadania i zatwierdza zmianę wraz ze zdarzeniem TaskAdded przez jednostkę pracy. Zapisz siedem bloków Pythona od tej sekcji do „Uruchom przykład” włącznie, po kolei, jako task_example.py. Tworzą jeden moduł dla Pythona 3.12 z Pydantic 2; ostatni blok go uruchamia.

1. Zdefiniuj propozycję przeznaczoną dla modelu

Trzymaj propozycję blisko tego, co model może wywnioskować. Nie proś go o wymyślanie identyfikatorów bazy danych ani zaufanych identyfikatorów właściciela.

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

Jeśli użytkownik mówi „jutro”, aplikacja powinna przekazać modelowi jawną lokalną datę albo rozstrzygnąć wyrażenie względne przetestowanym parserem dat. Nigdy nie używaj zegara serwera inferencji jako niejawnego kontekstu biznesowego.

2. Umieść niezmiennik w agregacie

TaskList.add_task() odrzuca puste opisy i zduplikowane zadania, zanim zmieni listę. Dzięki temu, że te kontrole są tutaj, obowiązują niezależnie od tego, czy żądanie pochodzi od LLM, czy ze zwykłego formularza.

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

Domena zgłasza własne typy wyjątków. pydantic.ValidationError jest podklasą ValueError, więc narzędzie, które przechwytywałoby ValueError, by zgłosić konflikt domenowy, przechwytywałoby też niepoprawne propozycje.

TaskAdded jest zamrożoną klasą danych (frozen dataclass): obiektem wartości, który zapisuje fakt. Agregat trzyma zadania w niemodyfikowalnej krotce, więc wywołujący nie dostają modyfikowalnej kolekcji, do której mogliby dopisywać lub którą mogliby edytować z pominięciem add_task(). Jego metody zmieniające podmieniają tę krotkę dopiero po wymuszeniu reguły.

Wykrywanie duplikatów jest tu celowo proste. Prawdziwe reguły mogą wymagać normalizacji uwzględniającej ustawienia regionalne, semantyki powtarzalności albo ograniczenia unikalności w bazie danych jako ostatecznego zabezpieczenia odpornego na wyścigi.

3. Zdefiniuj porty

Te protokoły Pythona wymieniają, czego aplikacja potrzebuje od magazynu danych i od kontroli uprawnień. TaskListRepository ładuje i zapisuje jedną listę zadań. UnitOfWork przechowuje to repozytorium oraz oczekujące zdarzenia i zatwierdza oba. TaskAuthorizer zgłasza Unauthorized, gdy aktor nie może zmienić listy.

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: ...

Adapter bazy danych musi zapisać dane tylko wtedy, gdy zapisana wersja nadal zgadza się z expected_version, czyli wersją załadowaną wcześniej. Udany zapis zwiększa tę wersję. Jeśli inne żądanie zmieniło listę, adapter zgłasza ConcurrentUpdate zamiast nadpisać pracę tamtego żądania. To optymistyczna kontrola współbieżności (optimistic concurrency control); dokumentacja licznika wersji w SQLAlchemy pokazuje leżące u jej podstaw porównanie. Protokół określa wymaganie; konkretny adapter musi zaimplementować sprawdzenie i aktualizację razem.

4. Skoordynuj przypadek użycia

AddTaskService ustawia w kolejności uprawnienia, ładowanie, operację domenową i zapis. Identyfikatory aktora i właściciela pochodzą z zaufanego kontekstu aplikacji, a nie z propozycji modelu.

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

Jeśli linia przed commit() zgłosi wyjątek, jednostka pracy odrzuci niezatwierdzoną zmianę listy zadań i zdarzenie. Błąd połączenia lub przekroczenie limitu czasu podczas commit() może pozostawić wynik nieznany: baza danych mogła zatwierdzić transakcję, zanim jej odpowiedź została utracona. Wyjście z bloku nie może cofnąć tej zatwierdzonej transakcji (Amazon Builders’ Library, Making retries safe with idempotent APIs).

Modelu nie ma w tej usłudze. Jeden adapter może pobierać AddTaskProposal z LLM, inny z formularza HTTP, a testy mogą go tworzyć bezpośrednio. Zachowanie biznesowe pozostaje identyczne.

5. Dodaj adaptery w pamięci

Te adaptery pozwalają uruchomić przykład bez bazy danych. InMemoryUnitOfWork.commit() najpierw sprawdza każdą wersję, a dopiero potem zapisuje listy zadań i zdarzenia, więc zapisuje oba albo żaden. AllowList to fałszywy mechanizm autoryzacji do testów i demonstracji.

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}")

Adapter bazy danych implementuje UnitOfWork za pomocą jednej transakcji SQL; wersja w pamięci pokazuje kontrakt.

Mapuj narzędzia na komendy aplikacyjne

Narzędzia agenta powinny udostępniać przypadki użycia, a nie prymitywy bazy danych. Preferuj:

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

zamiast:

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

Pierwszy zestaw mówi językiem domeny i daje aplikacji miejsce na autoryzację i wymuszanie reguł. Drugi pozwala modelowi opisywać dowolne mutacje warstwy trwałej. Rysunek przepuszcza to samo żądanie duplikatu przez oba projekty: narzędzie bazy danych zapisuje drugą kopię, a narzędzie komendy domenowej ją odrzuca.

Kod frameworka agentów, taki jak węzeł LangGraph lub narzędzie Pydantic AI, należy do warstwy interfejsu i wywołuje usługi aplikacyjne, jak robi to poniższe add_task_tool(). Jeśli orkiestracja w warstwie aplikacji używa frameworka, umieść go za portem, którego właścicielem jest aplikacja. Domena nigdy nie importuje frameworka.

Wynik narzędzia powinien rozróżniać błędy, na które agent może zareagować. Nie spłaszczaj wszystkich błędów do napisu, który zachęca do ślepych ponowień. Każdy błąd wymaga innego następnego kroku:

Tutaj unavailable informuje o wyjątku infrastruktury; nie dowodzi, że nic nie zostało zapisane. Ponawiaj z rosnącymi opóźnieniami dopiero wtedy, gdy aplikacja potwierdzi brak zapisu, albo gdy atomowo zapisuje operację i jej wynik pod identyfikatorem żądania oraz zwraca ten wynik przy ponowieniach. Używaj tego samego identyfikatora i argumentów w okresie ich przechowywania (Amazon Builders’ Library, Making retries safe with idempotent APIs). W innym przypadku ustal wynik na podstawie zapisów aplikacji. Jeśli nie pozwalają go ustalić, poinformuj użytkownika, że wynik jest nieznany.

Demonstracja odrzuca zadania o zduplikowanej treści, ale nie implementuje idempotentności żądań: ponowienie po zatwierdzonym zapisie zwraca domain_conflict zamiast pierwotnego sukcesu.

Ta funkcja narzędzia opakowuje AddTaskService.execute() i zwraca te kody. Framework przekazuje aktora i właściciela z sesji zalogowanego użytkownika, a nie od modelu.

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)}

Uruchom przykład

Ostatni blok wywołuje narzędzie cztery razy: jedno poprawne zadanie, jeden duplikat z inną wielkością liter i odstępami, jeden priorytet spoza dozwolonych wartości oraz jednego aktora bez uprawnień.

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)}")

Po zainstalowaniu Pythona 3.12 uruchom te polecenia w katalogu zawierającym task_example.py:

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

Moduł wypisuje:

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

Stan zmieniło tylko pierwsze wywołanie. Zapisana lista ma wersję 1 z jednym zadaniem, a outbox zawiera jedno zdarzenie TaskAdded.

Testy dla każdej warstwy

Testy domeny

Testuj agregaty za pomocą pytest, bez modelu, sieci ani bazy danych:

• zduplikowane zadania są odrzucane
• poprawne zadania zgłaszają oczekiwane zdarzenie
• udostępnione kolekcje nie mogą zmienić stanu wewnętrznego
• reguły przejść obowiązują przy powtarzanych operacjach

Drugie wywołanie w powyższym uruchomieniu już sprawdza pierwszy punkt przez narzędzie. Test domeny wykonuje to samo wywołanie bezpośrednio na TaskList i oczekuje DuplicateTask.

Testy aplikacji

Użyj fałszywych repozytoriów i mechanizmów autoryzacji, by zweryfikować ładowanie, kolejność autoryzacji, zapisy z oczekiwaną wersją, zachowanie outbox i mapowanie błędów. Jednostka pracy w pamięci powyżej jest takim fałszywym obiektem. Zapisz poniższy blok osobno jako test_task_example.py, obok task_example.py. Otwiera dwie jednostki pracy na tych samych danych, by zasymulować dwa żądania, które ładują tę samą wersję:

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"]

Zainstaluj pytest w tym samym środowisku i uruchom test:

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

Ewaluacje kontraktu modelu

Oceniaj adapter probabilistyczny oddzielnie:

• dokładność rozpoznawania intencji i wyodrębniania pól
• rozstrzyganie dat względnych przy podanym kontekście strefy czasowej
• odmowa lub prośba o doprecyzowanie, gdy brakuje wymaganych informacji
• odporność na prompt injection w cytowanym tekście zadania
• odsetek propozycji poprawnych względem schematu, ale bezużytecznych semantycznie

Dla testów, które nie powinny wywoływać prawdziwego modelu, Pydantic AI udostępnia TestModel, który wywołuje narzędzia agenta i zwraca dane wyjściowe zgodne ze schematem, oraz FunctionModel, który pozwala testowi zdecydować o każdej odpowiedzi (testowanie w Pydantic AI). Powyższe ewaluacje zachowaj dla prawdziwego modelu.

Test end-to-end powinien następnie potwierdzić, że złe propozycje nigdy nie omijają tych samych metod domenowych, których używają zaufane interfejsy.

Kolejność prac w nowym projekcie agenta

1. Przeprowadź wywiady z ekspertami domenowymi i napisz glosariusz, z definicją i przykładem dla każdego terminu.
2. Narysuj mapę kontekstów. Zacznij od jednego kontekstu podstawowego.
3. Zamodeluj encje i obiekty wartości. Umieść każdą regułę w metodzie, która zmienia stan.
4. Wybierz korzenie agregatów i zapisz ich niezmienniki.
5. Zdefiniuj porty repozytorium, jednostki pracy i mechanizmu autoryzacji. Używaj adapterów w pamięci, dopóki testy domeny nie przejdą.
6. Zgłaszaj zdarzenia domenowe dla zmian potrzebnych innym kontekstom i zapisuj je do outbox w tej samej transakcji.
7. Zdefiniuj typy propozycji przeznaczone dla modelu oraz parser, który je waliduje.
8. Udostępnij każdy przypadek użycia jako narzędzie, które wywołuje usługę aplikacyjną i mapuje jej błędy.
9. Orkiestrację dodaj na końcu, wysyłając komendy między kontekstami.

Kiedy projekt działa prawidłowo

Powinieneś móc zmienić dostawcę modelu bez zmiany testu domeny. Zmiana polityki powinna dotykać jednego agregatu lub usługi domenowej zamiast kilku tekstów typu prompt. Ślady powinny używać terminów, które rozpoznaje zespół będący ich właścicielem. Błędne lub nieautoryzowane propozycje powinny zawodzić przed zapisem, a współbieżny zapis powinien zawieść zamiast po cichu nadpisać stan.

Model nadal będzie zgłaszał błędne propozycje; agregat odrzuca te, które łamią regułę.

Materiały referencyjne

• Eric Evans, Domain-Driven Design Reference — definicje wzorców strategicznych i taktycznych
• Martin Fowler, Bounded Context — dlaczego jeden model nie powinien obejmować wszystkich znaczeń terminu
• Martin Fowler, Repository — abstrakcja kolekcji w warstwie trwałej
• Harry Percival i Bob Gregory, Architecture Patterns with Python: Unit of Work — wzorzec jednostki pracy w Pythonie
• Chris Richardson, Transactional Outbox — publikowanie po transakcji bazy danych bez utraty zdarzeń
• Dokumentacja Pydantic — walidacja typowanych propozycji
• Pydantic AI output — ustrukturyzowane dane wyjściowe z ponowieniami po walidacji
• Instructor retrying — zwalidowane dane wyjściowe Pydantic nad SDK dostawcy