Domain-Driven Design pour les AI agents : contextes et règles

Viacheslav Dubrov

2025-10-20 · Mis à jour 2026-10-08 · 19 min · Agents IA

Traduction automatique

Cet article a été traduit automatiquement depuis la version originale en anglais.

Ce guide s’adresse aux ingénieurs Python qui construisent des AI agents modifiant des données métier. Vous verrez comment le Domain-Driven Design (DDD) organise le vocabulaire et les règles métier, et comment transformer la proposition d’un modèle en opération autorisée.

Un projet d’agent devient difficile à faire évoluer quand les prompts, le code et les processus métier emploient des termes différents. La conformité demande un « contrôle de politique », alors que l’implémentation expose process_data(). Ce nom vague cache la règle appliquée, son responsable et l’endroit où modifier le code.

Le DDD place ce langage métier et ces responsabilités au centre. Le modèle de langage interprète une demande et propose une opération avec des champs nommés et typés. Le code applicatif fournit les informations fiables sur l’utilisateur, vérifie la permission et appelle les règles métier avant d’enregistrer un changement.

Le problème : à qui appartient la règle

Dans les systèmes d’agents, une même règle se répartit souvent entre un system prompt, une description de tool, un handler d’API et une contrainte de base de données. Les copies divergent. Un plafond de remboursement change, un prompt garde l’ancienne valeur, et un tool call syntaxiquement valide atteint la mauvaise politique.

Le DDD commence par d’autres questions :

• Quelle équipe est responsable de la règle ?
• Quel langage les experts du domaine emploient-ils pour la décrire ?
• Dans quelle frontière ce terme a-t-il un seul sens ?
• Quels changements d’état doivent rester cohérents entre eux ?

Ces questions servent quand un workflow est assez important pour avoir des politiques et un cycle de vie. Un simple chatbot en lecture seule n’a pas forcément besoin des patterns de code décrits plus loin dans cet article. Utilisez ces patterns quand un agent modifie un état soumis à des règles ; ignorez-les pour les fonctionnalités en lecture seule ou en une seule étape.

Posez ces quatre questions dans l’ordre. Chaque réponse « oui » ajoute un pattern que les sections suivantes expliquent.

1. L’agent modifie-t-il des données métier ? Sinon, utilisez les termes du domaine dans les tools et les prompts, et arrêtez-vous là.
2. Des règles décident-elles si un changement est autorisé ? Ajoutez un aggregate et un service applicatif.
3. Un même terme a-t-il des sens différents selon les équipes ? Ajoutez des bounded contexts et une context map.
4. Un autre service doit-il réagir au changement ? Ajoutez un événement de domaine et une transactional outbox.

La conception stratégique avant le code

Construire un ubiquitous language

Un bounded context est une partie du système où les termes et les règles métier ont un sens unique et convenu. Son ubiquitous language est le vocabulaire partagé par les experts du domaine et les développeurs. Si les opérations de support disent RefundRequest, approval limit et settlement, ces termes doivent apparaître dans les exigences, le code, les contrats de tools et les évaluations.

Il ne s’agit pas seulement de choisir des noms de méthodes descriptifs. Les termes ont besoin de définitions et d’exemples. « Approuvé » signifie-t-il qu’un manager a cliqué sur un bouton, que le processeur de paiement a accepté le virement, ou les deux ? Une ambiguïté découverte dans un glossaire coûte moins cher qu’une ambiguïté découverte dans une trace d’agent.

Tracer les bounded contexts autour des modèles et des responsabilités

Un domain model désigne les objets, les données et les règles qui représentent un domaine d’activité ; il ne faut pas le confondre avec le modèle de langage. Un même nom peut avoir des sens différents selon le contexte. « Produit » peut être une unité de gestion de stock dans l’Inventaire, une ligne tarifée dans la Facturation et un engagement de livraison dans la Gestion des commandes.

La figure compare une seule classe Product partagée et un modèle Product par contexte. Dans la moitié basse, chaque flèche est un mapping que le code exécute quand une ligne de commande passe à un autre contexte.

