Domain-Driven Design für AI Agents: Kontexte und Regeln

Viacheslav Dubrov

2025-10-20 · Aktualisiert 2026-10-08 · 16 Min. · KI-Agenten

Automatische Übersetzung

Dieser Artikel wurde automatisch aus der englischen Originalversion übersetzt.

Dieser Leitfaden richtet sich an Python-Entwickler, die AI Agents bauen, die Geschäftsdaten ändern. Du lernst, wie Domain-Driven Design (DDD) Fachbegriffe und Geschäftsregeln ordnet und wie du aus dem Vorschlag eines Models eine autorisierte Operation machst.

Agent-Projekte lassen sich schwer ändern, wenn Prompts, Code und Geschäftsprozesse unterschiedliche Begriffe verwenden. Die Compliance-Abteilung spricht von einer „Richtlinienprüfung”, im Code steht aber process_data(). Der vage Name verbirgt, welche Regel angewendet wird, wer sie verantwortet und wo eine Änderung hingehört.

DDD stellt diese Fachsprache und die Zuständigkeiten in den Mittelpunkt. Das Language Model interpretiert eine Anfrage und schlägt eine Operation mit benannten, typisierten Feldern vor. Der Anwendungscode liefert vertrauenswürdige Nutzerinformationen, prüft die Berechtigung und ruft die Geschäftsregeln auf, bevor eine Änderung gespeichert wird.

Das Problem ist die Zuständigkeit für Regeln

Agent-Systeme verteilen eine Regel oft auf einen System Prompt, eine Tool-Beschreibung, einen API-Handler und ein Datenbank-Constraint. Die Kopien laufen auseinander. Ein Erstattungslimit ändert sich, ein Prompt bleibt veraltet, und ein syntaktisch gültiger Tool Call trifft auf die falsche Richtlinie.

DDD beginnt mit anderen Fragen:

• Welches Team verantwortet die Regel?
• Welche Sprache verwenden die Fachexperten dafür?
• Innerhalb welcher Grenze hat dieser Begriff genau eine Bedeutung?
• Welche Zustandsänderungen müssen gemeinsam konsistent bleiben?

Diese Fragen helfen, wenn ein Workflow wichtig genug ist, um Richtlinien und einen Lebenszyklus zu haben. Ein einfacher Chatbot, der nur liest, braucht die später beschriebenen Code-Patterns womöglich nicht. Nutze sie, wenn ein Agent Zustand mit Regeln ändert; lass sie weg bei reinen Lese-Funktionen oder Features mit nur einem Schritt.

Stelle diese vier Fragen der Reihe nach. Jedes Ja fügt ein Pattern hinzu, das spätere Abschnitte erklären.

1. Ändert der Agent Geschäftsdaten? Wenn nicht, nutze Fachbegriffe in Tools und Prompts und höre auf.
2. Entscheiden Regeln, ob eine Änderung erlaubt ist? Füge ein Aggregate und einen Application Service hinzu.
3. Hat ein Begriff für verschiedene Teams verschiedene Bedeutungen? Füge Bounded Contexts und eine Context Map hinzu.
4. Muss ein anderer Service auf die Änderung reagieren? Füge ein Domain Event und eine Transactional Outbox hinzu.

Strategisches Design vor dem Code

Eine Ubiquitous Language aufbauen

Ein Bounded Context ist ein Teil des Systems, in dem Fachbegriffe und Regeln eine einheitlich vereinbarte Bedeutung haben. Seine Ubiquitous Language ist das Vokabular, das Fachexperten und Entwickler gemeinsam nutzen. Wenn der Support RefundRequest, approval limit und settlement sagt, sollten genau diese Begriffe in Anforderungen, Code, Tool-Verträgen und Evals vorkommen.

Das ist mehr, als beschreibende Methodennamen zu wählen. Begriffe brauchen Definitionen und Beispiele. Bedeutet „genehmigt”, dass ein Manager einen Knopf gedrückt hat, dass der Zahlungsdienstleister die Überweisung akzeptiert hat, oder beides? Eine Mehrdeutigkeit, die in einem Glossar auffällt, ist billiger als eine, die erst im Agent Trace auffällt.

