Domain-Driven Design para AI Agents: Contextos e Regras

Viacheslav Dubrov

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

Tradução automática

Este artigo foi traduzido automaticamente a partir da versão original em inglês.

Este guia destina-se a engenheiros de Python que constroem AI agents que alteram dados de negócio. Vai aprender como o Domain-Driven Design (DDD) organiza o vocabulário e as regras do negócio, e como transformar a proposta de um modelo numa operação autorizada.

Os projetos de agentes tornam-se difíceis de alterar quando os prompts, o código e os processos de negócio usam termos diferentes. A equipa de compliance pede uma “verificação de política”, enquanto a implementação expõe process_data(). O nome vago esconde que regra está a ser aplicada, quem é o seu responsável e onde deve ser feita uma alteração.

O DDD coloca essa linguagem de negócio e a responsabilidade no centro. O modelo de linguagem interpreta um pedido e propõe uma operação com campos nomeados e tipados. O código da aplicação fornece os dados fiáveis do utilizador, verifica a permissão e invoca as regras de negócio antes de guardar uma alteração.

A responsabilidade pelas regras é o problema

Os sistemas de agentes costumam distribuir uma regra por um system prompt, uma descrição de tool, um handler de API e uma restrição da base de dados. As cópias divergem. O limite de reembolso muda, um prompt continua antigo, e uma tool call sintaticamente válida chega à política errada.

O DDD começa por fazer outras perguntas:

• Que equipa é responsável pela regra?
• Que linguagem usam os especialistas do domínio para a descrever?
• Dentro de que fronteira esse termo tem um único significado?
• Que alterações de estado têm de permanecer consistentes em conjunto?

Estas perguntas são úteis quando um fluxo de trabalho é suficientemente importante para ter políticas e um ciclo de vida. Um chatbot simples, só de leitura, pode não precisar dos padrões de código descritos mais à frente neste artigo. Use esses padrões quando um agente altera estado sujeito a regras; dispense-os em funcionalidades só de leitura ou de um único passo.

Faça estas quatro perguntas por ordem. Cada resposta afirmativa acrescenta um padrão que as secções seguintes explicam.

1. O agente altera dados de negócio? Se não, use termos do domínio nas tools e nos prompts e pare aqui.
2. Existem regras que decidem se uma alteração é permitida? Acrescente um aggregate e um application service.
3. Um termo significa coisas diferentes para equipas diferentes? Acrescente bounded contexts e um context map.
4. Outro serviço tem de reagir à alteração? Acrescente um domain event e uma transactional outbox.

Design estratégico antes do código

Criar uma linguagem ubíqua

Um bounded context é uma parte do sistema onde os termos e as regras de negócio têm um significado acordado. A sua linguagem ubíqua é o vocabulário partilhado por especialistas do domínio e programadores. Se as operações de suporte dizem RefundRequest, approval limit e settlement, esses termos devem aparecer nos requisitos, no código, nos contratos das tools e nas avaliações.

Isto vai além de escolher nomes descritivos para os métodos. Os termos precisam de definições e exemplos. “Aprovado” significa que um gestor clicou num botão, que o processador de pagamentos aceitou a transferência, ou ambos? Descobrir a ambiguidade num glossário custa menos do que descobri-la no trace de um agente.

Desenhar bounded contexts em torno dos modelos e da responsabilidade

Um domain model é o conjunto de objetos, dados e regras usados para representar uma área de negócio; não se confunde com o modelo de linguagem. O mesmo substantivo pode significar coisas diferentes em contextos diferentes. “Produto” pode ser uma unidade de gestão de stock em Inventário, uma linha com preço em Faturação e um compromisso de entrega na Gestão de Encomendas.

A figura compara uma única classe Product partilhada com um modelo Product por contexto. Na metade inferior, cada seta é um mapeamento que o código executa quando uma linha de encomenda passa para outro contexto.

