Domain-Driven Design para AI Agents: contextos y reglas

Viacheslav Dubrov

2025-10-20 · Actualizado 2026-10-08 · 19 min · Agentes de IA

Traducción automática

Este artículo se tradujo automáticamente a partir de la versión original en inglés.

Esta guía está dirigida a ingenieros de Python que construyen AI agents que modifican datos de negocio. Aprenderás cómo el Domain-Driven Design (DDD) organiza el vocabulario y las reglas de negocio, y cómo convertir una propuesta del modelo en una operación autorizada.

Los proyectos de agentes se vuelven difíciles de cambiar cuando los prompts, el código y los procesos de negocio usan términos distintos. Compliance pide una «comprobación de políticas», mientras que la implementación expone process_data(). Ese nombre vago oculta qué regla se aplica, quién es su responsable y dónde debe hacerse un cambio.

DDD pone ese lenguaje de negocio y esa responsabilidad en el centro. El modelo de lenguaje interpreta una petición y propone una operación con campos nombrados y tipados. El código de aplicación aporta la información fiable del usuario, comprueba los permisos e invoca las reglas de negocio antes de guardar un cambio.

El problema es quién es responsable de cada regla

Los sistemas de agentes suelen repartir una misma regla entre un system prompt, la descripción de una tool, un API handler y una restricción de la base de datos. Las copias divergen. Cambia un límite de reembolso, un prompt conserva el valor antiguo y un tool call sintácticamente válido llega a la política equivocada.

DDD empieza haciendo otras preguntas:

• ¿Qué equipo es responsable de la regla?
• ¿Qué lenguaje usan los expertos del dominio para referirse a ella?
• ¿Dentro de qué límite tiene ese término un único significado?
• ¿Qué cambios de estado deben mantenerse consistentes entre sí?

Estas preguntas son útiles cuando un workflow es lo bastante importante como para tener políticas y un ciclo de vida. Un chatbot sencillo de solo lectura puede no necesitar los patrones de código que se describen más adelante en este artículo. Usa esos patrones cuando un agente cambie un estado sujeto a reglas; sáltatelos en funciones de solo lectura o de un único paso.

Hazte estas cuatro preguntas en orden. Cada respuesta afirmativa añade un patrón que se explica en secciones posteriores.

1. ¿El agente cambia datos de negocio? Si no, usa términos del dominio en las tools y los prompts y detente.
2. ¿Hay reglas que deciden si un cambio está permitido? Añade un aggregate y un servicio de aplicación.
3. ¿Un mismo término significa cosas distintas para equipos distintos? Añade bounded contexts y un context map.
4. ¿Otro servicio debe reaccionar al cambio? Añade un evento de dominio y una transactional outbox.

Diseño estratégico antes del código

Construir un lenguaje ubicuo

Un bounded context es una parte del sistema en la que los términos y las reglas de negocio tienen un único significado acordado. Su ubiquitous language es el vocabulario que comparten los expertos del dominio y los desarrolladores. Si en operaciones de soporte se dice RefundRequest, approval limit y settlement, esos términos deben aparecer en los requisitos, el código, los contratos de las tools y las evaluaciones.

Esto va más allá de elegir nombres descriptivos para los métodos. Los términos necesitan definiciones y ejemplos. ¿«Aprobado» significa que un manager ha pulsado un botón, que el procesador de pagos ha aceptado la transferencia, o ambas cosas? Descubrir una ambigüedad en un glosario es más barato que descubrirla en el trace de un agente.

Dibujar bounded contexts alrededor de los modelos y la responsabilidad

Un domain model son los objetos, los datos y las reglas que se usan para representar un área de negocio; no es lo mismo que el modelo de lenguaje. Un mismo sustantivo puede significar cosas distintas en contextos distintos. «Producto» puede ser una unidad de stock en Inventory, una línea con precio en Billing y un compromiso de entrega en Order Management.

