Domain-Driven Design voor AI agents: contexten en regels

Viacheslav Dubrov

2025-10-20 · Bijgewerkt 2026-10-08 · 17 min · AI-agenten

Automatische vertaling

Dit artikel is automatisch vertaald vanuit de oorspronkelijke Engelse versie.

Deze gids is bedoeld voor Python-engineers die AI agents bouwen die bedrijfsdata wijzigen. Je leert hoe Domain-Driven Design (DDD) het vocabulaire en de regels van een bedrijf ordent, en hoe je een voorstel van een model omzet in een geautoriseerde operatie.

Agentprojecten worden moeilijk te wijzigen wanneer prompts, code en bedrijfsprocessen verschillende termen gebruiken. Compliance vraagt om een “policy check”, terwijl de implementatie process_data() aanbiedt. Die vage naam verbergt welke regel wordt toegepast, wie er eigenaar van is en waar een wijziging thuishoort.

DDD zet die bedrijfstaal en het eigenaarschap centraal. Het language model interpreteert een verzoek en stelt een operatie voor met benoemde, getypeerde velden. De applicatiecode levert vertrouwde gebruikersinformatie, controleert de toestemming en roept de bedrijfsregels aan voordat een wijziging wordt opgeslagen.

Eigenaarschap van regels is het probleem

Agentsystemen verspreiden één regel vaak over een system prompt, een tool-beschrijving, een API-handler en een databaseconstraint. De kopieën lopen uiteen. Een limiet voor terugbetalingen verandert, één prompt blijft oud en een syntactisch geldige tool call komt bij het verkeerde beleid terecht.

DDD begint met andere vragen:

• Welk team is eigenaar van de regel?
• Welke taal gebruiken domeinexperts ervoor?
• Binnen welke grens heeft die term één betekenis?
• Welke statuswijzigingen moeten samen consistent blijven?

Die vragen zijn nuttig wanneer een workflow belangrijk genoeg is om beleid en een levenscyclus te hebben. Een eenvoudige chatbot die alleen leest heeft de codepatronen uit dit artikel misschien niet nodig. Gebruik de codepatronen wanneer een agent state wijzigt waarvoor regels gelden; sla ze over bij functies die alleen lezen of uit één stap bestaan.

Stel deze vier vragen in volgorde. Elk ja voegt één patroon toe dat in latere secties wordt uitgelegd.

1. Wijzigt de agent bedrijfsdata? Zo niet, gebruik dan domeintermen in tools en prompts en stop.
2. Bepalen regels of een wijziging is toegestaan? Voeg een aggregate en een application service toe.
3. Betekent één term iets anders voor verschillende teams? Voeg bounded contexts en een context map toe.
4. Moet een andere service op de wijziging reageren? Voeg een domain event en een transactional outbox toe.

Strategisch ontwerp vóór de code

Bouw een ubiquitous language

Een bounded context is een deel van het systeem waarin bedrijfstermen en -regels één afgesproken betekenis hebben. De ubiquitous language is het vocabulaire dat domeinexperts en developers delen. Als support operations RefundRequest, approval limit en settlement zeggen, moeten die termen terugkomen in requirements, code, tool contracts en evaluaties.

Dit is meer dan beschrijvende methodenamen kiezen. Termen hebben definities en voorbeelden nodig. Betekent “approved” dat een manager op een knop heeft geklikt, dat de betalingsverwerker de overboeking heeft geaccepteerd, of allebei? Onduidelijkheid die je in een glossary ontdekt, is goedkoper dan onduidelijkheid die je in een agent trace ontdekt.

Trek bounded contexts rond models en eigenaarschap

Een domain model bestaat uit de objecten, data en regels waarmee een bedrijfsgebied wordt weergegeven; het is iets anders dan het language model. Hetzelfde zelfstandig naamwoord kan in verschillende contexten iets anders betekenen. “Product” kan in Inventory een stock-keeping unit zijn, in Billing een geprijsde orderregel en in Order Management een leverbelofte.

De figuur vergelijkt één gedeelde Product-klasse met één Product-model per context. In de onderste helft is elke pijl een mapping die code uitvoert wanneer een orderregel naar een andere context gaat.