Cada contexto é dono do seu domain model. Quando a informação passa para outro contexto, o código mapeia os termos e os identificadores entre eles. Fowler descreve esta separação como uma forma de impedir que um único modelo abranja todos os significados de um termo (bounded contexts). Não é automaticamente um microsserviço, um repositório, um agente ou uma equipa, embora essas fronteiras muitas vezes coincidam.

No design de agentes, isto significa que:

• um contexto pode usar internamente várias chamadas ao modelo ou agentes especializados
• um agente que abrange vários contextos precisa de tradução explícita e de autoridade para cada um
• a orquestração é uma questão da aplicação; não apaga a responsabilidade do domínio

A figura seguinte desenha estes três casos. Compare a segunda e a terceira colunas: no segundo, um agente traduz termos para dois contextos; no terceiro, um orquestrador limita-se a enviar comandos, e cada contexto aplica as suas próprias regras.

Antes de desenhar um grafo de agentes, comece por um context map que mostre as áreas de negócio e a forma como trocam informação. Caso contrário, o grafo tende a reproduzir as tools disponíveis em vez do negócio.

Classificar os subdomínios

O DDD costuma separar:

• Core domain: a capacidade que cria valor diferenciador
• Supporting subdomain: trabalho necessário e específico do negócio que não é o diferenciador
• Generic subdomain: uma capacidade já resolvida, como a identidade ou o envio de e-mail

Num assistente de tarefas, a gestão de tarefas pode ser o core, o agendamento um supporting e o envio de notificações um generic. O mapa coloca cada contexto na faixa do seu subdomínio. O Agendamento fornece datas à Gestão de Tarefas, que comunica cada tarefa adicionada às Notificações através de um evento TaskAdded.

Dedique o esforço de modelação ao core domain e escreva as suas regras internamente. Use um serviço existente para os generic subdomains, como um fornecedor de e-mail para as notificações. A classificação não decide onde entra um LLM.

Os tactical patterns definem a fronteira de estado

Entities e value objects

Uma entity tem identidade e um ciclo de vida. Uma tarefa continua a ser a mesma tarefa depois de a sua descrição mudar. Um value object é definido pelos seus valores e é normalmente imutável: um endereço de e-mail, um montante em dinheiro ou um intervalo de tempo.

O estado de um agente encaixa na mesma divisão. Uma ConversationSession com o seu histórico de mensagens é uma entity, tal como cada Task que o agente planeia. Um tool result registado, como um ToolOutput com os argumentos da chamada e a resposta, é um value object: não muda depois de guardado. Quando uma Task é um objeto, pode recusar a conclusão até as suas dependências terminarem, e essa regra fica num único método em vez de em vários templates de prompt.

Aggregates e invariants

Um aggregate agrupa objetos relacionados cujas alterações têm de obedecer às mesmas regras (Evans, Domain-Driven Design Reference). O restante código altera esses objetos através de um único objeto de entrada, chamado aggregate root.

No exemplo abaixo, TaskList é a root e os seus objetos Task são membros. Uma invariant é uma regra que tem de continuar verdadeira depois de cada operação. Esta lista rejeita uma tarefa com a mesma descrição normalizada e a mesma data de vencimento que uma tarefa existente. Outros domínios têm invariants diferentes:

• uma tarefa concluída não pode ser concluída outra vez
• um dono não pode ter lembretes abertos duplicados para o mesmo dia
• um reembolso não pode exceder o montante reembolsável restante
• um plano aceita no máximo 10 tarefas abertas, mesmo que o modelo proponha 50

Um aggregate não se torna seguro só porque uma lista Python está atrás de um método add_task(). O código externo não deve receber uma referência mutável que contorne esse método. A persistência também precisa de controlo de concorrência, ou dois pedidos válidos podem violar uma invariant quando guardados em simultâneo.

Repositories e application services