La figura compara una única clase Product compartida con un modelo Product por contexto. En la mitad inferior, cada flecha es una correspondencia que el código aplica cuando una línea de pedido pasa a otro contexto.

Cada contexto es dueño de su domain model. Cuando la información pasa a otro contexto, el código traduce entre ellos los términos y los identificadores. Fowler describe esta separación como una forma de evitar que un único modelo abarque todos los significados de un término (bounded contexts). No es automáticamente un microservicio, un repositorio, un agente ni un equipo, aunque esos límites suelen coincidir.

En el diseño de agentes esto significa:

• un contexto puede usar internamente varias llamadas al modelo o agentes especializados
• un agente que abarca varios contextos necesita traducción explícita y autoridad para cada uno
• la orquestación es una responsabilidad de la aplicación; no elimina la responsabilidad sobre el dominio

La siguiente figura dibuja estos tres casos. Compara la segunda y la tercera columna: en el segundo, un agente traduce términos para dos contextos; en el tercero, un orquestador solo envía comandos y cada contexto aplica sus propias reglas.

Antes de dibujar un grafo de agentes, empieza con un context map que muestre las áreas de negocio y cómo intercambian información. Si no, el grafo tiende a reproducir las tools disponibles en lugar del negocio.

Clasificar los subdominios

DDD suele distinguir entre:

• Core domain: la capacidad que crea valor diferencial
• Supporting subdomain: trabajo necesario y específico del negocio que no es el elemento diferenciador
• Generic subdomain: una capacidad ya resuelta, como la identidad o el envío de correo electrónico

En un asistente de tareas, la gestión de tareas puede ser el core, la planificación un supporting subdomain y el envío de notificaciones un generic subdomain. El mapa sitúa cada contexto en su franja de subdomain. Scheduling proporciona fechas a Task Management, que informa de cada tarea añadida a Notifications mediante un evento TaskAdded.

Dedica el esfuerzo de modelado al core domain y escribe tú mismo sus reglas. Usa un servicio existente para los generic subdomains, como un proveedor de correo para las notificaciones. La clasificación no decide dónde va un LLM.

Los patrones tácticos definen la frontera del estado

Entidades y value objects

Una entidad tiene identidad y un ciclo de vida. Una tarea sigue siendo la misma tarea después de cambiar su descripción. Un value object se define por sus valores y suele ser inmutable: una dirección de correo, un importe o una franja horaria.

El estado de un agente encaja en la misma división. Una ConversationSession con su historial de mensajes es una entidad, y también lo es cada Task que planifica el agente. Un tool result registrado, como un ToolOutput con los argumentos de la llamada y la respuesta, es un value object: no cambia una vez almacenado. Cuando una Task es un objeto, puede negarse a completarse hasta que terminen sus dependencias, y esa regla está en un único método en lugar de en varias plantillas de prompt.

Aggregates e invariantes

Un aggregate agrupa objetos relacionados cuyos cambios deben cumplir las mismas reglas (Evans, Domain-Driven Design Reference). El resto del código modifica esos objetos a través de un único objeto de entrada, el aggregate root.

En el ejemplo de abajo, TaskList es el root y sus objetos Task son los miembros. Una invariante es una regla que debe seguir cumpliéndose después de cada operación. Esta lista rechaza una tarea con la misma descripción normalizada y la misma fecha de vencimiento que una tarea existente. Otros dominios tienen otras invariantes:

• una tarea completada no puede completarse otra vez
• un propietario no puede tener recordatorios abiertos duplicados para el mismo día
• un reembolso no puede superar el importe reembolsable restante
• un plan acepta como máximo 10 tareas abiertas, aunque el modelo proponga 50

Un aggregate no es seguro solo porque una lista de Python esté detrás de un método add_task(). El código externo no debe recibir una referencia mutable que se salte ese método. La persistencia también necesita control de concurrencia; si no, dos peticiones válidas pueden violar una invariante al guardarse a la vez.

Repositorios y application services