Bounded Contexts um Models und Zuständigkeiten ziehen

Ein Domain Model besteht aus den Objekten, Daten und Regeln, die einen Geschäftsbereich abbilden; es ist nicht dasselbe wie das Language Model. Dasselbe Substantiv kann in verschiedenen Kontexten Unterschiedliches bedeuten. „Produkt” kann in der Lagerhaltung eine Artikelnummer sein, in der Abrechnung eine bepreiste Position und in der Auftragsverwaltung ein Lieferversprechen.

Die Abbildung vergleicht eine gemeinsame Product-Klasse mit einem Product-Model pro Kontext. In der unteren Hälfte ist jeder Pfeil eine Abbildung, die Code ausführt, wenn eine Auftragsposition in einen anderen Kontext wechselt.

Jeder Kontext besitzt sein Domain Model. Wenn Informationen in einen anderen Kontext übergehen, bildet Code die Begriffe und Bezeichner zwischen beiden ab. Fowler beschreibt diese Trennung als Weg, ein Model davon abzuhalten, jede Bedeutung eines Begriffs abzudecken (Bounded Contexts). Ein Bounded Context ist nicht automatisch ein Microservice, Repository, Agent oder Team, auch wenn diese Grenzen oft zusammenfallen.

Für das Agent-Design heißt das:

• Ein Kontext kann intern mehrere Model-Aufrufe oder spezialisierte Agents nutzen.
• Ein Agent, der mehrere Kontexte überspannt, braucht für jeden eine explizite Übersetzung und Befugnis.
• Orchestration ist ein Anwendungsthema; sie hebt die fachliche Zuständigkeit nicht auf.

Die nächste Abbildung zeigt diese drei Fälle. Vergleiche die zweite und dritte Spalte: Im zweiten übersetzt ein Agent Begriffe für zwei Kontexte; im dritten sendet ein Orchestrator nur Commands, und jeder Kontext wendet seine eigenen Regeln an.

Beginne mit einer Context Map, die die Geschäftsbereiche und ihren Informationsaustausch zeigt, bevor du einen Agent-Graphen zeichnest. Sonst bildet der Graph eher die verfügbaren Tools ab als das Geschäft.

Die Subdomains klassifizieren

DDD unterscheidet üblicherweise:

• Core Domain: die Fähigkeit, die differenzierenden Wert schafft
• Supporting Subdomain: notwendige, geschäftsspezifische Arbeit, die kein Unterscheidungsmerkmal ist
• Generic Subdomain: eine gelöste Fähigkeit wie Identität oder E-Mail-Versand

Bei einem Task-Assistenten kann Task Management die Core Domain sein, Scheduling eine Supporting Subdomain und die Benachrichtigungszustellung eine Generic Subdomain. Die Map platziert jeden Kontext in seinem Subdomain-Band. Scheduling liefert Termine an Task Management, das jede hinzugefügte Aufgabe über ein TaskAdded-Event an Notifications meldet.

Investiere Modellierungsaufwand in die Core Domain und schreibe ihre Regeln selbst. Nutze für Generic Subdomains einen bestehenden Dienst, etwa einen E-Mail-Anbieter für Benachrichtigungen. Die Klassifikation entscheidet nicht, wo ein LLM eingesetzt wird.

Taktische Patterns definieren die Zustandsgrenze

Entities und Value Objects

Eine Entity hat eine Identität und einen Lebenszyklus. Eine Aufgabe bleibt dieselbe Aufgabe, auch wenn sich ihre Beschreibung ändert. Ein Value Object ist durch seine Werte definiert und in der Regel unveränderlich: eine E-Mail-Adresse, ein Geldbetrag oder ein Zeitfenster.

Der Agent-Zustand passt zu derselben Aufteilung. Eine ConversationSession mit ihrem Nachrichtenverlauf ist eine Entity, ebenso jede Task, die der Agent plant. Ein aufgezeichnetes Tool Result, etwa ein ToolOutput mit den Aufrufargumenten und der Antwort, ist ein Value Object: Nach dem Speichern ändert es sich nicht mehr. Sobald eine Task ein Objekt ist, kann sie den Abschluss verweigern, bis ihre Abhängigkeiten erledigt sind, und diese Regel steht in einer Methode statt in mehreren Prompt-Vorlagen.

