Uma demo de agente de atendimento fica pronta em uma tarde: mande a mensagem do cliente para um modelo de linguagem com um bom prompt e publique a resposta. Um agente que roda sobre tickets reais, com pedidos reais, dinheiro real e clientes que já chegam chateados, tem um formato bem diferente.

Este artigo percorre esse formato, passo a passo, em Python. Ele é baseado no AI Central, a plataforma de atendimento que desenhei para um grande varejista online. Construí a prova de conceito e depois liderei os desenvolvedores que a levaram para produção. Ela responde tickets como “onde está meu pedido?”, atrasos de entrega, pacotes avariados e problemas técnicos, lendo dados reais dos pedidos e seguindo uma base de conhecimento curada. Desde abril de 2026, ela processou mais de 61 mil tickets e enviou mais de 110 mil respostas automáticas.

O código abaixo foi simplificado e reescrito para este artigo, mas a arquitetura, os números e as lições vêm do sistema real.

O formato do sistema

A decisão mais importante é o que o modelo não pode fazer. No AI Central, o modelo não decide quais sistemas chamar, não fica em loop até se dar por satisfeito e não envia nada por conta própria. Ele trabalha dentro de um fluxo determinístico:

webhook do helpdesk ──▶ FastAPI ──▶ SQS ──▶ worker
                                               │
                         ┌─────────────────────┘
                         ▼
          pré-checagem de segurança ─▶ coleta do pedido (ERP)
                         │
                         ▼
          classificação ─▶ política (pgvector) ─▶ geração
                         │
                         ▼
          resposta · pergunta ao cliente · encaminhamento humano

Cada seta é código comum. O modelo é chamado em poucos pontos bem definidos: para classificar, para verificar abusos e para escrever a resposta. Isso torna o sistema testável, auditável e muito mais barato de depurar.

Passo 1: aceite webhooks rápido, processe depois

Ferramentas de helpdesk entregam eventos por webhook e tentam de novo quando não recebem resposta rápida. Processar um ticket pode levar muitos segundos (chamadas ao ERP, recuperação de contexto, uma ou mais chamadas ao modelo), então o handler do webhook deve apenas validar, enfileirar e devolver 202 Accepted.

from fastapi import FastAPI, Depends, status
from pydantic import BaseModel

app = FastAPI()


class TicketEvent(BaseModel):
    ticket_id: int
    group_id: int
    kind: str  # "new" ou "customer_reply"


@app.post("/webhooks/tickets", status_code=status.HTTP_202_ACCEPTED)
async def receive(event: TicketEvent, _: None = Depends(verify_token)):
    await queue.send(event.model_dump())
    return {"queued": True}

Dois detalhes fizeram diferença na prática. Primeiro, autentique o webhook com um token guardado em configuração, nunca no código. Segundo, seja tolerante com o payload: automações de helpdesk nem sempre produzem JSON válido, e por isso o parser do AI Central aceita JSON, form data e query string.

Passo 2: um worker com retentativas e fila de mensagens mortas

O worker faz long polling na fila e processa vários tickets em paralelo. O AI Central usa SQS com long polling de 20 segundos, visibility timeout de 900 segundos e concorrência padrão de quatro.

import asyncio

semaphore = asyncio.Semaphore(4)


async def worker_loop():
    while True:
        messages = await queue.receive(max_messages=4, wait_seconds=20)
        await asyncio.gather(*(handle(m) for m in messages))


async def handle(message):
    async with semaphore:
        try:
            await process_ticket(message.body)
            await queue.delete(message)
        except IntegrationError:
            await retry_later(message)       # transitório: ERP ou helpdesk falhou
        except Exception:
            await send_to_dead_letter(message)

A política de retentativa é propositalmente estreita. Só falhas de integração, como um timeout do ERP ou um 5xx do helpdesk, são repetidas: até três vezes, com espera de 2 ** n segundos limitada a 900. Todo o resto vai para uma fila de mensagens mortas (DLQ) e para uma tabela failed_tasks, porque repetir um bug três vezes só produz três falhas.