Um repository carrega e guarda aggregates sem deixar que questões da base de dados passem para o domínio (Fowler, Repository). Um application service coordena um use case: carrega o estado, invoca a operação do domínio, guarda com uma versão esperada e publica os eventos resultantes.

Uma unit of work agrupa as gravações de um use case numa única transação da base de dados: confirma todas ou nenhuma (Percival e Gregory, Unit of Work). O application service abre-a, faz as suas alterações e chama commit(). Sair sem chamar commit() descarta as alterações ainda não confirmadas.

O domínio não deve chamar um LLM, um cliente HTTP nem um ORM. Isso pertence aos adapters, o código que liga a aplicação a sistemas externos.

Os domain events são factos, não um message bus

TaskAdded regista um facto: uma tarefa foi adicionada. A aplicação pode guardar este evento numa outbox table, na mesma transação da base de dados que a alteração da tarefa. A transação guarda as duas alterações em conjunto ou nenhuma. Um publisher separado envia o evento guardado para outros serviços só depois de a transação ter êxito. Este é o transactional-outbox pattern; um objeto de evento, por si só, não o fornece (Richardson, Transactional Outbox). Enviar diretamente para um broker a partir de uma entity arrisca publicar um evento de uma transação que depois falha.

A figura compara três execuções: a entidade envia o evento ao broker e a transação faz depois rollback; o outbox com rollback; e o outbox com commit.

Os eventos podem coordenar agentes, mas por si só não tornam a coordenação fiável. A semântica de entrega, a idempotência, a ordenação e os contratos versionados continuam a ser trabalho de infraestrutura.

Escolha o destino do publisher consoante quem consome o evento. Quando os listeners correm no mesmo processo, uma biblioteca de sinais em processo, como o blinker, pode chamá-los depois do commit. Quando outros serviços consomem o evento, envie-o para um broker como o RabbitMQ, ou para Redis Streams se a equipa já usa Redis. Não use Redis Pub/Sub para estes eventos: entrega cada mensagem no máximo uma vez, e um subscritor desligado perde-a (semântica de entrega do Redis Pub/Sub).

Trate o output do modelo como uma proposta não fiável

Uma integração com um LLM assemelha-se a uma anti-corruption layer: traduz uma representação externa e probabilística para termos que o domínio compreende. A camada de tradução verifica apenas a forma e os valores da proposta. A permissão fica no application service e as regras de negócio ficam no aggregate.

Um port é uma interface que lista as operações de que a aplicação precisa. O repository port lista como carregar e guardar listas de tarefas; um adapter de base de dados implementa essas operações. A figura desenha os quatro pacotes como anéis e mostra dependências de código, não a ordem de execução. A aplicação usa o aggregate e o repository port; o adapter de infraestrutura implementa esse port. O domínio não depende nem do adapter de base de dados nem do modelo.

Num projeto Python, as quatro camadas podem ser quatro pacotes: domain/, application/, infrastructure/ e interface/. Trocar a base de dados passa então a mexer em infrastructure/; mudar o fornecedor do modelo mexe em interface/.

O diagrama seguinte percorre as quatro chamadas do exemplo prático abaixo através dos cinco passos da lista que vem a seguir. Cada linha mostra onde uma chamada pára e o que a tool devolve.

O código entre o modelo e o domínio executa estes passos:

1. Receber a proposta: o modelo devolve AddTaskProposal; trate todos os valores como não fiáveis.
2. Analisar e normalizar: valide-a contra o contrato tipado e depois resolva datas, unidades, identificadores e locale com contexto fiável.
3. Autorizar: decida se este ator pode pedir a operação.
4. Impor a invariant: invoque um método do aggregate que verifica a regra e emite TaskAdded.
5. Confirmar em conjunto: guarde a lista de tarefas e o evento da outbox numa única transação.

Três ferramentas podem fazer o modelo devolver o output tipado e validá-lo (passos 1 e 2):