Aggregates und Invariants

Ein Aggregate gruppiert zusammengehörige Objekte, deren Änderungen denselben Regeln folgen müssen (Evans, Domain-Driven Design Reference). Anderer Code ändert diese Objekte über ein einziges Einstiegsobjekt, die Aggregate Root.

Im folgenden Beispiel ist TaskList die Root und ihre Task-Objekte sind Mitglieder. Eine Invariant ist eine Regel, die nach jeder Operation gültig bleiben muss. Diese Liste lehnt eine Aufgabe ab, die dieselbe normalisierte Beschreibung und dasselbe Fälligkeitsdatum hat wie eine vorhandene Aufgabe. Andere Domänen haben andere Invariants:

• Eine abgeschlossene Aufgabe kann nicht erneut abgeschlossen werden.
• Ein Eigentümer kann keine doppelten offenen Erinnerungen für denselben Tag haben.
• Eine Erstattung darf den verbleibenden erstattungsfähigen Betrag nicht übersteigen.
• Ein Plan nimmt höchstens 10 offene Aufgaben an, auch wenn das Model 50 vorschlägt.

Ein Aggregate wird nicht schon deshalb sicher, weil eine Python-Liste hinter einer add_task()-Methode liegt. Externer Code darf keine veränderbare Referenz erhalten, die diese Methode umgeht. Die Persistenz braucht außerdem Concurrency Control, sonst können zwei gültige Anfragen beim gleichzeitigen Speichern eine Invariant verletzen.

Repositories und Application Services

Ein Repository lädt und speichert Aggregates, ohne Datenbankdetails in die Domäne durchsickern zu lassen (Fowler, Repository). Ein Application Service koordiniert genau einen Use Case: Zustand laden, die Domänenoperation aufrufen, mit erwarteter Version speichern und entstandene Events veröffentlichen.

Eine Unit of Work fasst die Speichervorgänge eines Use Case in einer Datenbanktransaktion zusammen: Sie schreibt alle oder keinen (Percival und Gregory, Unit of Work). Der Application Service öffnet sie, nimmt seine Änderungen vor und ruft commit() auf. Das Verlassen ohne Aufruf von commit() verwirft die noch nicht committeten Änderungen.

Die Domäne sollte kein LLM, keinen HTTP-Client und kein ORM aufrufen. Das gehört in Adapter, also den Code, der die Anwendung mit externen Systemen verbindet.

Domain Events sind Fakten, kein Message Bus

TaskAdded hält einen Fakt fest: Eine Aufgabe wurde hinzugefügt. Die Anwendung kann dieses Event in derselben Datenbanktransaktion wie die Aufgabenänderung in einer Outbox-Tabelle speichern. Die Transaktion speichert beide Änderungen zusammen oder keine. Ein separater Publisher sendet das gespeicherte Event erst nach erfolgreicher Transaktion an andere Dienste. Das ist das Transactional-Outbox-Pattern; ein Event-Objekt allein bietet es nicht (Richardson, Transactional Outbox). Wer direkt aus einer Entity an einen Broker sendet, riskiert, ein Event für eine Transaktion zu veröffentlichen, die danach scheitert.

Die Abbildung vergleicht drei Abläufe: Die Entity sendet das Event an den Broker, und die Transaktion wird danach zurückgerollt; das Outbox-Muster mit Rollback; und das Outbox-Muster mit Commit.

Events können Agents koordinieren, machen die Koordination aber nicht von selbst zuverlässig. Liefersemantik, Idempotenz, Reihenfolge und versionierte Verträge bleiben Infrastrukturarbeit.

Wähle das Ziel des Publishers danach, wer das Event konsumiert. Laufen die Listener im selben Prozess, kann eine In-Process-Signal-Bibliothek wie blinker sie nach dem Commit aufrufen. Konsumieren andere Dienste das Event, sende es an einen Broker wie RabbitMQ oder an Redis Streams, falls das Team bereits Redis betreibt. Nutze für diese Events nicht Redis Pub/Sub: Es liefert jede Nachricht höchstens einmal, und ein getrennter Subscriber verliert sie (Redis Pub/Sub delivery semantics).