Un repositorio carga y guarda aggregates sin filtrar detalles de la base de datos al dominio (Fowler, Repository). Un application service coordina un caso de uso: carga el estado, invoca la operación del dominio, guarda con una versión esperada y publica los eventos resultantes.

Una unit of work agrupa los guardados de un caso de uso en una sola transacción de base de datos: confirma todos o ninguno (Percival y Gregory, Unit of Work). El application service la abre, hace sus cambios y llama a commit(). Salir sin llamar a commit() descarta los cambios aún no confirmados.

El dominio no debe llamar a un LLM, a un cliente HTTP ni a un ORM. Eso pertenece a los adapters, el código que conecta la aplicación con sistemas externos.

Los eventos de dominio son hechos, no un bus de mensajes

TaskAdded registra un hecho: se añadió una tarea. La aplicación puede guardar este evento en una outbox table dentro de la misma transacción de base de datos que el cambio de la tarea. La transacción guarda ambos cambios juntos o ninguno. Un publisher independiente envía el evento almacenado a otros servicios solo cuando la transacción ha tenido éxito. Este es el transactional-outbox pattern; un objeto de evento por sí solo no lo proporciona (Richardson, Transactional Outbox). Enviar directamente a un broker desde una entidad corre el riesgo de publicar un evento de una transacción que después falla.

La figura compara tres ejecuciones: la entidad envía el evento al broker y después la transacción hace rollback; el outbox con rollback; y el outbox con commit.

Los eventos pueden coordinar agentes, pero por sí solos no hacen fiable la coordinación. La semántica de entrega, la idempotencia, el orden y los contratos versionados siguen siendo trabajo de infraestructura.

Elige el destino del publisher según quién consuma el evento. Cuando los oyentes se ejecutan en el mismo proceso, una biblioteca de señales en proceso como blinker puede llamarlos después del commit. Cuando consumen el evento otros servicios, envíalo a un broker como RabbitMQ, o a Redis Streams si el equipo ya usa Redis. No uses Redis Pub/Sub para estos eventos: entrega cada mensaje como máximo una vez, y un suscriptor desconectado lo pierde (semántica de entrega de Redis Pub/Sub).

Trata la salida del modelo como una propuesta no fiable

Una integración con un LLM se parece a una anti-corruption layer: traduce una representación externa y probabilística a términos que el dominio entiende. La capa de traducción solo comprueba la forma y los valores de la propuesta. Los permisos se quedan en el application service y las reglas de negocio en el aggregate.

Un port es una interfaz que enumera las operaciones que la aplicación necesita. El port del repositorio enumera cómo cargar y guardar listas de tareas; un adapter de base de datos implementa esas operaciones. La figura dibuja los cuatro paquetes como anillos y muestra dependencias de código, no el orden de ejecución. La aplicación usa el aggregate y el port del repositorio; el adapter de infraestructura implementa ese port. El dominio no depende ni del adapter de base de datos ni del modelo.

En un proyecto de Python, las cuatro capas pueden ser cuatro paquetes: domain/, application/, infrastructure/ e interface/. Cambiar la base de datos solo afecta entonces a infrastructure/; cambiar el proveedor del modelo afecta a interface/.

El siguiente diagrama recorre las cuatro llamadas del ejemplo desarrollado más abajo a través de los cinco pasos de la lista que lo sigue. Cada fila muestra dónde se detiene una llamada y qué devuelve la tool.

El código entre el modelo y el dominio ejecuta estos pasos:

1. Recibir la propuesta: el modelo devuelve AddTaskProposal; trata cada valor que contiene como no fiable.
2. Analizar y normalizar: valídala contra el contrato tipado y después resuelve fechas, unidades, identificadores y configuración regional con contexto fiable.
3. Autorizar: decide si este actor puede solicitar la operación.
4. Hacer cumplir la invariante: invoca un método del aggregate que comprueba la regla y genera TaskAdded.
5. Confirmar juntos: guarda la lista de tareas y el evento de la outbox en una sola transacción.

Tres herramientas permiten que el modelo devuelva la salida tipada y la valide (pasos 1 y 2):