Limites de requisição são uma categoria à parte. A API do helpdesk permite um número limitado de chamadas por minuto, então o AI Central mantém um limitador de janela deslizante abaixo dessa cota e manda tickets limitados para a DLQ com a marcação rate_limit. Um consumidor pequeno os devolve à fila depois do período de Retry-After, com um pouco de atraso aleatório para que não voltem todos ao mesmo tempo.

Passo 3: um ticket por vez, e nunca a mesma resposta duas vezes

Clientes costumam mandar duas mensagens seguidas. Sem proteção, dois workers podem processar o mesmo ticket simultaneamente e os dois responderem.

O AI Central se protege disso em dois níveis:

_processing: set[int] = set()
_lock = asyncio.Lock()


async def run_exclusively(ticket_id: int, fn) -> bool:
    async with _lock:
        if ticket_id in _processing:
            return False
        _processing.add(ticket_id)
    try:
        await fn()
        return True
    finally:
        async with _lock:
            _processing.discard(ticket_id)

Esse lock por ticket vive dentro do processo do worker. É suficiente para um único worker. Com vários processos, a mesma ideia precisa de um lock compartilhado no Redis ou no banco, e essa é uma das primeiras melhorias que eu faria.

A segunda proteção fica no final, logo antes do envio: uma chave de idempotência derivada do texto da resposta.

import hashlib


def should_send(redis, ticket_id: int, text: str) -> bool:
    normalized = " ".join(text.split())
    digest = hashlib.sha256(normalized.encode()).hexdigest()[:16]
    return bool(redis.set(f"ticket_sent:{ticket_id}:{digest}", "1", nx=True, ex=120))

Se a mesma resposta já foi enviada para o mesmo ticket nos últimos dois minutos, ela não é enviada de novo. O agente também verifica se a última mensagem pública veio do cliente, para nunca atropelar um atendente humano que respondeu nesse meio-tempo.

Passo 4: modele a conversa como uma máquina de estados

Esta foi a decisão que tornou o sistema administrável. Em vez de um loop de agente aberto, cada ticket tem um estado explícito, guardado no banco:

from enum import StrEnum


class TicketStep(StrEnum):
    NEW_TICKET = "new_ticket"
    COLLECTING_DATA = "collecting_data"
    AWAITING_ORDER_NUMBER = "awaiting_order_number"
    AWAITING_DAMAGE_PHOTOS = "awaiting_damage_photos"
    AWAITING_NEW_ADDRESS = "awaiting_new_address"
    WAITING_CUSTOMER = "waiting_customer"
    COMPLETED = "completed"
    REDIRECTED = "redirected"   # encaminhado para um humano

No AI Central, a maioria dos estados é do tipo awaiting_*: cada um é uma pergunta específica que o agente fez e cuja resposta está aguardando. Os demais cobrem tickets novos, coleta de dados, geração de resposta, espera pelo cliente e as duas formas de um ticket terminar.

Mensagens novas passam por um pipeline linear de fases. Cada fase deixa o ticket seguir ou interrompe a execução devolvendo o próximo estado:

from collections.abc import Awaitable, Callable

Phase = Callable[[TicketContext], Awaitable[TicketStep | None]]

PIPELINE: list[Phase] = [
    security_precheck,
    extract_order_identifiers,
    load_order_from_erp,
    match_requester_to_order,
    categorize,
    retrieve_policy,
    handle_special_flows,   # ex.: extravio, entrega parcial
    generate_reply,
]


async def run_pipeline(ctx: TicketContext) -> TicketStep:
    for phase in PIPELINE:
        next_step = await phase(ctx)
        await record_event(ctx, phase.__name__, next_step)
        if next_step is not None:
            return next_step
    return TicketStep.WAITING_CUSTOMER

O AI Central monta esse pipeline com LangGraph: 18 nós ligados por arestas condicionais que param quando uma fase devolve um estado. A ideia, porém, não depende de framework. Uma lista de funções assíncronas resolve.

Respostas a um estado awaiting_* pulam o pipeline e vão direto para o handler daquela pergunta. Se o agente pediu fotos de um pacote avariado, a próxima mensagem é verificada primeiro em busca das fotos.