Model Output als nicht vertrauenswürdigen Vorschlag behandeln

Eine LLM-Integration ähnelt einem Anti-Corruption-Layer: Sie übersetzt eine externe, probabilistische Darstellung in Begriffe, die die Domäne versteht. Die Übersetzungsschicht prüft nur Form und Werte des Vorschlags. Die Berechtigung bleibt im Application Service, und die Geschäftsregeln bleiben im Aggregate.

Ein Port ist eine Schnittstelle, die die Operationen auflistet, welche die Anwendung braucht. Der Repository-Port nennt, wie Aufgabenlisten geladen und gespeichert werden; ein Datenbank-Adapter implementiert diese Operationen. Die Abbildung zeichnet die vier Packages als Ringe und zeigt Code-Abhängigkeiten, nicht die Ausführungsreihenfolge. Die Anwendung nutzt das Aggregate und den Repository-Port; der Infrastruktur-Adapter implementiert diesen Port. Die Domäne hängt weder vom Datenbank-Adapter noch vom Model ab.

In einem Python-Projekt können die vier Schichten vier Pakete sein: domain/, application/, infrastructure/ und interface/. Ein Datenbankwechsel betrifft dann infrastructure/; ein Wechsel des Model-Anbieters betrifft interface/.

Das nächste Diagramm führt die vier Aufrufe aus dem folgenden Beispiel durch die fünf Schritte der Liste darunter. Jede Zeile zeigt, wo ein Aufruf stoppt und was das Tool zurückgibt.

Der Code zwischen Model und Domäne führt diese Schritte aus:

1. Vorschlag empfangen: Das Model liefert AddTaskProposal; behandle jeden Wert darin als nicht vertrauenswürdig.
2. Parsen und normalisieren: Validiere ihn gegen den typisierten Vertrag und löse dann Datumsangaben, Einheiten, Bezeichner und Locale mit vertrauenswürdigem Kontext auf.
3. Autorisieren: Entscheide, ob dieser Akteur die Operation anfordern darf.
4. Invariant durchsetzen: Rufe eine Aggregate-Methode auf, die die Regel prüft und TaskAdded auslöst.
5. Atomar committen: Speichere Aufgabenliste und Outbox-Event in einer Transaktion.

Drei Werkzeuge können das Model dazu bringen, typisierte Ausgabe zu liefern, und sie validieren (Schritte 1 und 2):

• Pydantic AI, wenn du ein Agent-Framework willst. Setze den Output-Typ des Agents auf ein Pydantic-Model; schlägt die Validierung fehl, sendet es den Fehler an das Model zurück und fragt erneut.
• Instructor, wenn du dein Provider-SDK behältst und nur geparste Pydantic-Objekte brauchst. Es umhüllt den Client, und max_retries fragt das Model mit dem Validierungsfehler erneut.
• Der Structured Output-Modus des Anbieters, etwa OpenAI Structured Outputs, wenn der Anbieter die Generierung auf das JSON Schema beschränken soll.

Sende einen Validierungsfehler ein- oder zweimal an das Model zurück. Besteht der Vorschlag danach immer noch nicht, frage den Nutzer, statt es weiter zu versuchen.

Pydantic kann ein fehlendes Feld oder eine Priorität außerhalb von low, normal, high über sein typisiertes Validierungsmodell ablehnen (Pydantic documentation). Es kann nicht entscheiden, dass „morgen” zum richtigen Datum aufgelöst wird, dass der Nutzer die Aufgabenliste besitzt oder dass eine ähnliche Aufgabe bereits offen ist.

Durchgespieltes Beispiel: eine Aufgabe hinzufügen

Stell dir eine Anfrage vor, „Send the invoice” zu einer Aufgabenliste hinzuzufügen. Der Actor ist der angemeldete Anfragende (actor_id); der Owner ist die Person, deren Liste sich ändert (owner_id). Das können verschiedene Personen sein, daher prüft die Anwendung die Berechtigung für diesen Actor und diese Liste, bevor sie etwas ändert.

