SQLAlchemy assíncrono e Alembic: persistindo o histórico de um agente

Terceiro post da série sobre o Konhecia, meu agente de suporte com RAG local. Já temos a infraestrutura e a API. Falta o estado.

Um agente que não lembra do turno anterior não é um agente — é um formulário caro. Este post é sobre a camada mais “chata” do projeto, que é justamente onde encontrei o bug mais interessante dele.

Primeiro: por que Postgres se já existe um banco vetorial?

Pergunta legítima, e a resposta define a arquitetura inteira.

São dois tipos de dado diferentes, com padrões de acesso opostos:

	Banco vetorial (Qdrant)	Relacional (Postgres)
Guarda	trechos de documentos + embeddings	conversas e mensagens
Consulta	“o que se parece com isto?”	“me dê exatamente a sessão X, em ordem”
Escrita	em lote, na ingestão	constante, a cada turno

Histórico de conversa exige leitura exata, ordenada, com chave estrangeira e integridade transacional. É exatamente o que um relacional faz há décadas.

Forçar isso no banco vetorial seria usar a ferramenta errada — busca por similaridade não tem ordenação cronológica nem garantia transacional. E o inverso também vale: guardar embeddings numa tabela comum sem índice vetorial transforma cada pergunta num full table scan.

Cada banco com o trabalho para o qual foi construído.

Assíncrono de ponta a ponta

Como a API é async, o acesso ao banco também precisa ser — um driver síncrono dentro de uma rota assíncrona bloqueia o event loop inteiro.

Isso significa asyncpg como driver e a URL com o dialeto explícito:

DATABASE_URL=postgresql+asyncpg://rag_user:senha@postgres:5432/rag_support

O +asyncpg não é enfeite: postgresql:// sozinho resolveria para o driver síncrono padrão, e o SQLAlchemy reclamaria na cara — ou pior, funcionaria e travaria o loop.

O engine é criado uma vez por processo:

engine = create_async_engine(
    settings.database_url,
    echo=settings.app_env == "development",
    pool_pre_ping=True,
)

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

Três parâmetros que valem entender:

• echo liga o log de todo SQL emitido. Impagável em desenvolvimento, proibido em produção — despeja parâmetros de query no log, e isso inclui dado sensível.
• pool_pre_ping=True manda um “ping” barato antes de reusar uma conexão do pool. Sem isso, uma conexão morta (banco reiniciado, timeout de rede, container recriado) só é descoberta quando a query falha, na cara do usuário.
• expire_on_commit=False impede que o SQLAlchemy invalide os objetos após o commit. Com o padrão True, acessar mensagem.content depois de um commit dispara um novo SELECT — que num contexto assíncrono explode com erro de I/O em atributo lazy.

Uma sessão por requisição

A sessão é entregue como dependência do FastAPI:

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

O yield é o que faz isso funcionar: o FastAPI entrega a sessão para a rota e, ao final da requisição, retoma o gerador para que o async with feche a sessão — inclusive se a rota levantou exceção.

Note o que essa função não faz: commit. A decisão de confirmar ou desfazer pertence a quem conhece a regra de negócio, não ao encanamento. Um commit automático aqui gravaria estados parciais de operações que deveriam ser atômicas.

O padrão Repository

Em vez de espalhar select() pelas rotas, cada agregado ganha uma classe que encapsula seu acesso a dados:

class MessageRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def add(self, conversation_id, role, content, sources=None) -> Message:
        message = Message(conversation_id=conversation_id, role=role,
                          content=content, sources=sources)
        self._session.add(message)
        await self._session.flush()
        return message

Repare em flush() e não commit(). São coisas distintas, e confundi-las é fonte clássica de bug:

• flush() envia o SQL pendente ao banco e obtém os valores gerados (o id), mas continua na mesma transação;
• commit() encerra a transação e torna tudo definitivo.

O repositório precisa do id para devolver o objeto, mas não tem autoridade para decidir que a operação inteira terminou. Quem chama é que dá o commit — e um rollback no meio desfaz tudo junto.

O bug: duas mensagens com o mesmo timestamp

Aqui está o problema que me custou mais tempo, e que só aparece em produção.

O padrão para carimbar a criação de uma linha é server_default=func.now(). Parece óbvio. Está errado — sutilmente.

No Postgres, NOW() retorna o horário de início da transação, não o do comando. Duas inserções dentro da mesma transação recebem o valor idêntico.