Cada fase também adiciona uma linha a uma tabela ticket_flow_events: a fase, o estado antes e depois e um payload sanitizado (no máximo 35 chaves, textos truncados em 400 caracteres). O estado atual fica em uma linha comum; os eventos formam uma linha do tempo só de inserção. Quando alguém pergunta “por que o agente disse isso?”, a resposta está a uma consulta de distância.

Passo 5: guardrails antes de o modelo ver qualquer coisa

Mensagens de atendimento são entrada não confiável. Antes de qualquer geração, o AI Central executa duas verificações.

A primeira é código simples: truncar a mensagem (5.000 caracteres), remover marcadores de papel e delimitadores que parecem sintaxe de prompt e sinalizar padrões conhecidos de injeção.

A segunda é uma chamada pequena e barata ao modelo, com saída estruturada. Com instructor e um modelo Pydantic, o classificador só pode devolver um dos valores permitidos:

from typing import Literal

import instructor
from openai import AsyncOpenAI
from pydantic import BaseModel

client = instructor.from_openai(AsyncOpenAI())


class MessageCheck(BaseModel):
    verdict: Literal["ok", "attack", "serious_complaint"]
    reason: str


async def precheck(message: str) -> MessageCheck:
    return await client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        response_model=MessageCheck,
        messages=[
            {"role": "system", "content": PRECHECK_PROMPT},
            {"role": "user", "content": message[:4000]},
        ],
    )

Um ataque encerra o ticket com uma tag de segurança. Uma reclamação grave, como ameaça de acionar um órgão de defesa do consumidor, um advogado ou a Justiça, vai direto para um humano. Nenhum modelo deveria negociar com um cliente chateado que menciona um advogado.

Passo 6: busque os fatos com código, não com tool calling

Muitos tutoriais de agentes dão ferramentas ao modelo e deixam que ele decida quando consultar um pedido. O AI Central não faz isso. Consultas de pedido e entrega são fases fixas do pipeline: extrair o número do pedido ou da nota fiscal da mensagem, chamar o ERP (com timeout de 30 segundos e retentativas em 5xx) e transformar o resultado em um resumo de texto compacto para o prompt.

Isso tem três vantagens. As consultas sempre acontecem, e sempre do mesmo jeito. Elas são testáveis sem modelo. E o modelo nunca vê uma ferramenta que poderia chamar com argumentos errados.

Também permite controlar exatamente o que o modelo vê. Uma das decisões de arquitetura do AI Central removeu do contexto os status de rastreio de “disponível para retirada”, porque o modelo inventava endereços de retirada sempre que os via. Às vezes o melhor guardrail é não mostrar o dado.

Passo 7: recuperação de contexto com pgvector

Políticas mudam: o que fazer quando uma entrega atrasa, quando um pacote chega avariado, quando o endereço está errado. O AI Central as mantém em uma base de conhecimento em markdown, dividida em seções, cada uma com um título (uma das categorias de ticket) e uma linha de palavras-chave.

Cada seção vira um embedding com text-embedding-3-small (1.536 dimensões) e é guardada no PostgreSQL com pgvector:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE knowledge (
  id            bigserial PRIMARY KEY,
  section_title text NOT NULL UNIQUE,
  content       text NOT NULL,
  embedding     vector(1536) NOT NULL
);

A busca usa distância de cosseno:

SEARCH = """
SELECT section_title, content, 1 - (embedding <=> :query) AS similarity
FROM knowledge
WHERE (:category IS NULL OR lower(section_title) = lower(:category))
  AND 1 - (embedding <=> :query) >= :min_similarity
ORDER BY embedding <=> :query
LIMIT :limit
"""

A parte interessante é a recuperação em duas passadas:

async def retrieve(query_vec, category):
    first = await search(query_vec, category=category, limit=2, min_similarity=0.6)
    if first and first[0].similarity >= 0.70:
        return first[0], category

    fallback = await search(query_vec, category=None, limit=2, min_similarity=0.6)
    if fallback and fallback[0].section_title in VALID_CATEGORIES:
        # A recuperação discorda do classificador: confie na base de conhecimento.
        return fallback[0], fallback[0].section_title
    return (fallback[0] if fallback else None), category

A primeira passada busca só na seção da categoria escolhida pelo classificador. Se o melhor resultado for fraco (abaixo de 0,70), a segunda passada busca na base inteira. Se encontrar uma seção forte de outra categoria, o ticket é reclassificado. A recuperação vira uma segunda opinião sobre o roteamento.