Der Model-Adapter liefert die Aufgabendetails als AddTaskProposal. Der Application Service prüft die Berechtigung, bittet TaskList, die Aufgabe hinzuzufügen, und committet die Änderung samt TaskAdded-Event über eine Unit of Work. Speichere die sieben Python-Blöcke von diesem Abschnitt bis einschließlich „Das Beispiel ausführen” der Reihe nach als task_example.py. Sie bilden ein Modul für Python 3.12 mit Pydantic 2; der letzte Block führt es aus.

1. Den Model-seitigen Vorschlag definieren

Halte den Vorschlag nah an dem, was das Model ableiten kann. Verlange von ihm keine erfundenen Datenbank-IDs oder vertrauenswürdigen Owner-Bezeichner.

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

Sagt der Nutzer „morgen”, sollte die Anwendung dem Model ein explizites lokales Datum geben oder den relativen Ausdruck mit einem getesteten Datums-Parser auflösen. Verwende nie die Uhr des Inference Servers als impliziten Geschäftskontext.

2. Die Invariant im Aggregate verankern

TaskList.add_task() lehnt leere Beschreibungen und doppelte Aufgaben ab, bevor es die Liste ändert. Weil diese Prüfungen hier liegen, gelten sie, egal ob eine Anfrage von einem LLM oder aus einem normalen Formular kommt.

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

Die Domäne wirft eigene Exception-Typen. pydantic.ValidationError ist eine Unterklasse von ValueError; ein Tool, das ValueError fängt, um einen Domänenkonflikt zu melden, würde daher auch ungültige Vorschläge fangen.

TaskAdded ist eine Frozen Dataclass: ein Value Object, das den Fakt festhält. Das Aggregate hält Aufgaben in einem unveränderlichen Tupel, sodass Aufrufer keine veränderbare Collection erhalten, an die sie anhängen oder die sie an add_task() vorbei bearbeiten könnten. Seine verändernden Methoden ersetzen dieses Tupel erst, nachdem sie die Regel durchgesetzt haben.

Die Duplikaterkennung ist hier bewusst einfach. Echte Regeln können eine locale-abhängige Normalisierung, Wiederholungssemantik oder ein Unique-Constraint der Datenbank als letzte, Race-sichere Absicherung brauchen.

3. Die Ports definieren

Diese Python-Protocols listen auf, was die Anwendung von Speicher und Berechtigungsprüfung braucht. TaskListRepository lädt und speichert eine Aufgabenliste. UnitOfWork hält dieses Repository und die ausstehenden Events und committet beide. TaskAuthorizer wirft Unauthorized, wenn der Actor die Liste nicht ändern darf.

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

Der Datenbank-Adapter darf nur speichern, wenn die gespeicherte Version noch expected_version entspricht, also der zuvor geladenen Version. Ein erfolgreiches Speichern erhöht diese Version. Hat eine andere Anfrage die Liste geändert, wirft er ConcurrentUpdate, statt deren Arbeit zu überschreiben. Das ist Optimistic Concurrency Control; die Dokumentation zum Versionszähler von SQLAlchemy zeigt den zugrunde liegenden Vergleich. Das Protocol nennt die Anforderung; ein konkreter Adapter muss Prüfung und Update gemeinsam umsetzen.

4. Den Use Case koordinieren

AddTaskService bringt Berechtigung, Laden, Domänenoperation und Persistenz in die richtige Reihenfolge. Die Bezeichner von Actor und Owner stammen aus vertrauenswürdigem Anwendungskontext, nicht aus dem Model-Vorschlag.

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

Wirft eine Zeile vor commit() eine Exception, verwirft die Unit of Work die noch nicht committete Änderung der Aufgabenliste und das Event. Ein Verbindungsfehler oder Timeout während commit() kann das Ergebnis unklar lassen: Die Datenbank könnte bereits committet haben, bevor ihre Antwort verloren ging. Das Verlassen des Blocks kann diese committete Transaktion nicht rückgängig machen (Amazon Builders’ Library, Making retries safe with idempotent APIs).

Das Model kommt in diesem Service nicht vor. Ein Adapter kann AddTaskProposal von einem LLM beziehen, ein anderer aus einem HTTP-Formular, und Tests können ihn direkt konstruieren. Das Geschäftsverhalten bleibt identisch.