• Pydantic AI, quando quer um framework de agentes. Defina o tipo de output do agente como um modelo Pydantic; quando o output falha a validação, envia o erro de volta ao modelo e pede novamente.
• Instructor, quando mantém o SDK do seu fornecedor e só precisa de objetos Pydantic já analisados. Envolve o cliente, e max_retries volta a pedir ao modelo com o erro de validação.
• O modo de structured output do próprio fornecedor, como os OpenAI Structured Outputs, quando quer que o fornecedor restrinja a geração ao JSON Schema.

Envie um erro de validação de volta ao modelo uma ou duas vezes. Se a proposta continuar a falhar, pergunte ao utilizador em vez de repetir.

O Pydantic consegue rejeitar um campo em falta ou uma prioridade fora de low, normal, high através do seu modelo de validação tipada (documentação do Pydantic). Não consegue decidir que “amanhã” corresponde à data correta, que o utilizador é dono da lista de tarefas, ou que já existe uma tarefa semelhante em aberto.

Exemplo prático: adicionar uma tarefa

Considere um pedido para adicionar “Send the invoice” a uma lista de tarefas. O actor é o requerente autenticado (actor_id); o owner é a pessoa cuja lista vai mudar (owner_id). Podem ser pessoas diferentes, por isso a aplicação verifica a permissão desse actor e dessa lista antes de fazer a alteração.

O adapter do modelo fornece os detalhes da tarefa como AddTaskProposal. O application service verifica a permissão, pede a TaskList que adicione a tarefa e confirma a alteração e o respetivo evento TaskAdded através de uma unit of work. Guarde os sete blocos Python desde esta secção até «Executar o exemplo», por ordem, como task_example.py. Formam um módulo para Python 3.12 com Pydantic 2; o último bloco executa-o.

1. Definir a proposta orientada para o modelo

Mantenha a proposta próxima do que o modelo consegue inferir. Não lhe peça que invente IDs da base de dados nem identificadores fiáveis de owner.

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

Se o utilizador disser “amanhã”, a aplicação deve dar ao modelo uma data local explícita ou resolver a expressão relativa com um analisador de datas testado. Nunca use o relógio do servidor de inferência como contexto de negócio implícito.

2. Colocar a invariant no aggregate

TaskList.add_task() rejeita descrições em branco e tarefas duplicadas antes de alterar a lista. Manter estas verificações aqui faz com que se apliquem quer o pedido venha de um LLM quer de um formulário 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

O domínio lança os seus próprios tipos de exceção. pydantic.ValidationError é uma subclasse de ValueError, por isso uma tool que apanhasse ValueError para reportar um conflito de domínio apanharia também propostas inválidas.

TaskAdded é uma dataclass imutável (frozen): um value object que regista o facto. O aggregate guarda as tarefas num tuple imutável, por isso os chamadores não recebem nenhuma coleção mutável a que possam acrescentar itens ou que possam editar contornando add_task(). Os seus métodos de alteração substituem esse tuple de suporte só depois de impor a regra.

A deteção de duplicados é aqui deliberadamente simples. As regras reais podem exigir normalização sensível ao locale, semântica de recorrência ou uma restrição de unicidade na base de dados como última salvaguarda contra condições de corrida.

3. Definir os ports

Estes protocolos Python listam o que a aplicação precisa do armazenamento e das verificações de permissão. TaskListRepository carrega e guarda uma lista de tarefas. UnitOfWork contém esse repository e os eventos pendentes, e confirma ambos. TaskAuthorizer lança Unauthorized quando o actor não pode alterar a 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: ...

O adapter de base de dados só deve guardar se a versão armazenada ainda corresponder a expected_version, a versão carregada antes. Uma gravação bem-sucedida incrementa essa versão. Se outro pedido tiver alterado a lista, lança ConcurrentUpdate em vez de sobrescrever o trabalho desse pedido. Isto é optimistic concurrency control; a documentação do contador de versão do SQLAlchemy mostra a comparação subjacente. O protocolo enuncia o requisito; um adapter concreto tem de implementar a verificação e a atualização em conjunto.