Chaque contexte possède son domain model. Quand une information passe à un autre contexte, le code convertit les termes et les identifiants entre les deux. Fowler décrit cette séparation comme un moyen d’éviter qu’un seul modèle couvre tous les sens d’un terme (bounded contexts). Ce n’est pas automatiquement un microservice, un dépôt, un agent ou une équipe, même si ces frontières coïncident souvent.

Pour la conception d’agents, cela signifie :

• un contexte peut utiliser en interne plusieurs appels de modèle ou des agents spécialisés
• un agent qui couvre plusieurs contextes a besoin d’une traduction explicite et d’une autorité pour chacun
• l’orchestration relève de l’application ; elle n’efface pas la responsabilité du domaine

La figure suivante dessine ces trois cas. Comparez la deuxième et la troisième colonne : dans le deuxième, un agent traduit les termes de deux contextes ; dans le troisième, un orchestrateur envoie seulement des commandes, et chaque contexte applique ses propres règles.

Avant de dessiner un graphe d’agents, commencez par une context map qui montre les domaines d’activité et la manière dont ils échangent des informations. Sinon, le graphe reproduit plutôt les tools disponibles que le métier.

Classer les subdomains

Le DDD distingue couramment :

• Core domain : la capacité qui crée une valeur différenciante
• Supporting subdomain : un travail nécessaire et propre à l’entreprise, mais qui ne la différencie pas
• Generic subdomain : une capacité déjà résolue, comme l’identité ou l’envoi d’e-mails

Pour un assistant de tâches, la gestion des tâches peut être le core domain, la planification un supporting subdomain et l’envoi de notifications un generic subdomain. La carte place chaque contexte dans la bande de son subdomain. Scheduling fournit les dates à Task Management, qui signale chaque tâche ajoutée à Notifications par un événement TaskAdded.

Concentrez l’effort de modélisation sur le core domain et écrivez vous-même ses règles. Pour les generic subdomains, utilisez un service existant, par exemple un fournisseur d’e-mail pour les notifications. Cette classification ne décide pas où placer un LLM.

Les patterns tactiques définissent la frontière d’état

Entities et value objects

Une entity a une identité et un cycle de vie. Une tâche reste la même tâche après la modification de sa description. Un value object est défini par ses valeurs et il est en général immuable : une adresse e-mail, un montant, une plage horaire.

L’état d’un agent suit le même découpage. Une ConversationSession avec son historique de messages est une entity, tout comme chaque Task que l’agent planifie. Un tool result enregistré, comme un ToolOutput avec les arguments de l’appel et la réponse, est un value object : il ne change plus une fois stocké. Dès qu’une Task est un objet, elle peut refuser son achèvement tant que ses dépendances ne sont pas terminées, et cette règle se trouve dans une seule méthode au lieu de plusieurs templates de prompt.

Aggregates et invariants

Un aggregate regroupe des objets liés dont les changements doivent respecter les mêmes règles (Evans, Domain-Driven Design Reference). Le reste du code modifie ces objets par un seul objet d’entrée, appelé aggregate root.

Dans l’exemple ci-dessous, TaskList est la root et ses objets Task en sont les membres. Un invariant est une règle qui doit rester vraie après chaque opération. Cette liste rejette une tâche dont la description normalisée et la date d’échéance sont identiques à celles d’une tâche existante. D’autres domaines ont d’autres invariants :

• une tâche terminée ne peut pas être terminée une seconde fois
• un propriétaire ne peut pas avoir deux rappels ouverts identiques pour le même jour
• un remboursement ne peut pas dépasser le montant encore remboursable
• un plan accepte au plus 10 tâches ouvertes, même si le modèle en propose 50

Un aggregate ne devient pas sûr du seul fait qu’une liste Python se trouve derrière une méthode add_task(). Le code externe ne doit pas recevoir de référence modifiable qui contourne cette méthode. La persistance a aussi besoin d’un contrôle de concurrence, sinon deux requêtes valides enregistrées en même temps peuvent violer un invariant.

Repositories et services applicatifs