5. In-Memory-Adapter hinzufügen

Mit diesen Adaptern läuft das Beispiel ohne Datenbank. InMemoryUnitOfWork.commit() prüft zuerst alle Versionen und schreibt erst dann Aufgabenlisten und Events, speichert also beides oder nichts. AllowList ist ein Fake-Authorizer für Tests und Demos.

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

Ein Datenbank-Adapter implementiert UnitOfWork mit einer SQL-Transaktion; die In-Memory-Version zeigt den Vertrag.

Tools auf Application Commands abbilden

Agent-Tools sollten Use Cases bereitstellen, keine Datenbank-Primitive. Bevorzuge:

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

statt:

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

Die erste Gruppe spricht die Domänensprache und gibt der Anwendung einen Ort, um zu autorisieren und Regeln durchzusetzen. Die zweite lässt das Model beliebige Persistenz-Mutationen beschreiben. Die Abbildung sendet dieselbe doppelte Anfrage durch beide Designs: Das Datenbank-Tool speichert eine zweite Kopie, das Domain-Command-Tool lehnt sie ab.

Agent-Framework-Code, etwa ein LangGraph-Node oder ein Tool von Pydantic AI, gehört in die Interface-Schicht und ruft Application Services auf, wie es add_task_tool() unten tut. Nutzt die Orchestration in der Anwendungsschicht ein Framework, hänge es hinter einen Port, den die Anwendung besitzt. Die Domäne importiert das Framework nie.

Ein Tool Result sollte Fehler unterscheiden, auf die der Agent reagieren kann. Reduziere nicht alle Fehler auf einen String, der zu blinden Wiederholungen einlädt. Jeder Fehler hat eine andere nächste Aktion:

Hier meldet unavailable eine Infrastruktur-Exception; es beweist nicht, dass nichts gespeichert wurde. Wiederhole mit Backoff nur, nachdem die Anwendung bestätigt hat, dass kein Schreibvorgang stattgefunden hat, oder wenn sie den Schreibvorgang und sein Ergebnis atomar unter einer Request-ID speichert und dieses Ergebnis bei Wiederholungen zurückgibt. Verwende innerhalb der Aufbewahrungsfrist dieselbe ID und dieselben Argumente (Amazon Builders’ Library, Making retries safe with idempotent APIs). Andernfalls kläre das Ergebnis anhand der Aufzeichnungen der Anwendung. Lässt es sich damit nicht klären, informiere den Nutzer, dass das Ergebnis unbekannt ist.

Die Demo lehnt doppelte Aufgabeninhalte ab, implementiert aber keine Request-Idempotenz: Eine Wiederholung nach einem committeten Schreibvorgang liefert domain_conflict statt des ursprünglichen Erfolgs.

Diese Tool-Funktion umhüllt AddTaskService.execute() und liefert diese Codes. Das Framework übergibt Actor und Owner aus der angemeldeten Session, nicht vom 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)}

Das Beispiel ausführen

Der letzte Block ruft das Tool viermal auf: eine gültige Aufgabe, ein Duplikat mit anderer Groß- und Kleinschreibung und anderen Leerzeichen, eine Priorität außerhalb der erlaubten Werte und einen Actor ohne Berechtigung.

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

Wenn Python 3.12 installiert ist, führe diese Befehle im Verzeichnis mit task_example.py aus:

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

Das Modul gibt aus:

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

Nur der erste Aufruf hat den Zustand geändert. Die gespeicherte Liste hat Version 1 mit einer Aufgabe, und die Outbox enthält ein TaskAdded-Event.

Tests für jede Schicht

Domain-Tests

Teste Aggregates mit pytest, ohne Model, Netzwerk oder Datenbank:

• Doppelte Aufgaben werden abgelehnt.
• Gültige Aufgaben lösen das erwartete Event aus.
• Nach außen gegebene Collections können den internen Zustand nicht verändern.
• Übergangsregeln gelten auch bei wiederholten Operationen.

Der zweite Aufruf im Lauf oben prüft den ersten Punkt bereits über das Tool. Ein Domain-Test macht denselben Aufruf direkt auf TaskList und erwartet DuplicateTask.