Elke context is eigenaar van zijn eigen domain model. Wanneer informatie naar een andere context gaat, mapt code de termen en identifiers tussen beide. Fowler beschrijft deze scheiding als een manier om te voorkomen dat één model alle betekenissen van een term moet dekken (bounded contexts). Het is niet automatisch een microservice, repository, agent of team, hoewel die grenzen vaak samenvallen.

Voor het ontwerp van agents betekent dit:

• één context kan intern meerdere modelaanroepen of gespecialiseerde agents gebruiken
• één agent die meerdere contexten overspant, heeft voor elke context expliciete vertaling en bevoegdheid nodig
• orchestration is een zaak van de applicatie; het wist het eigenaarschap in het domein niet uit

De volgende figuur tekent deze drie gevallen. Vergelijk de tweede en derde kolom: in het tweede vertaalt één agent termen voor twee contexten; in het derde stuurt een orchestrator alleen commando’s, en past elke context zijn eigen regels toe.

Begin met een context map die de bedrijfsgebieden toont en hoe ze informatie uitwisselen, voordat je een agentgraaf tekent. Anders volgt de graaf eerder de beschikbare tools dan het bedrijf.

Classificeer de subdomeinen

DDD maakt gewoonlijk onderscheid tussen:

• Core domain: de capability die onderscheidende waarde creëert
• Supporting subdomain: noodzakelijk, bedrijfsspecifiek werk dat niet onderscheidend is
• Generic subdomain: een opgeloste capability, zoals identiteit of het bezorgen van e-mail

Voor een taakassistent kan taakbeheer het core domain zijn, planning supporting en het bezorgen van notificaties generic. De kaart plaatst elke context in de band van zijn subdomein. Scheduling levert data aan Task Management, dat elke toegevoegde taak via een TaskAdded-event meldt aan Notifications.

Besteed je modelleerinspanning aan het core domain en schrijf de regels daarvan zelf. Gebruik voor generic subdomains een bestaande service, zoals een e-mailprovider voor notificaties. De classificatie bepaalt niet waar een LLM thuishoort.

Tactische patronen bepalen de grens van state

Entities en value objects

Een entity heeft een identiteit en een levenscyclus. Een taak blijft dezelfde taak nadat de omschrijving is gewijzigd. Een value object wordt bepaald door zijn waarden en is meestal immutable: een e-mailadres, een geldbedrag of een tijdvenster.

De state van een agent past in dezelfde indeling. Een ConversationSession met de berichtgeschiedenis is een entity, en dat geldt ook voor elke Task die de agent plant. Een vastgelegd tool result, zoals een ToolOutput met de aanroepargumenten en het antwoord, is een value object: het verandert niet meer nadat het is opgeslagen. Zodra een Task een object is, kan die weigeren om af te ronden totdat de afhankelijkheden klaar zijn, en die regel staat dan in één methode in plaats van in meerdere prompttemplates.

Aggregates en invarianten

Een aggregate groepeert samenhangende objecten waarvan de wijzigingen aan dezelfde regels moeten voldoen (Evans, Domain-Driven Design Reference). Andere code wijzigt die objecten via één toegangsobject, de aggregate root.

In het voorbeeld hieronder is TaskList de root en zijn de Task-objecten de leden. Een invariant is een regel die na elke operatie moet blijven kloppen. Deze lijst weigert een taak met dezelfde genormaliseerde omschrijving en vervaldatum als een bestaande taak. Andere domeinen hebben andere invarianten:

• een afgeronde taak kan niet nog een keer worden afgerond
• een eigenaar kan geen dubbele open herinneringen voor dezelfde dag hebben
• een terugbetaling kan niet hoger zijn dan het nog terug te betalen bedrag
• een plan accepteert hooguit 10 open taken, ook als het model er 50 voorstelt

Een aggregate wordt niet veilig alleen doordat een Python-lijst achter een add_task()-methode zit. Externe code mag geen muteerbare referentie krijgen waarmee die methode wordt omzeild. Voor persistence is ook concurrency control nodig, anders kunnen twee geldige verzoeken een invariant schenden wanneer ze tegelijk worden opgeslagen.

Repositories en application services