Un repository charge et enregistre des aggregates sans propager les détails de la base de données dans le domaine (Fowler, Repository). Un service applicatif coordonne un use case : charger l’état, appeler l’opération du domaine, enregistrer avec une version attendue et publier les événements produits.

Une unit of work regroupe les enregistrements d’un use case dans une seule transaction de base de données : elle les valide tous ou aucun (Percival et Gregory, Unit of Work). Le service applicatif l’ouvre, effectue ses changements et appelle commit(). Sortir sans appeler commit() abandonne les changements non encore validés.

Le domaine ne doit appeler ni un LLM, ni un client HTTP, ni un ORM. Ces appels relèvent des adapters, le code qui relie l’application aux systèmes externes.

Les domain events sont des faits, pas un message bus

TaskAdded enregistre un fait : une tâche a été ajoutée. L’application peut stocker cet événement dans une table outbox, dans la même transaction de base de données que le changement de la tâche. La transaction enregistre les deux changements ensemble, ou aucun. Un publisher distinct envoie ensuite l’événement stocké aux autres services, seulement après le succès de la transaction. C’est le pattern transactional outbox ; un objet événement seul ne l’offre pas (Richardson, Transactional Outbox). Envoyer directement à un broker depuis une entity risque de publier un événement pour une transaction qui échoue ensuite.

La figure compare trois exécutions : l’entité envoie l’événement au broker puis la transaction est annulée ; l’outbox avec un rollback ; et l’outbox avec un commit.

Les événements peuvent coordonner des agents, mais ils ne rendent pas la coordination fiable à eux seuls. Les garanties de livraison, l’idempotence, l’ordre et les contrats versionnés restent du travail d’infrastructure.

Choisissez la cible du publisher selon le consommateur de l’événement. Quand les listeners s’exécutent dans le même processus, une bibliothèque de signaux en processus comme blinker peut les appeler après le commit. Quand d’autres services consomment l’événement, envoyez-le à un broker comme RabbitMQ, ou à Redis Streams si l’équipe exploite déjà Redis. N’utilisez pas Redis Pub/Sub pour ces événements : il livre chaque message au plus une fois, et un abonné déconnecté le perd (sémantique de livraison de Redis Pub/Sub).

Traiter la sortie du modèle comme une proposition non fiable

Une intégration de LLM ressemble à une anti-corruption layer : elle traduit une représentation externe et probabiliste en termes que le domaine comprend. La couche de traduction vérifie seulement la forme et les valeurs de la proposition. La permission reste dans le service applicatif et les règles métier dans l’aggregate.

Un port est une interface qui liste les opérations dont l’application a besoin. Le repository port liste la manière de charger et d’enregistrer les listes de tâches ; un adapter de base de données implémente ces opérations. La figure dessine les quatre packages comme des anneaux et montre les dépendances du code, pas l’ordre d’exécution. L’application utilise l’aggregate et le repository port ; l’adapter d’infrastructure implémente ce port. Le domaine ne dépend ni de l’adapter de base de données ni du modèle.

Dans un projet Python, les quatre couches peuvent être quatre packages : domain/, application/, infrastructure/ et interface/. Changer de base de données ne touche alors que infrastructure/ ; changer de fournisseur de modèle ne touche que interface/.

Le diagramme suivant fait passer les quatre appels de l’exemple détaillé ci-dessous par les cinq étapes de la liste qui le suit. Chaque ligne montre où un appel s’arrête et ce que le tool renvoie.

Le code situé entre le modèle et le domaine exécute ces étapes :

1. Recevoir la proposition : le modèle renvoie AddTaskProposal ; traitez chaque valeur comme non fiable.
2. Analyser et normaliser : validez-la selon le contrat typé, puis résolvez les dates, les unités, les identifiants et la locale à partir d’un contexte fiable.
3. Autoriser : décidez si cet acteur peut demander l’opération.
4. Faire respecter l’invariant : appelez une méthode de l’aggregate qui vérifie la règle et produit TaskAdded.
5. Valider ensemble : enregistrez la liste de tâches et l’événement outbox dans une seule transaction.

Trois outils permettent de faire renvoyer au modèle une sortie typée et de la valider (étapes 1 et 2) :