• Pydantic AI, cuando quieres un framework de agentes. Define el tipo de salida del agente como un modelo de Pydantic; cuando la salida no pasa la validación, devuelve el error al modelo y vuelve a preguntar.
• Instructor, cuando conservas el SDK de tu proveedor y solo necesitas objetos de Pydantic ya analizados. Envuelve el cliente, y max_retries vuelve a preguntar al modelo con el error de validación.
• El propio modo de structured output del proveedor, como OpenAI Structured Outputs, cuando quieres que el proveedor restrinja la generación al JSON Schema.

Devuelve un error de validación al modelo una o dos veces. Si la propuesta sigue fallando, pregunta al usuario en lugar de reintentar.

Pydantic puede rechazar un campo ausente o una prioridad fuera de low, normal, high mediante su modelo de validación tipado (documentación de Pydantic). No puede decidir que «mañana» se resuelve en la fecha correcta, que el usuario es el propietario de la lista de tareas ni que ya hay abierta una tarea similar.

Ejemplo práctico: añadir una tarea

Imagina una petición para añadir «Send the invoice» a una lista de tareas. El actor es quien hace la petición y ha iniciado sesión (actor_id); el owner es la persona cuya lista va a cambiar (owner_id). Pueden ser personas distintas, así que la aplicación comprueba el permiso de ese actor sobre esa lista antes de hacer el cambio.

El adapter del modelo aporta los detalles de la tarea como AddTaskProposal. El application service comprueba el permiso, le pide a TaskList que añada la tarea y confirma el cambio y su evento TaskAdded mediante una unit of work. Guarda los siete bloques de Python desde esta sección hasta «Ejecutar el ejemplo», en orden, como task_example.py. Forman un módulo para Python 3.12 con Pydantic 2; el último bloque lo ejecuta.

1. Definir la propuesta orientada al modelo

Mantén la propuesta cerca de lo que el modelo puede inferir. No le pidas que invente IDs de base de datos ni identificadores de propietario 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 el usuario dice «mañana», la aplicación debe darle al modelo una fecha local explícita o resolver la expresión relativa con un analizador de fechas probado. No uses nunca el reloj del servidor de inferencia como contexto de negocio implícito.

2. Poner la invariante en el aggregate

TaskList.add_task() rechaza las descripciones en blanco y las tareas duplicadas antes de modificar la lista. Al mantener esas comprobaciones aquí, se aplican tanto si la petición viene de un LLM como de un formulario normal.

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

El dominio lanza sus propios tipos de excepción. pydantic.ValidationError es una subclase de ValueError, así que una tool que capturara ValueError para informar de un conflicto de dominio capturaría también las propuestas no válidas.

TaskAdded es un dataclass inmutable (frozen): un value object que registra el hecho. El aggregate guarda las tareas en una tupla inmutable, de modo que quien lo llama no recibe ninguna colección mutable a la que pueda añadir elementos o que pueda editar saltándose add_task(). Sus métodos que modifican el estado reemplazan esa tupla de respaldo solo después de hacer cumplir la regla.

La detección de duplicados aquí es deliberadamente simple. Las reglas reales pueden necesitar normalización según la configuración regional, semántica de recurrencia o una restricción de unicidad en la base de datos como última salvaguarda frente a condiciones de carrera.

3. Definir los ports

Estos protocolos de Python enumeran lo que la aplicación necesita del almacenamiento y de las comprobaciones de permisos. TaskListRepository carga y guarda una lista de tareas. UnitOfWork contiene ese repositorio y los eventos pendientes, y confirma ambos. TaskAuthorizer lanza Unauthorized cuando el actor no puede modificar la lista.

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

El adapter de base de datos debe guardar solo si la versión almacenada sigue coincidiendo con expected_version, la versión cargada antes. Un guardado correcto incrementa esa versión. Si otra petición ha cambiado la lista, lanza ConcurrentUpdate en lugar de sobrescribir el trabajo de esa petición. Esto es el optimistic concurrency control; la documentación del contador de versión de SQLAlchemy muestra la comparación subyacente. El protocolo expone el requisito; un adapter concreto debe implementar la comprobación y la actualización juntas.