Een repository laadt en bewaart aggregates zonder dat databasezaken in het domein lekken (Fowler, Repository). Een application service coördineert één use case: state laden, de domeinoperatie aanroepen, opslaan met een verwachte versie en de resulterende events publiceren.

Een unit of work bundelt de opslagacties van één use case in één databasetransactie: alles wordt gecommit of niets (Percival en Gregory, Unit of Work). De application service opent hem, maakt zijn wijzigingen en roept commit() aan. Het verlaten zonder commit() aan te roepen verwijdert de nog niet gecommitte wijzigingen.

Het domein mag geen LLM, HTTP-client of ORM aanroepen. Die horen thuis in adapters, de code die de applicatie met externe systemen verbindt.

Domain events zijn feiten, geen message bus

TaskAdded legt een feit vast: er is een taak toegevoegd. De applicatie kan dit event opslaan in een outbox-tabel, in dezelfde databasetransactie als de taakwijziging. De transactie slaat beide wijzigingen samen op of geen van beide. Een aparte publisher stuurt het opgeslagen event pas naar andere services nadat de transactie is geslaagd. Dit is het transactional-outbox-patroon; een eventobject alleen biedt dat niet (Richardson, Transactional Outbox). Rechtstreeks vanuit een entity naar een broker sturen riskeert een event te publiceren voor een transactie die daarna mislukt.

De figuur vergelijkt drie runs: de entity stuurt het event naar de broker en de transactie wordt daarna teruggedraaid; de outbox met een rollback; en de outbox met een commit.

Events kunnen agents coördineren, maar maken de coördinatie niet vanzelf betrouwbaar. Leveringsgaranties, idempotentie, volgorde en versioned contracts blijven infrastructuurwerk.

Kies het doel van de publisher op basis van wie het event consumeert. Als de listeners in hetzelfde proces draaien, kan een in-process signaalbibliotheek zoals blinker ze na de commit aanroepen. Als andere services het event consumeren, stuur het dan naar een broker zoals RabbitMQ, of naar Redis Streams als het team al Redis draait. Gebruik Redis Pub/Sub niet voor deze events: het levert elk bericht hooguit één keer af, en een subscriber die op dat moment is losgekoppeld, mist het (Redis Pub/Sub delivery semantics).

Behandel modeloutput als een onbetrouwbaar voorstel

Een LLM-integratie lijkt op een anti-corruption layer: ze vertaalt een externe, probabilistische representatie naar termen die het domein begrijpt. De vertaallaag controleert alleen de vorm en de waarden van het voorstel. Toestemming blijft in de application service en bedrijfsregels blijven in de aggregate.

Een port is een interface die de operaties opsomt die de applicatie nodig heeft. De repository-port beschrijft hoe taaklijsten worden geladen en opgeslagen; een database-adapter implementeert die operaties. De figuur tekent de vier packages als ringen en toont codeafhankelijkheden, niet de volgorde van uitvoering. De applicatie gebruikt de aggregate en de repository-port; de infrastructuuradapter implementeert die port. Het domein hangt niet af van de database-adapter en ook niet van het model.

In een Python-project kunnen de vier lagen vier packages zijn: domain/, application/, infrastructure/ en interface/. Een andere database raakt dan alleen infrastructure/; een andere modelprovider raakt alleen interface/.

Het volgende diagram laat de vier aanroepen uit het uitgewerkte voorbeeld hieronder door de vijf stappen van de lijst eronder lopen. Elke rij toont waar een aanroep stopt en wat de tool teruggeeft.

De code tussen het model en het domein voert deze stappen uit:

1. Ontvang het voorstel: het model geeft AddTaskProposal terug; behandel elke waarde erin als onbetrouwbaar.
2. Parseer en normaliseer: valideer het tegen het getypeerde contract en los daarna datums, eenheden, identifiers en locale op met vertrouwde context.
3. Autoriseer: bepaal of deze actor de operatie mag aanvragen.
4. Handhaaf de invariant: roep een aggregate-methode aan die de regel controleert en TaskAdded aanmaakt.
5. Commit samen: sla de taaklijst en het outbox-event op in één transactie.