Application-Tests

Nutze Fake-Repositories und -Authorizer, um Laden, Reihenfolge der Autorisierung, Speichern mit erwarteter Version, Outbox-Verhalten und Fehlerabbildung zu prüfen. Die In-Memory-Unit-of-Work oben ist dieser Fake. Speichere den folgenden Block separat als test_task_example.py neben task_example.py. Er öffnet zwei Units of Work auf denselben Daten, um zwei Anfragen zu simulieren, die dieselbe Version 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"]

Installiere pytest in derselben Umgebung und führe den Test aus:

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

Evaluierung des Model-Vertrags

Evaluiere den probabilistischen Adapter getrennt:

• Genauigkeit bei Intent- und Feldextraktion
• Auflösung relativer Datumsangaben mit mitgelieferter Zeitzonen-Information
• Ablehnung oder Rückfrage, wenn erforderliche Informationen fehlen
• Widerstandsfähigkeit gegen Prompt Injection in zitiertem Aufgabentext
• Anteil der schemagültigen, aber fachlich unbrauchbaren Vorschläge

Für Tests, die kein echtes Model aufrufen sollen, bietet Pydantic AI TestModel, das die Tools des Agents aufruft und schemagültige Ausgabe liefert, und FunctionModel, mit dem der Test jede Antwort selbst bestimmt (Pydantic AI testing). Hebe dir die oben genannten Evaluierungen für das echte Model auf.

Ein End-to-End-Test sollte danach bestätigen, dass schlechte Vorschläge nie an denselben Domänenmethoden vorbeikommen, die vertrauenswürdige Schnittstellen nutzen.

Reihenfolge der Arbeit in einem neuen Agent-Projekt

1. Interviewe die Fachexperten und schreibe das Glossar, mit Definition und Beispiel für jeden Begriff.
2. Zeichne die Context Map. Beginne mit einem Core Context.
3. Modelliere Entities und Value Objects. Lege jede Regel in die Methode, die den Zustand ändert.
4. Wähle die Aggregate Roots und schreibe ihre Invariants auf.
5. Definiere die Ports für Repository, Unit of Work und Authorizer. Nutze In-Memory-Adapter, bis die Domain-Tests bestehen.
6. Löse Domain Events für Änderungen aus, die andere Kontexte brauchen, und schreibe sie in derselben Transaktion in eine Outbox.
7. Definiere die Model-seitigen Vorschlagstypen und den Parser, der sie validiert.
8. Stelle jeden Use Case als Tool bereit, das einen Application Service aufruft und dessen Fehler abbildet.
9. Füge die Orchestration zuletzt hinzu und sende Commands zwischen Kontexten.

Woran du erkennst, dass das Design funktioniert

Du solltest den Model-Anbieter wechseln können, ohne einen Domain-Test zu ändern. Eine Richtlinienänderung sollte ein Aggregate oder einen Domain Service betreffen statt mehrerer Prompts. Traces sollten Begriffe verwenden, die das zuständige Team erkennt. Fehlerhafte oder nicht autorisierte Vorschläge sollten vor der Persistenz scheitern, und ein gleichzeitiges Speichern sollte fehlschlagen, statt den Zustand stillschweigend zu überschreiben.

Das Model wird weiterhin falsche Vorschläge machen; das Aggregate lehnt die ab, die eine Regel brechen.

Referenzen

• Eric Evans, Domain-Driven Design Reference — Definitionen strategischer und taktischer Patterns
• Martin Fowler, Bounded Context — warum ein Model nicht jede Bedeutung eines Begriffs abdecken sollte
• Martin Fowler, Repository — Abstraktion der Persistenz als Collection
• Harry Percival und Bob Gregory, Architecture Patterns with Python: Unit of Work — das Unit-of-Work-Pattern in Python
• Chris Richardson, Transactional Outbox — Veröffentlichen nach einer Datenbanktransaktion, ohne Events zu verlieren
• Pydantic documentation — Validierung typisierter Vorschläge
• Pydantic AI output — Structured Output mit Validierungs-Retries
• Instructor retrying — validierte Pydantic-Ausgabe auf Basis eines Provider-SDK