4. Coordinar el caso de uso

AddTaskService ordena el permiso, la carga, la operación del dominio y la persistencia. Sus identificadores de actor y de propietario proceden del contexto fiable de la aplicación, no de la propuesta del modelo.

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 una línea anterior a commit() lanza una excepción, la unit of work descarta el cambio de la lista de tareas y el evento que aún no se han confirmado. Un fallo de conexión o un timeout durante commit() puede dejar el resultado sin determinar: la base de datos puede haber confirmado la transacción antes de que se perdiera su respuesta. Salir del bloque no puede deshacer esa transacción confirmada (Amazon Builders’ Library, Making retries safe with idempotent APIs).

El modelo no aparece en este servicio. Un adapter puede obtener AddTaskProposal de un LLM, otro de un formulario HTTP, y los tests pueden construirlo directamente. El comportamiento de negocio es idéntico.

5. Añadir adapters en memoria

Estos adapters permiten ejecutar el ejemplo sin base de datos. InMemoryUnitOfWork.commit() comprueba primero todas las versiones y solo después escribe las listas de tareas y los eventos, así que guarda ambos o ninguno. AllowList es un authorizer falso para tests y 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}")

Un adapter de base de datos implementa UnitOfWork con una transacción SQL; la versión en memoria muestra el contrato.

Asignar tools a comandos de la aplicación

Las tools de un agente deben exponer casos de uso, no primitivas de base de datos. Prefiere:

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

en lugar de:

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

El primer conjunto habla el lenguaje del dominio y le da a la aplicación un lugar donde autorizar y hacer cumplir las reglas. El segundo deja que el modelo describa mutaciones arbitrarias de persistencia. La figura envía la misma petición duplicada por los dos diseños: la tool de base de datos guarda una segunda copia y la tool de comandos de dominio la rechaza.

El código del framework de agentes, como un nodo de LangGraph o una tool de Pydantic AI, pertenece a la capa de interfaz y llama a application services, como hace add_task_tool() más abajo. Si la orquestación de la capa de aplicación usa un framework, ponlo detrás de un port que la aplicación posea. El dominio nunca importa el framework.

Un tool result debe distinguir los fallos sobre los que el agente puede actuar. No reduzcas todos los fallos a una cadena de texto que invite a reintentos a ciegas. Cada fallo tiene una acción siguiente distinta:

Aquí, unavailable informa de una excepción de infraestructura; no demuestra que no se haya guardado nada. Reintenta con backoff solo cuando la aplicación confirme que no hubo ninguna escritura, o cuando guarde atómicamente la escritura y su resultado con un ID de petición y devuelva ese resultado en los reintentos. Reutiliza el mismo ID y los mismos argumentos dentro de su periodo de retención (Amazon Builders’ Library, Making retries safe with idempotent APIs). En los demás casos, averigua el resultado consultando los registros de la aplicación. Si esos registros no permiten determinarlo, informa al usuario de que el resultado es desconocido.

La demo rechaza el contenido de tareas duplicadas, pero no implementa la idempotencia de las peticiones: un reintento después de una escritura confirmada devuelve domain_conflict en lugar del éxito original.

Esta función de tool envuelve AddTaskService.execute() y devuelve esos códigos. El framework pasa el actor y el propietario desde la sesión iniciada, no desde el modelo.

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

Ejecutar el ejemplo

El último bloque llama a la tool cuatro veces: una tarea válida, un duplicado con otras mayúsculas y espaciado, una prioridad fuera de los valores permitidos y un actor sin permiso.

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

Con Python 3.12 instalado, ejecuta estos comandos en el directorio que contiene task_example.py:

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

El módulo imprime:

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

Solo la primera llamada cambió el estado. La lista almacenada está en la versión 1 con una tarea, y la outbox contiene un evento TaskAdded.