Drie tools kunnen het model getypeerde output laten teruggeven en die valideren (stap 1 en 2):

• Pydantic AI, wanneer je een agentframework wilt. Stel het outputtype van de agent in op een Pydantic-model; als de output de validatie niet doorstaat, stuurt het de fout terug naar het model en vraagt het opnieuw.
• Instructor, wanneer je de SDK van je provider houdt en alleen geparste Pydantic-objecten nodig hebt. Het wikkelt de client in, en max_retries vraagt het model opnieuw met de validatiefout.
• De eigen modus voor structured output van de provider, zoals OpenAI Structured Outputs, wanneer je wilt dat de provider de generatie beperkt tot JSON Schema.

Stuur een validatiefout een of twee keer terug naar het model. Als het voorstel nog steeds faalt, vraag het dan aan de gebruiker in plaats van opnieuw te proberen.

Pydantic kan een ontbrekend veld afwijzen, of een prioriteit buiten low, normal, high, via zijn getypeerde validatiemodel (Pydantic-documentatie). Het kan niet bepalen dat “morgen” naar de juiste datum wordt omgezet, dat de gebruiker eigenaar is van de taaklijst, of dat er al een vergelijkbare taak openstaat.

Uitgewerkt voorbeeld: een taak toevoegen

Neem een verzoek om “Send the invoice” aan een taaklijst toe te voegen. De actor is de ingelogde aanvrager (actor_id); de owner is de persoon wiens lijst wordt gewijzigd (owner_id). Dat kunnen verschillende personen zijn, dus de applicatie controleert de toestemming voor die actor en die lijst voordat ze de wijziging doorvoert.

De model-adapter levert de taakdetails aan als AddTaskProposal. De application service controleert de toestemming, vraagt TaskList de taak toe te voegen en commit de wijziging en het bijbehorende TaskAdded-event via een unit of work. Sla de zeven Python-blokken vanaf deze sectie tot en met “Voer het voorbeeld uit” op volgorde op als task_example.py. Ze vormen één module voor Python 3.12 met Pydantic 2; het laatste blok voert die uit.

1. Definieer het voorstel dat het model ziet

Houd het voorstel dicht bij wat het model kan afleiden. Vraag het niet om database-ID’s of vertrouwde eigenaar-identifiers te verzinnen.

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

Als de gebruiker “morgen” zegt, moet de applicatie het model een expliciete lokale datum geven of de relatieve uitdrukking oplossen met een geteste datumparser. Gebruik de klok van de inference-server nooit als impliciete bedrijfscontext.

2. Zet de invariant in de aggregate

TaskList.add_task() weigert lege omschrijvingen en dubbele taken voordat de lijst wordt gewijzigd. Doordat die controles hier staan, gelden ze zowel voor een verzoek van een LLM als voor een gewoon formulier.

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

Het domein raiset zijn eigen exceptietypes. pydantic.ValidationError is een subklasse van ValueError, dus een tool die ValueError afving om een domeinconflict te melden, zou ook ongeldige voorstellen afvangen.

TaskAdded is een frozen dataclass: een value object dat het feit vastlegt. De aggregate bewaart taken in een immutable tuple, zodat aanroepers geen muteerbare collectie krijgen waaraan ze kunnen toevoegen of waarmee ze add_task() kunnen omzeilen. De muterende methoden vervangen die onderliggende tuple pas nadat de regel is gehandhaafd.

De detectie van duplicaten is hier bewust eenvoudig. Echte regels kunnen locale-bewuste normalisatie, semantiek voor terugkerende taken of een uniekheidsconstraint in de database nodig hebben als laatste vangnet tegen race conditions.

3. Definieer de ports

Deze Python-protocols beschrijven wat de applicatie nodig heeft van opslag en van toestemmingscontroles. TaskListRepository laadt en bewaart één taaklijst. UnitOfWork bevat die repository en de openstaande events, en commit beide. TaskAuthorizer raiset Unauthorized wanneer de actor de lijst niet mag wijzigen.

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