E qual é o caso mais comum aqui? Exatamente esse: num turno de conversa, a pergunta do usuário e a resposta do assistente são gravadas na mesma transação. Ambas ganhavam o mesmo created_at.

Com timestamps iguais, o ORDER BY created_at fica indefinido. O banco pode devolver a resposta antes da pergunta. E aí a janela de histórico injetada no prompt do LLM chega fora de ordem — o modelo lê a conversa embaralhada. O sintoma aparece longe da causa: você culpa o prompt, não o carimbo de data.

A correção é uma função diferente:

# clock_timestamp() retorna o horário real de execução de cada statement,
# preservando a ordem de inserção. now()/CURRENT_TIMESTAMP retorna o horário
# de início da transação — idêntico para inserções na mesma transação.
_now = func.clock_timestamp()

created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=_now)

A lição que levei: quando a ordenação é semântica, ela precisa de garantia explícita. Timestamp é uma ordenação frágil por natureza. Se eu fosse além, a solução robusta seria uma coluna de sequência monotônica por conversa, em vez de depender de relógio.

Vale registrar também: DateTime(timezone=True), sempre. Guardar horário sem fuso é adiar um bug para o dia da mudança de horário ou do primeiro usuário em outro fuso.

Modelagem: o que o schema garante

class Message(Base):
    __tablename__ = "messages"
    __table_args__ = (
        Index("ix_messages_conversation_created", "conversation_id", "created_at"),
    )

    conversation_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("conversations.id", ondelete="CASCADE"),
        nullable=False,
    )
    content: Mapped[str] = mapped_column(Text, nullable=False)
    sources: Mapped[list | None] = mapped_column(JSONB, nullable=True)

Quatro decisões:

UUID como chave primária. Um id sequencial (1, 2, 3) exposto como session_id numa API é um convite: troque o número e leia a conversa do vizinho. UUID não é controle de acesso, mas remove a enumeração trivial.

Índice composto (conversation_id, created_at). A consulta quente do sistema é sempre “as últimas N mensagens desta conversa, em ordem”. O índice composto atende o filtro e a ordenação na mesma estrutura.

ondelete="CASCADE" no banco, não só no ORM. O cascade="all, delete-orphan" do SQLAlchemy só funciona quando a exclusão passa pelo ORM. Um DELETE via psql deixaria mensagens órfãs. A constraint no banco vale para todo mundo — e, de quebra, é o que permite limpar a tabela filha nos testes apagando só a pai.

JSONB para as citações. As fontes de uma resposta (arquivo, página, score) são uma lista de tamanho variável cuja forma pode mudar. Normalizar isso em tabela própria seria rigidez sem retorno. JSONB no Postgres é indexável e consultável — não é um TEXT disfarçado.

Alembic: migrations versionadas

Base.metadata.create_all() cria as tabelas na primeira execução e nunca mais serve para nada. Assim que o schema evolui com dados em produção, você precisa de migrations.

Alembic gera scripts versionados a partir do diff entre os models e o banco:

alembic revision --autogenerate -m "create conversations and messages tables"
alembic upgrade head      # aplica as pendentes
alembic downgrade -1      # volta uma

Cada script tem upgrade() e downgrade(), e um encadeamento explícito (down_revision) que forma o histórico.

Duas regras que aprendi a não quebrar:

Sempre revise o autogenerate. Ele é um bom rascunho, não uma resposta. Erra em renomeações (vê como drop + create, o que apaga dados), em tipos customizados e em constraints nomeadas implicitamente.

Aplique as migrations no start do container, não no build. No entrypoint.sh:

alembic upgrade head
exec uvicorn app.main:app --host 0.0.0.0 --port 8000

No build não existe banco para migrar. No start, o container já pode esperar o service_healthy do Postgres — e o ambiente sobe consistente, sem passo manual.

O que fica

A camada de dados é onde a “chatice” paga: tipos explícitos, constraints no banco, migrations versionadas e uma fronteira clara entre quem lê e quem decide commitar.

E o bug do clock_timestamp() é o lembrete de que o padrão nem sempre é o correto — func.now() é o que todo tutorial mostra, e estava silenciosamente embaralhando o histórico que alimenta o modelo.

No próximo post, a parte que todo mundo acha que é o projeto inteiro: rodar o LLM localmente com Ollama.