4. Coordenar o use case

AddTaskService põe por ordem a permissão, o carregamento, a operação do domínio e a persistência. Os identificadores de actor e de owner vêm do contexto fiável da aplicação, não da proposta do 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

Se uma linha antes de commit() lançar uma exceção, a unit of work descarta a alteração da lista de tarefas e o evento ainda não confirmados. Uma falha de ligação ou um timeout durante commit() pode deixar o resultado desconhecido: a base de dados pode ter confirmado a transação antes de a resposta se perder. Sair do bloco não pode desfazer essa transação já confirmada (Amazon Builders’ Library, Making retries safe with idempotent APIs).

O modelo está ausente deste serviço. Um adapter pode obter AddTaskProposal de um LLM, outro de um formulário HTTP, e os testes podem construí-lo diretamente. O comportamento de negócio mantém-se idêntico.

5. Adicionar adapters em memória

Estes adapters permitem executar o exemplo sem base de dados. InMemoryUnitOfWork.commit() verifica primeiro todas as versões e só depois escreve as listas de tarefas e os eventos, por isso guarda ambos ou nenhum. AllowList é um authorizer falso para testes e demonstrações.

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

Um adapter de base de dados implementa UnitOfWork com uma única transação SQL; a versão em memória mostra o contrato.

Mapear tools para comandos da aplicação

As tools de um agente devem expor use cases, não primitivas de base de dados. Prefira:

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

em vez de:

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

O primeiro conjunto fala a linguagem do domínio e dá à aplicação um sítio onde autorizar e impor regras. O segundo deixa o modelo descrever mutações de persistência arbitrárias. A figura envia o mesmo pedido duplicado pelos dois designs: a tool de base de dados guarda uma segunda cópia e a tool de comandos de domínio recusa-a.

O código de um framework de agentes, como um nó do LangGraph ou uma tool do Pydantic AI, pertence à camada de interface e chama application services, como faz add_task_tool() abaixo. Se a orquestração na camada de aplicação usar um framework, coloque-o atrás de um port que a aplicação controla. O domínio nunca importa o framework.

Um tool result deve distinguir as falhas sobre as quais o agente pode agir. Não reduza todas as falhas a uma string que convida a repetições às cegas. Cada falha tem uma ação seguinte diferente:

Aqui, unavailable indica uma exceção de infraestrutura; não prova que nada foi guardado. Repita com backoff apenas depois de a aplicação confirmar que não ocorreu nenhuma escrita, ou quando guardar atomicamente a escrita e o seu resultado sob um ID de pedido e devolver esse resultado nas repetições. Reutilize o mesmo ID e os mesmos argumentos durante o respetivo período de retenção (Amazon Builders’ Library, Making retries safe with idempotent APIs). Caso contrário, determine o resultado a partir dos registos da aplicação. Se esses registos não o permitirem determinar, informe o utilizador de que o resultado é desconhecido.

A demonstração rejeita tarefas com conteúdo duplicado, mas não implementa a idempotência dos pedidos: uma repetição após uma escrita confirmada devolve domain_conflict em vez do sucesso original.

Esta função de tool envolve AddTaskService.execute() e devolve esses códigos. O framework passa o actor e o owner a partir da sessão autenticada, não a partir do 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)}

Executar o exemplo

O último bloco chama a tool quatro vezes: uma tarefa válida, um duplicado com maiúsculas e espaçamento diferentes, uma prioridade fora dos valores permitidos e um actor sem permissão.

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

Com o Python 3.12 instalado, execute estes comandos no diretório que contém task_example.py:

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

O módulo imprime:

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

Só a primeira chamada alterou o estado. A lista guardada está na versão 1 com uma tarefa, e a outbox contém um evento TaskAdded.