De database-adapter mag alleen opslaan als de opgeslagen versie nog overeenkomt met expected_version, de versie die eerder is geladen. Een geslaagde opslag verhoogt die versie. Als een ander verzoek de lijst heeft gewijzigd, raiset de adapter ConcurrentUpdate in plaats van het werk van dat verzoek te overschrijven. Dit is optimistic concurrency control; de documentatie over versietellers van SQLAlchemy toont de onderliggende vergelijking. Het protocol beschrijft de eis; een concrete adapter moet de controle en de update samen implementeren.

4. Coördineer de use case

AddTaskService zet toestemming, laden, de domeinoperatie en persistence in de juiste volgorde. De identifiers van actor en owner komen uit vertrouwde applicatiecontext, niet uit het voorstel van het model.

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

Als een regel vóór commit() een exception veroorzaakt, verwijdert de unit of work de nog niet gecommitte wijziging van de taaklijst en het event. Een verbindingsfout of timeout tijdens commit() kan het resultaat onbekend laten: de database kan al hebben gecommit voordat het antwoord verloren ging. Het verlaten van het blok kan die gecommitte transactie niet ongedaan maken (Amazon Builders’ Library, Making retries safe with idempotent APIs).

Het model komt in deze service niet voor. Eén adapter kan AddTaskProposal van een LLM halen, een andere uit een HTTP-formulier, en tests kunnen het direct construeren. Het bedrijfsgedrag blijft identiek.

5. Voeg in-memory adapters toe

Met deze adapters draait het voorbeeld zonder database. InMemoryUnitOfWork.commit() controleert eerst alle versies en schrijft pas daarna de taaklijsten en de events, dus slaat het beide op of geen van beide. AllowList is een nep-authorizer voor tests en demo’s.

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

Een database-adapter implementeert UnitOfWork met één SQL-transactie; de in-memory versie toont het contract.

Koppel tools aan application commands

Agent-tools horen use cases bloot te leggen, geen database-primitieven. Geef de voorkeur aan:

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

boven:

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

De eerste set spreekt de taal van het domein en geeft de applicatie een plek om te autoriseren en regels af te dwingen. De tweede laat het model willekeurige persistence-mutaties beschrijven. De figuur stuurt hetzelfde duplicaatverzoek door beide ontwerpen: de databasetool slaat een tweede kopie op en de domeincommando-tool weigert het.

Agentframeworkcode, zoals een LangGraph-node of een Pydantic AI-tool, hoort in de interfacelaag en roept application services aan, zoals add_task_tool() hieronder doet. Als orchestration in de applicatielaag een framework gebruikt, zet het dan achter een port die de applicatie zelf bezit. Het domein importeert het framework nooit.

Een tool result moet fouten onderscheiden waarop de agent kan reageren. Platslaan van alle fouten tot één string nodigt uit tot blind opnieuw proberen. Elke fout vraagt om een andere vervolgactie:

Hier meldt unavailable een infrastructuurexception; het bewijst niet dat er niets is opgeslagen. Probeer alleen opnieuw met backoff nadat de applicatie heeft bevestigd dat er niets is geschreven, of wanneer zij de schrijfactie en het resultaat atomair opslaat onder een verzoek-ID en dat resultaat bij nieuwe pogingen teruggeeft. Hergebruik dezelfde ID en argumenten binnen de bewaartermijn (Amazon Builders’ Library, Making retries safe with idempotent APIs). Bepaal anders het resultaat aan de hand van de gegevens van de applicatie. Als die gegevens geen uitsluitsel geven, vertel de gebruiker dan dat het resultaat onbekend is.

De demo weigert taken met dezelfde inhoud, maar implementeert geen idempotentie voor verzoeken: een nieuwe poging na een gecommitte schrijfactie geeft domain_conflict terug in plaats van het oorspronkelijke succes.

Deze toolfunctie omhult AddTaskService.execute() en geeft die codes terug. Het framework geeft de actor en de owner door vanuit de session van de ingelogde gebruiker, niet vanuit het model.

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

Voer het voorbeeld uit

Het laatste blok roept de tool vier keer aan: één geldige taak, één duplicaat met andere hoofdletters en spaties, één prioriteit buiten de toegestane waarden en één actor zonder toestemming.

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

Voer met Python 3.12 geïnstalleerd deze commando’s uit in de map met task_example.py:

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

De module drukt dit af:

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