O texto da busca é montado com a última mensagem do cliente (até 500 caracteres), o assunto e o começo da descrição; em respostas seguintes, só a última mensagem.

Uma observação sobre escala: a base de conhecimento é pequena, então o AI Central não usa índice aproximado (HNSW ou IVFFlat). Uma leitura sequencial em algumas dezenas de vetores é instantânea. Crie o índice quando a tabela crescer, não antes.

Isso substituiu uma tentativa anterior de fine-tuning. Por cerca de um dia, o projeto teve um pipeline de fine-tuning com um conjunto de treino de 300 exemplos. Trocar para recuperação de contexto significou que as políticas passaram a ser atualizadas editando markdown, que cada resposta pode ser rastreada até uma seção e que nada precisa ser retreinado.

Passo 8: deixe o modelo escrever, mas não decidir

A resposta é gerada com o resumo do pedido, a política recuperada e a conversa recente no prompt. O AI Central usa gpt-4o-mini como modelo principal.

Às vezes a resposta implica uma ação: iniciar um fluxo de extravio, pedir um novo endereço, encaminhar para uma pessoa. O modelo só pode sinalizar isso por um bloco pequeno e fixo no fim da saída, que o código remove e valida contra uma lista de permissões:

VALID_WORKFLOWS = {
    "lost_shipment",
    "address_change",
    "damaged_package",
    "human_handoff",
    # ...
}


def extract_workflow(text: str) -> tuple[str, str | None]:
    reply_lines, workflow = [], None
    for line in text.splitlines():
        if line.lower().startswith("workflow:"):
            slug = line.split(":", 1)[1].strip().strip("'\"").lower()
            if slug in VALID_WORKFLOWS:
                workflow = slug
            continue
        reply_lines.append(line)
    return "\n".join(reply_lines).strip(), workflow

Qualquer coisa fora da lista é ignorada, e o prompt orienta o modelo a omitir o bloco na dúvida. Um marcador especial na saída significa “escalar para um humano agora”, e ele sempre vence.

Passo 9: mantenha custo e disponibilidade sob controle

Duas técnicas fizeram a maior diferença.

Um cache de planos de resposta, não de respostas. Guardar respostas literais não funciona em atendimento, porque cada cliente, pedido e data é diferente. O AI Central guarda blueprints: um plano reutilizável e sem dados pessoais para um tipo de resposta (intenção, objetivo, até oito passos, campos dinâmicos, regras especiais).

SELECT blueprint, 1 - (embedding <=> :query) AS similarity
FROM response_blueprints
WHERE lower(category) = lower(:category)
  AND playbook_version = :version
  AND 1 - (embedding <=> :query) >= 0.86
ORDER BY embedding <=> :query
LIMIT 1;

Quando há acerto no cache, o prompt é montado de forma compacta: o blueprint substitui o texto bruto da política, o histórico da conversa cai de cinco mensagens para duas e o orçamento de saída cai de 1.000 para 600 tokens. Quando não há, um blueprint novo é gerado e guardado. Incrementar playbook_version invalida todos os planos quando as políticas mudam. No momento em que escrevo, o cache guarda 2.298 planos e já serviu 1.985 acertos.

Um provedor reserva. APIs de modelos falham. O AI Central tenta o provedor principal três vezes com backoff exponencial e então recorre ao Claude, da Anthropic:

async def complete(messages, **kwargs) -> str:
    last_error = None
    for attempt in range(3):
        try:
            response = await openai.chat.completions.create(messages=messages, **kwargs)
            return response.choices[0].message.content
        except Exception as error:
            last_error = error
            await asyncio.sleep(2 ** attempt)
    return await anthropic_fallback(messages, reason=str(last_error), **kwargs)

Um detalhe: se a saída falha nas verificações de conteúdo, isso não é problema de disponibilidade, então o código levanta um erro em vez de recorrer ao outro provedor. Pedir para um segundo modelo tentar de novo não conserta uma resposta suspeita.