Testes para cada camada

Testes do domínio

Teste os aggregates com pytest, sem modelo, rede ou base de dados:

• as tarefas duplicadas são rejeitadas
• as tarefas válidas emitem o evento esperado
• as coleções expostas não conseguem alterar o estado interno
• as regras de transição mantêm-se ao longo de operações repetidas

A segunda chamada da execução acima já verifica o primeiro ponto através da tool. Um teste de domínio faz a mesma chamada diretamente em TaskList e espera DuplicateTask.

Testes da aplicação

Use repositories e authorizers falsos para verificar o carregamento, a ordem da autorização, as gravações com versão esperada, o comportamento da outbox e o mapeamento de erros. A unit of work em memória acima é esse falso. Guarde o bloco seguinte separadamente como test_task_example.py, junto de task_example.py. Abre duas units of work sobre os mesmos dados para simular dois pedidos que carregam a mesma versão:

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

Instale pytest no mesmo ambiente e execute o teste:

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

Avaliações do contrato do modelo

Avalie o adapter probabilístico em separado:

• a precisão na extração da intenção e dos campos
• a resolução de datas relativas com o contexto de fuso horário fornecido
• a recusa ou o pedido de esclarecimento quando falta informação obrigatória
• a resistência a prompt injection dentro do texto citado de uma tarefa
• a taxa de propostas válidas segundo o schema mas semanticamente inutilizáveis

Para testes que não devem chamar um modelo real, o Pydantic AI fornece TestModel, que chama as tools do agente e devolve output válido segundo o schema, e FunctionModel, que deixa o teste decidir cada resposta (Pydantic AI testing). Reserve as avaliações acima para o modelo real.

Um teste de ponta a ponta deve então confirmar que propostas más nunca contornam os mesmos métodos de domínio usados pelas interfaces fiáveis.

Ordem de trabalho para um novo projeto de agentes

1. Entreviste os especialistas do domínio e escreva o glossário, com uma definição e um exemplo para cada termo.
2. Desenhe o context map. Comece com um único contexto core.
3. Modele as entities e os value objects. Ponha cada regra no método que altera o estado.
4. Escolha as aggregate roots e escreva as suas invariants.
5. Defina os ports de repository, unit of work e authorizer. Use adapters em memória até os testes de domínio passarem.
6. Emita domain events para as alterações de que outros contextos precisam, e escreva-os numa outbox na mesma transação.
7. Defina os tipos de proposta orientados para o modelo e o parser que os valida.
8. Exponha cada use case como uma tool que chama um application service e mapeia as suas falhas.
9. Acrescente a orquestração por último, enviando comandos entre contextos.

Quando o design está a funcionar

Deve ser possível mudar o fornecedor do modelo sem mudar um teste de domínio. Uma alteração de política deve mexer num aggregate ou num domain service em vez de vários prompts. Os traces devem usar termos que a equipa responsável reconhece. As propostas malformadas ou não autorizadas devem falhar antes da persistência, e uma gravação concorrente deve falhar em vez de sobrescrever o estado em silêncio.

O modelo continuará a fazer propostas erradas; o aggregate rejeita as que violam uma regra.

Referências

• Eric Evans, Domain-Driven Design Reference — definições dos padrões estratégicos e táticos
• Martin Fowler, Bounded Context — porque é que um modelo não deve abranger todos os significados de um termo
• Martin Fowler, Repository — abstração de coleção para a persistência
• Harry Percival e Bob Gregory, Architecture Patterns with Python: Unit of Work — o padrão unit of work em Python
• Chris Richardson, Transactional Outbox — publicar depois de uma transação da base de dados sem perder eventos
• Pydantic documentation — validação tipada da proposta
• Pydantic AI output — structured output com repetições de validação
• Instructor retrying — output Pydantic validado sobre o SDK de um fornecedor