• Pydantic AI, si vous voulez un framework d’agents. Définissez le type de sortie de l’agent comme un modèle Pydantic ; quand la sortie échoue à la validation, il renvoie l’erreur au modèle et redemande.
• Instructor, si vous gardez le SDK de votre fournisseur et n’avez besoin que d’objets Pydantic analysés. Il enveloppe le client, et max_retries interroge de nouveau le modèle avec l’erreur de validation.
• Le mode structured output du fournisseur, comme OpenAI Structured Outputs, si vous voulez que le fournisseur contraigne la génération au JSON Schema.

Renvoyez une erreur de validation au modèle une ou deux fois. Si la proposition échoue encore, interrogez l’utilisateur au lieu de réessayer.

Pydantic peut rejeter un champ manquant ou une priorité hors de low, normal, high grâce à son modèle de validation typé (documentation Pydantic). Il ne peut pas décider que « demain » correspond à la bonne date, que l’utilisateur possède la liste de tâches, ou qu’une tâche similaire est déjà ouverte.

Exemple complet : ajouter une tâche

Prenons une demande d’ajout de « Send the invoice » à une liste de tâches. L’acteur est le demandeur connecté (actor_id) ; le propriétaire est la personne dont la liste va changer (owner_id). Ils peuvent être différents, donc l’application vérifie la permission de cet acteur sur cette liste avant de faire le changement.

L’adapter du modèle fournit les détails de la tâche sous la forme AddTaskProposal. Le service applicatif vérifie la permission, demande à TaskList d’ajouter la tâche, puis valide le changement et son événement TaskAdded par une unit of work. Enregistrez les sept blocs Python depuis cette section jusqu’à « Exécuter l’exemple », dans l’ordre, dans task_example.py. Ils forment un module pour Python 3.12 avec Pydantic 2 ; le dernier bloc le lance.

1. Définir la proposition destinée au modèle

Gardez la proposition proche de ce que le modèle peut déduire. Ne lui demandez pas d’inventer des identifiants de base de données ni des identifiants de propriétaire fiables.

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

Si l’utilisateur dit « demain », l’application doit donner au modèle une date locale explicite, ou résoudre l’expression relative avec un parseur de dates testé. N’utilisez jamais l’horloge du serveur d’inférence comme contexte métier implicite.

2. Placer l’invariant dans l’aggregate

TaskList.add_task() rejette les descriptions vides et les tâches en double avant de modifier la liste. En plaçant ces contrôles ici, on les applique que la requête vienne d’un LLM ou d’un formulaire ordinaire.

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

Le domaine lève ses propres types d’exception. pydantic.ValidationError est une sous-classe de ValueError : un tool qui attraperait ValueError pour signaler un conflit du domaine attraperait aussi les propositions invalides.

TaskAdded est une dataclass figée : un value object qui enregistre le fait. L’aggregate garde les tâches dans un tuple immuable, donc les appelants ne reçoivent aucune collection modifiable qui leur permette d’ajouter ou de changer des éléments en contournant add_task(). Ses méthodes de modification remplacent ce tuple sous-jacent seulement après avoir appliqué la règle.

La détection des doublons est volontairement simple ici. Des règles réelles peuvent exiger une normalisation sensible à la locale, une sémantique de récurrence, ou une contrainte d’unicité en base comme dernière protection contre les accès concurrents.

3. Définir les ports

Ces protocoles Python listent ce que l’application attend du stockage et des contrôles de permission. TaskListRepository charge et enregistre une liste de tâches. UnitOfWork contient ce repository et les événements en attente, et valide les deux. TaskAuthorizer lève Unauthorized quand l’acteur n’a pas le droit de modifier la liste.

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

L’adapter de base de données ne doit enregistrer que si la version stockée correspond encore à expected_version, la version chargée plus tôt. Un enregistrement réussi incrémente cette version. Si une autre requête a modifié la liste, il lève ConcurrentUpdate au lieu d’écraser le travail de cette requête. C’est le contrôle de concurrence optimiste ; la documentation du compteur de version de SQLAlchemy montre la comparaison sous-jacente. Le protocole énonce l’exigence ; un adapter concret doit implémenter ensemble la vérification et la mise à jour.