Tests para cada capa

Tests de dominio

Prueba los aggregates con pytest, sin modelo, red ni base de datos:

• se rechazan las tareas duplicadas
• las tareas válidas generan el evento esperado
• las colecciones expuestas no pueden mutar el estado interno
• las reglas de transición se mantienen tras operaciones repetidas

La segunda llamada de la ejecución anterior ya comprueba el primer punto a través de la tool. Un test de dominio hace la misma llamada directamente sobre TaskList y espera DuplicateTask.

Tests de aplicación

Usa repositorios y authorizers falsos para verificar la carga, el orden de la autorización, los guardados con versión esperada, el comportamiento de la outbox y la asignación de errores. La unit of work en memoria de arriba es ese fake. Guarda el siguiente bloque por separado como test_task_example.py, junto a task_example.py. Abre dos units of work sobre los mismos datos para simular dos peticiones que cargan la misma versión:

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

Instala pytest en el mismo entorno y ejecuta el test:

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

Evaluaciones del contrato del modelo

Evalúa por separado el adapter probabilístico:

• precisión en la detección de la intención y la extracción de campos
• resolución de fechas relativas con el contexto de zona horaria suministrado
• rechazo o petición de aclaración cuando falta información necesaria
• resistencia a los ataques de prompt injection dentro del texto citado de una tarea
• proporción de propuestas válidas según el esquema pero inutilizables semánticamente

Para los tests que no deben llamar a un modelo real, Pydantic AI ofrece TestModel, que llama a las tools del agente y devuelve una salida válida según el esquema, y FunctionModel, que deja que el test decida cada respuesta (testing en Pydantic AI). Reserva las evaluaciones anteriores para el modelo real.

Un test de extremo a extremo debe confirmar después que las propuestas incorrectas nunca se saltan los mismos métodos de dominio que usan las interfaces fiables.

Orden de trabajo para un nuevo proyecto de agentes

1. Entrevista a los expertos del dominio y escribe el glosario, con una definición y un ejemplo para cada término.
2. Dibuja el context map. Empieza con un único contexto core.
3. Modela las entidades y los value objects. Pon cada regla en el método que cambia el estado.
4. Elige los aggregate roots y anota sus invariantes.
5. Define los ports de repositorio, unit of work y authorizer. Usa adapters en memoria hasta que pasen los tests de dominio.
6. Genera eventos de dominio para los cambios que otros contextos necesitan, y escríbelos en una outbox en la misma transacción.
7. Define los tipos de propuesta orientados al modelo y el parser que los valida.
8. Expón cada caso de uso como una tool que llama a un application service y asigna sus fallos.
9. Añade la orquestación al final, enviando comandos entre contextos.

Cuándo funciona el diseño

Deberías poder cambiar el proveedor del modelo sin cambiar ningún test de dominio. Un cambio de política debería afectar a un aggregate o a un domain service en lugar de a varios prompts. Los traces deberían usar términos que el equipo responsable reconozca. Las propuestas mal formadas o no autorizadas deberían fallar antes de la persistencia, y un guardado concurrente debería fallar en lugar de sobrescribir el estado en silencio.

El modelo seguirá haciendo propuestas erróneas; el aggregate rechaza las que incumplen una regla.

Referencias

• Eric Evans, Domain-Driven Design Reference — definiciones de los patrones estratégicos y tácticos
• Martin Fowler, Bounded Context — por qué un modelo no debe abarcar todos los significados de un término
• Martin Fowler, Repository — la abstracción de colección para la persistencia
• Harry Percival y Bob Gregory, Architecture Patterns with Python: Unit of Work — el patrón unit of work en Python
• Chris Richardson, Transactional Outbox — publicar tras una transacción de base de datos sin perder eventos
• Documentación de Pydantic — validación tipada de la propuesta
• Salida de Pydantic AI — structured output con reintentos de validación
• Instructor retrying — salida de Pydantic validada sobre el SDK de un proveedor