Alleen de eerste aanroep wijzigde state. De opgeslagen lijst staat op versie 1 met één taak, en de outbox bevat één TaskAdded-event.

Tests voor elke laag

Domeintests

Test aggregates met pytest, zonder model, netwerk of database:

• dubbele taken worden geweigerd
• geldige taken produceren het verwachte event
• blootgestelde collecties kunnen de interne state niet muteren
• overgangsregels blijven gelden bij herhaalde operaties

De tweede aanroep in de run hierboven controleert het eerste punt al via de tool. Een domeintest doet dezelfde aanroep direct op TaskList en verwacht DuplicateTask.

Applicatietests

Gebruik fake repositories en authorizers om laden, de volgorde van autorisatie, opslaan met verwachte versie, outbox-gedrag en foutmapping te verifiëren. De in-memory unit of work hierboven is die fake. Sla het volgende blok apart op als test_task_example.py, naast task_example.py. Het opent twee units of work op dezelfde data om twee verzoeken te simuleren die dezelfde versie laden:

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

Installeer pytest in dezelfde omgeving en voer de test uit:

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

Evaluaties van het modelcontract

Evalueer de probabilistische adapter apart:

• nauwkeurigheid van intentie- en veldextractie
• oplossen van relatieve datums met aangeleverde tijdzonecontext
• weigering of verduidelijkingsvraag wanneer vereiste informatie ontbreekt
• weerstand tegen prompt injection in geciteerde tekst van de taak
• aandeel voorstellen dat schema-geldig maar inhoudelijk onbruikbaar is

Voor tests die geen echt model mogen aanroepen biedt Pydantic AI TestModel, dat de tools van de agent aanroept en schema-geldige output teruggeeft, en FunctionModel, waarmee de test elk antwoord zelf bepaalt (Pydantic AI testing). Bewaar de evaluaties hierboven voor het echte model.

Een end-to-end test moet daarna bevestigen dat slechte voorstellen nooit dezelfde domeinmethoden omzeilen die vertrouwde interfaces gebruiken.

Volgorde van werk voor een nieuw agentproject

1. Interview de domeinexperts en schrijf de glossary, met een definitie en een voorbeeld voor elke term.
2. Teken de context map. Begin met één core context.
3. Modelleer de entities en value objects. Zet elke regel in de methode die de state wijzigt.
4. Kies de aggregate roots en schrijf hun invarianten op.
5. Definieer de ports voor repository, unit of work en authorizer. Gebruik in-memory adapters totdat de domeintests slagen.
6. Produceer domain events voor wijzigingen die andere contexten nodig hebben, en schrijf ze in dezelfde transactie naar een outbox.
7. Definieer de voorstelstypen voor het model en de parser die ze valideert.
8. Maak elke use case beschikbaar als tool die een application service aanroept en de fouten ervan koppelt.
9. Voeg orchestration als laatste toe, waarbij commando’s tussen contexten worden gestuurd.

Wanneer het ontwerp werkt

Je moet de modelprovider kunnen wisselen zonder een domeintest te wijzigen. Een beleidswijziging moet één aggregate of domain service raken in plaats van meerdere prompts. Traces moeten termen gebruiken die het eigenaarsteam herkent. Misvormde of niet-geautoriseerde voorstellen moeten falen vóór persistence, en een gelijktijdige opslag moet falen in plaats van stilzwijgend state te overschrijven.

Het model blijft verkeerde voorstellen doen; de aggregate weigert degene die een regel schenden.

Referenties

• Eric Evans, Domain-Driven Design Reference — definities van strategische en tactische patronen
• Martin Fowler, Bounded Context — waarom één model niet elke betekenis van een term moet dekken
• Martin Fowler, Repository — abstractie van persistence als collectie
• Harry Percival en Bob Gregory, Architecture Patterns with Python: Unit of Work — het unit-of-work-patroon in Python
• Chris Richardson, Transactional Outbox — publiceren na een databasetransactie zonder events te verliezen
• Pydantic-documentatie — validatie van getypeerde voorstellen
• Pydantic AI output — structured output met validatie-retries
• Instructor retrying — gevalideerde Pydantic-output bovenop een provider-SDK