4. Coordonner le use case

AddTaskService ordonne la permission, le chargement, l’opération du domaine et la persistance. Les identifiants de l’acteur et du propriétaire viennent du contexte applicatif fiable, pas de la proposition du modèle.

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

Si une ligne avant commit() lève une exception, la unit of work abandonne le changement de la liste de tâches et l’événement non encore validés. Une erreur de connexion ou un timeout pendant commit() peut laisser son résultat inconnu : la base de données peut avoir validé la transaction avant que sa réponse soit perdue. Sortir du bloc ne peut pas annuler cette transaction déjà validée (Amazon Builders’ Library, Making retries safe with idempotent APIs).

Le modèle est absent de ce service. Un adapter peut obtenir AddTaskProposal d’un LLM, un autre d’un formulaire HTTP, et les tests peuvent le construire directement. Le comportement métier reste identique.

5. Ajouter des adapters en mémoire

Ces adapters permettent d’exécuter l’exemple sans base de données. InMemoryUnitOfWork.commit() vérifie d’abord toutes les versions, puis seulement écrit les listes de tâches et les événements, donc il enregistre les deux ou aucun. AllowList est un faux authorizer pour les tests et les démonstrations.

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

Un adapter de base de données implémente UnitOfWork avec une seule transaction SQL ; la version en mémoire illustre le contrat.

Associer les tools aux commandes applicatives

Les tools d’un agent doivent exposer des use cases, pas des primitives de base de données. Préférez :

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

à :

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

Le premier ensemble parle le langage du domaine et donne à l’application un endroit où autoriser et faire respecter les règles. Le second laisse le modèle décrire des mutations de persistance arbitraires. La figure envoie la même requête en doublon dans les deux conceptions : le tool de base de données enregistre une seconde copie, et le tool de commande du domaine la refuse.

Le code du framework d’agents, comme un nœud LangGraph ou un tool Pydantic AI, appartient à la couche d’interface et appelle les services applicatifs, comme le fait add_task_tool() ci-dessous. Si l’orchestration de la couche applicative utilise un framework, placez-le derrière un port que l’application possède. Le domaine n’importe jamais le framework.

Un tool result doit distinguer les échecs sur lesquels l’agent peut agir. N’aplatissez pas tous les échecs en une chaîne qui invite à réessayer à l’aveugle. Chaque échec appelle une action différente :

Ici, unavailable signale une exception d’infrastructure ; il ne prouve pas que rien n’a été enregistré. Réessayez avec un backoff seulement après que l’application a confirmé qu’aucune écriture n’a eu lieu, ou si elle enregistre atomiquement l’écriture et son résultat sous un identifiant de requête et renvoie ce résultat lors des nouvelles tentatives. Réutilisez le même identifiant et les mêmes arguments pendant sa durée de conservation (Amazon Builders’ Library, Making retries safe with idempotent APIs). Sinon, déterminez le résultat à partir des enregistrements de l’application. Si ces enregistrements ne permettent pas de le déterminer, indiquez à l’utilisateur que le résultat est inconnu.

La démonstration rejette les tâches dont le contenu est identique, mais n’implémente pas l’idempotence des requêtes : une nouvelle tentative après une écriture validée renvoie domain_conflict au lieu du succès initial.

Cette fonction de tool enveloppe AddTaskService.execute() et renvoie ces codes. Le framework transmet l’acteur et le propriétaire à partir de la session connectée, pas du modèle.

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

Exécuter l’exemple

Le dernier bloc appelle le tool quatre fois : une tâche valide, un doublon avec une casse et des espaces différents, une priorité hors des valeurs autorisées, et un acteur sans permission.

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

Une fois Python 3.12 installé, exécutez ces commandes dans le répertoire contenant task_example.py :

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

Le module affiche :

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

Seul le premier appel a changé l’état. La liste stockée est à la version 1 avec une tâche, et l’outbox contient un événement TaskAdded.

Des tests pour chaque couche