Erros do modelo também têm tratamento próprio. Um timeout, uma requisição recusada e uma resposta impossível de interpretar levam cada um a um desfecho definido (nova tentativa, fallback ou encaminhamento), em vez de uma exceção genérica que poderia deixar o cliente sem resposta.

Cada chamada registra o uso de tokens e o custo estimado, e cada trace no LangSmith carrega o id do ticket. Quando a conta ou o comportamento mudam, dá para descobrir por quê.

Passo 10: saiba quando parar

Um agente que nunca desiste é pior do que nenhum agente. O AI Central tem limites explícitos:

  • Um número pequeno e fixo de respostas automáticas por ticket.
  • Um número máximo de tentativas para cada pergunta que o agente faz antes de encaminhar.
  • Encaminhamento imediato em reclamações graves, em ambiguidade sobre o que o cliente escolheu ou quando o pedido existe mas não tem entrega.
  • Tickets parados são resolvidos automaticamente depois de alguns dias.

Encaminhar bem é uma funcionalidade. O atendente humano recebe o ticket com os dados coletados e a linha do tempo completa do que o agente fez.

De um agente para uma plataforma

O AI Central começou com uma única área de atendimento. Quando chegou a segunda (suporte técnico, com uma conversa bem diferente), copiar o pipeline teria criado dois sistemas se afastando um do outro. Em vez disso, o código foi dividido em um núcleo e módulos.

O núcleo cuida de tudo que não é regra de domínio: webhooks, fila, retentativas, clientes externos, persistência e locks. Cada módulo tem seu fluxo, seus prompts e sua base de conhecimento, atrás de um contrato pequeno:

from typing import Protocol


class ModuleProcessor(Protocol):
    name: str

    async def process_new(self, ctx: TicketContext) -> None:
        """Primeira execução depois que o ticket é gravado."""

    async def process_reply(self, ctx: TicketContext) -> None:
        """Uma resposta do cliente em um ticket existente."""


MODULE_PROCESSORS: dict[str, ModuleProcessor] = {
    "delivery": DeliveryProcessor(),
    "tech_support": TechSupportProcessor(),
}

Um ticket novo é roteado para um módulo uma única vez, e o nome do módulo fica gravado nele, para que as respostas seguintes voltem ao mesmo fluxo mesmo que o ticket mude de equipe. Os módulos não precisam ser parecidos: o de entregas é um pipeline em LangGraph, e o de suporte técnico é um fluxo menor, escrito à mão, com um handler por etapa. A única regra é que nenhum módulo mexe em outro.

Testes e avaliação

As partes determinísticas são testadas como qualquer outro código. O AI Central tem cerca de 250 testes unitários cobrindo o parsing do bloco de roteamento, transições de estado, regras de prompt, extratores e validadores do ERP. Nenhum deles chama um modelo.

Para as partes que dependem do modelo, a prática mais útil veio de um projeto irmão, um assistente de suporte para dúvidas internas sobre o ERP. Ele usa um conjunto de referência de perguntas reais com respostas esperadas, incluindo casos em que o comportamento correto é não responder. Medir a revocação da recuperação separadamente da abstenção correta, e verificar fidelidade com um LLM como juiz, pega a falha que mais importa em atendimento: uma resposta confiante que não está na política.

Lições

  • Faça do modelo um componente, não o controlador. Uma máquina de estados e um pipeline de funções simples são mais fáceis de entender, testar e auditar do que um loop aberto.
  • Busque fatos com código. Consultas determinísticas vencem tool calling quando o conjunto de consultas é conhecido.
  • Controle o que o modelo vê. Tirar um dado enganoso do contexto resolveu uma alucinação que ajustes de prompt não resolveram.
  • Use a recuperação como segunda opinião. Uma busca em duas passadas pode corrigir o classificador.
  • Guarde planos, não respostas.
  • Desenhe as saídas. Limites de respostas, regras de encaminhamento e entrega idempotente são o que tornam seguro deixar um agente rodando.
  • Estado compartilhado precisa de lock compartilhado. Um lock em processo serve para um worker; ao escalar, ele precisa ir para o Redis ou para o banco.

O modelo é a parte mais visível de um agente de atendimento, e provavelmente a menor. A maior parte do trabalho, e quase tudo que o torna confiável, é engenharia de backend comum em volta dele.