Tests du domaine

Testez les aggregates avec pytest, sans modèle, sans réseau et sans base de données :

• les tâches en double sont rejetées
• les tâches valides produisent l’événement attendu
• les collections exposées ne peuvent pas modifier l’état interne
• les règles de transition tiennent après des opérations répétées

Le deuxième appel de l’exécution ci-dessus vérifie déjà le premier point à travers le tool. Un test du domaine fait le même appel directement sur TaskList et attend DuplicateTask.

Tests applicatifs

Utilisez de faux repositories et authorizers pour vérifier le chargement, l’ordre des autorisations, les enregistrements à version attendue, le comportement de l’outbox et la conversion des erreurs. La unit of work en mémoire ci-dessus est ce faux. Enregistrez le bloc suivant séparément dans test_task_example.py, à côté de task_example.py. Il ouvre deux units of work sur les mêmes données pour simuler deux requêtes qui chargent la même version :

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

Installez pytest dans le même environnement et exécutez le test :

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

Évaluations du contrat du modèle

Évaluez séparément l’adapter probabiliste :

• précision de la détection de l’intention et de l’extraction des champs
• résolution des dates relatives avec un contexte de fuseau horaire fourni
• refus ou demande de clarification quand une information requise manque
• résistance aux attaques de prompt injection dans le texte cité d’une tâche
• taux de propositions valides selon le schéma mais inutilisables sur le plan sémantique

Pour les tests qui ne doivent pas appeler un vrai modèle, Pydantic AI fournit TestModel, qui appelle les tools de l’agent et renvoie une sortie valide selon le schéma, et FunctionModel, qui laisse le test décider de chaque réponse (tests Pydantic AI). Gardez les évaluations ci-dessus pour le vrai modèle.

Un test de bout en bout doit ensuite confirmer que les mauvaises propositions ne contournent jamais les méthodes du domaine qu’utilisent les interfaces fiables.

Ordre de travail pour un nouveau projet d’agent

1. Interrogez les experts du domaine et écrivez le glossaire, avec une définition et un exemple pour chaque terme.
2. Dessinez la context map. Commencez par un seul core context.
3. Modélisez les entities et les value objects. Placez chaque règle dans la méthode qui change l’état.
4. Choisissez les aggregate roots et écrivez leurs invariants.
5. Définissez les ports repository, unit of work et authorizer. Utilisez des adapters en mémoire jusqu’à ce que les tests du domaine passent.
6. Produisez des domain events pour les changements dont d’autres contextes ont besoin, et écrivez-les dans une outbox dans la même transaction.
7. Définissez les types de proposition destinés au modèle et le parseur qui les valide.
8. Exposez chaque use case comme un tool qui appelle un service applicatif et convertit ses échecs.
9. Ajoutez l’orchestration en dernier, en envoyant des commandes entre les contextes.

Quand la conception fonctionne

Vous devez pouvoir changer de fournisseur de modèle sans modifier un test du domaine. Un changement de politique doit toucher un aggregate ou un domain service au lieu de plusieurs prompts. Les traces doivent employer des termes que l’équipe responsable reconnaît. Les propositions mal formées ou non autorisées doivent échouer avant la persistance, et un enregistrement concurrent doit échouer au lieu d’écraser l’état en silence.

Le modèle fera encore de mauvaises propositions ; l’aggregate rejette celles qui enfreignent une règle.

Références

• Eric Evans, Domain-Driven Design Reference — définitions des patterns stratégiques et tactiques
• Martin Fowler, Bounded Context — pourquoi un seul modèle ne doit pas couvrir tous les sens d’un terme
• Martin Fowler, Repository — abstraction de collection pour la persistance
• Harry Percival et Bob Gregory, Architecture Patterns with Python: Unit of Work — le pattern unit of work en Python
• Chris Richardson, Transactional Outbox — publier après une transaction de base de données sans perdre d’événements
• Documentation Pydantic — validation typée des propositions
• Pydantic AI output — structured output avec nouvelles tentatives de validation
• Instructor retrying — sortie Pydantic validée au-dessus d’un SDK de fournisseur