Voltar ao blog

Construi Agentes de IA Que Processam 50 Mil Mensagens por Mes - O Que os Tutoriais Nao Contam

Licoes reais de 3 meses rodando agentes de IA em producao: rate limiting, alucinacoes, fallbacks e as partes que ninguem mostra nos tutoriais.

RS
Richard Sakaguchi Solution Architect
Construi Agentes de IA Que Processam 50 Mil Mensagens por Mes - O Que os Tutoriais Nao Contam

Ha tres meses, coloquei um agente de IA em producao. Hoje, ele processa mais de 50.000 mensagens por mes com zero downtime.

Mas aqui esta a verdade: nenhum tutorial me preparou para o que realmente aconteceu.

Todo mundo te mostra o chatbot “hello world” bonitinho. Ninguem mostra o que acontece quando usuarios reais spammam sua API as 3 da manha, ou quando seu LLM decide alucinar dados de clientes.

Essa e a historia real.

A Promessa vs. A Realidade

O que os tutoriais mostram:

# O agente de IA "perfeito"
agent = AIAgent(model="gpt-4")
response = agent.chat("Ola!")
print(response)  # Magica! ✨

Como e em producao:

Arquitetura do Sistema

graph TB
    A[Mensagem Usuario] --> B{Rate Limiter}
    B -->|Permitido| C[Sistema de Fila]
    B -->|Bloqueado| D[429 Response]
    C --> E{Health Check}
    E -->|Saudavel| F[Agente IA]
    E -->|Degradado| G[Handler Fallback]
    F --> H{Validador}
    H -->|Valido| I[Usuario]
    H -->|Alucinacao| J[Retry Logic]
    G --> I
    J --> F

Percebeu a diferenca? Agentes de IA em producao precisam de 6 camadas de protecao que tutoriais nunca mencionam.

As 5 Verdades Duras Sobre Agentes de IA em Producao

1. Rate Limiting Nao e Opcional - e Questao de Sobrevivencia

O jeito dos tutoriais:

# Abordagem YOLO
while True:
    message = get_message()
    response = ai_agent.process(message)

O jeito de producao:

from collections import defaultdict
from datetime import datetime, timedelta

class AdaptiveRateLimiter:
    def __init__(self, base_limit=100):
        self.limits = defaultdict(lambda: {"count": 0, "reset": datetime.now()})
        self.base_limit = base_limit

    def check_limit(self, user_id: str, risk_score: float) -> bool:
        """Rate limiting adaptativo baseado em comportamento"""
        limit_data = self.limits[user_id]

        # Resetar janela
        if datetime.now() > limit_data["reset"]:
            limit_data["count"] = 0
            limit_data["reset"] = datetime.now() + timedelta(hours=1)

        # Ajustar limite baseado em risco
        adjusted_limit = int(self.base_limit * (1 - risk_score))

        if limit_data["count"] >= adjusted_limit:
            return False

        limit_data["count"] += 1
        return True

Dashboard de Rate Limiting

Por que importa: No primeiro mes, bloqueei 2.847 tentativas de abuso. Sem rate limiting, seriam mais de R$ 2.500 desperdicados em chamadas de API.

2. LLMs Alucinam - Sempre Valide a Saida

Essa doeu. Um usuario perguntou o saldo da conta. O agente respondeu com confianca: “Seu saldo e R$ 127.549,32”

Saldo real? R$ 47,15

A correcao:

import re
from typing import Optional

class ResponseValidator:
    def __init__(self):
        # Padroes que NUNCA devem aparecer nas respostas
        self.forbidden_patterns = [
            r'R\$\s*[\d.,]+',  # Valores em reais
            r'\b\d{3}\.\d{3}\.\d{3}-\d{2}\b',  # CPF
            r'\b\d{2}\.\d{3}\.\d{3}/\d{4}-\d{2}\b',  # CNPJ
            r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b',  # Emails
        ]

    def validate(self, response: str, user_context: dict) -> Optional[str]:
        """Valida resposta da IA contra regras de negocio"""

        # Verificar dados alucinados
        for pattern in self.forbidden_patterns:
            if re.search(pattern, response, re.IGNORECASE):
                return None  # Rejeitar resposta

        # Verificar fatos contra banco de dados
        if "saldo" in response.lower():
            claimed_balance = self.extract_balance(response)
            actual_balance = user_context.get("balance")

            if claimed_balance and abs(claimed_balance - actual_balance) > 0.01:
                return None  # Alucinacao detectada

        return response

Resultado: Zero incidentes de dados financeiros alucinados em producao.

3. Gerenciar Context Window e Uma Arte

Visualizacao de Context Window

Aqui esta o que ninguem te conta: gerenciar contexto de conversacao em escala e mais dificil que construir o agente.

from collections import deque
from dataclasses import dataclass
from typing import List

@dataclass
class Message:
    role: str
    content: str
    tokens: int
    importance: float  # Score 0-1

class SmartContextManager:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.messages = deque()

    def add_message(self, message: Message):
        self.messages.append(message)
        self._trim_context()

    def _trim_context(self):
        """Manter mensagens mais importantes dentro do limite"""
        total_tokens = sum(m.tokens for m in self.messages)

        if total_tokens <= self.max_tokens:
            return

        # Ordenar por importancia, manter system prompts
        sorted_msgs = sorted(
            [m for m in self.messages if m.role != "system"],
            key=lambda x: x.importance
        )

        # Remover menos importantes ate caber
        while total_tokens > self.max_tokens and sorted_msgs:
            removed = sorted_msgs.pop(0)
            self.messages.remove(removed)
            total_tokens -= removed.tokens

Isso me economizou ~R$ 6.000/mes em custos de API ao podar inteligentemente o historico de conversas.

4. Monitoramento Precisa Ser Obsessivo

Metricas que realmente importam:

pie title "O Que Quebra Agentes de IA em Producao"
    "Abuso Rate Limit" : 35
    "Timeouts LLM" : 25
    "Alucinacoes" : 20
    "Problemas Rede" : 15
    "Locks Database" : 5

Meu stack de monitoramento:

from dataclasses import dataclass
from datetime import datetime
import logging

@dataclass
class AgentMetrics:
    timestamp: datetime
    response_time_ms: float
    tokens_used: int
    cost_brl: float
    user_satisfaction: float
    error_type: Optional[str]

    def log(self):
        logging.info(
            f"agent_response",
            extra={
                "duration_ms": self.response_time_ms,
                "tokens": self.tokens_used,
                "cost": self.cost_brl,
                "satisfaction": self.user_satisfaction,
                "error": self.error_type
            }
        )

class AgentMonitor:
    def __init__(self):
        self.metrics = []
        self.alerts = {
            "high_latency": 2000,  # ms
            "low_satisfaction": 0.6,  # 0-1
            "error_rate": 0.05  # 5%
        }

    async def track_request(self, request_fn):
        start = datetime.now()
        error = None

        try:
            result = await request_fn()
            satisfaction = self.calculate_satisfaction(result)
        except Exception as e:
            error = str(e)
            raise
        finally:
            duration = (datetime.now() - start).total_seconds() * 1000

            metric = AgentMetrics(
                timestamp=datetime.now(),
                response_time_ms=duration,
                tokens_used=getattr(result, 'tokens', 0),
                cost_brl=self.calculate_cost(result),
                user_satisfaction=satisfaction if error is None else 0,
                error_type=error
            )

            metric.log()
            self.check_alerts(metric)

Dashboard de Monitoramento

5. Fallbacks Salvam Sua Reputacao

O momento da verdade: Seu provedor de IA cai as 2 da manha. O que acontece?

Abordagem ruim:

# Rezar e esperar
response = openai.ChatCompletion.create(...)

Abordagem de producao:

from typing import List, Callable
import asyncio

class AIAgentWithFallbacks:
    def __init__(self):
        self.providers = [
            self.primary_ai,      # OpenAI GPT-4
            self.secondary_ai,    # Anthropic Claude
            self.rule_based,      # Respostas template
            self.human_handoff    # Ultimo recurso
        ]

    async def get_response(self, message: str, max_retries: int = 3) -> str:
        """Tentar provedores em ordem ate sucesso"""

        for provider in self.providers:
            for attempt in range(max_retries):
                try:
                    response = await provider(message)
                    if self.is_valid_response(response):
                        return response
                except Exception as e:
                    logging.warning(f"{provider.__name__} falhou: {e}")
                    await asyncio.sleep(2 ** attempt)  # Backoff exponencial
                    continue

        # Todos provedores falharam
        return "Desculpe, estou com dificuldades tecnicas. Um atendente humano vai te ajudar em breve."

Estatisticas de producao:

  • Uptime provedor primario: 99,2%
  • Ativacoes fallback: 124 vezes/mes
  • Reclamacoes de downtime: 0

A Arquitetura Que Realmente Funciona

Depois de 3 meses de iteracao, esse e o stack:

graph LR
    A[Usuario] --> B[Load Balancer]
    B --> C[API Gateway]
    C --> D{Rate Limiter}
    D --> E[Fila Mensagens]
    E --> F[Pool Agentes]
    F --> G[IA Primaria]
    F --> H[IA Fallback]
    F --> I[Engine Regras]
    G --> J[Validador]
    H --> J
    I --> J
    J --> K[Cache Respostas]
    K --> A

    L[Monitor] -.-> F
    L -.-> G
    L -.-> H
    M[Database] -.-> F

Arquitetura de Producao

Componentes principais:

  1. Load balancer - Distribui trafego
  2. Rate limiter - Protege contra abuso
  3. Fila de mensagens - Lida com picos
  4. Pool de agentes - Escala horizontalmente
  5. Validador - Pega alucinacoes
  6. Cache - Reduz custos em 40%
  7. Monitor - Alertas em tempo real

Numeros Reais Apos 3 Meses

MetricaValor
Total mensagens52.847
Tempo medio resposta847ms
Uptime99,97%
Custo por mensagemR$ 0,17
Satisfacao usuario4,6/5,0
Alucinacoes detectadas38
Tentativas abuso bloqueadas2.847
Ativacoes fallback124

O Que Eu Faria Diferente

Se comecasse do zero hoje:

  1. Comecar com rate limiting - Dia 1, nao dia 30
  2. Construir monitoramento primeiro - Nao da pra consertar o que nao ve
  3. Planejar alucinacoes - Elas VAO acontecer
  4. Projetar fallbacks cedo - Nao esperar por uma queda
  5. Cachear agressivamente - 40% reducao custo, zero esforco

O que funcionou perfeitamente:

  • SQLite para historico de conversas (sim, SQLite em producao)
  • Bun para servidor API (3x mais rapido que Node)
  • Fallbacks baseados em regras simples (salvou minha reputacao 2x)

O Codigo (Open Source)

Quer ver a implementacao real? Deixei open source os componentes principais:

GitHub: github.com/Richardmsbr/atlas-ai-chat

Inclui:

  • Rate limiter com limites adaptativos
  • Validador de respostas
  • Gerenciador de contexto
  • Sistema de fallbacks
  • Stack de monitoramento

Casos de Uso Brasileiros

Alem do meu caso, vejo isso funcionando perfeitamente para:

E-commerce

  • Atendimento 24/7 sem contratar time
  • Rastreio de pedidos automatico
  • Recomendacoes personalizadas
  • Custo: R$ 0,15-0,25 por interacao

Saude

  • Agendamento de consultas
  • Lembretes de medicacao
  • Triagem inicial de sintomas
  • Uptime critico: 99,9%+

Financeiro

  • Consulta saldo/extrato
  • PIX automatizado
  • Alertas de fraude
  • Validacao obrigatoria (zero alucinacoes)

SaaS/Software

  • Suporte tecnico L1
  • Onboarding usuarios
  • Coleta feedback
  • Integracao com Slack/Discord

Perguntas Frequentes

Q: Quanto custa rodar em producao? A: Meu custo: R$ 0,17/mensagem. Volume 50K/mes = ~R$ 8.500/mes. Muito menos que contratar 3 atendentes.

Q: Precisa de GPU? A: Nao. Uso APIs (OpenAI/Anthropic). Servidor roda em CPU simples (2 cores).

Q: E a LGPD? A: Critico. Nunca envio CPF/email/telefone pra LLM. Valido tudo antes. Tenho modulo open source pra isso: privacy-shield-br

Q: SQLite aguenta? A: Sim. 50K msgs/mes = ~1.600/dia = 1,1 msg/min. SQLite aguenta 100K writes/segundo.

Q: Como escalar pra 1M mensagens/mes? A: Horizontal scaling (mais workers), cache Redis, async processing. Arquitetura ja suporta.

Recursos Extras

Artigos Relacionados

Ferramentas Recomendadas

  • LLMs: OpenAI GPT-4, Anthropic Claude, Google Gemini
  • Infra: Railway (deploy), Render (alternativa)
  • Monitor: Sentry (erros), Grafana (metricas)
  • Database: PostgreSQL (relacional), Redis (cache)

Comunidades

Conclusao

Agentes de IA em producao sao completamente diferentes dos tutoriais.

Rate limiting, validacao, fallbacks, monitoramento - essas camadas nao sao opcionais. Sao essenciais.

Mas quando funciona? Vale cada linha de codigo.

50.847 mensagens processadas. Zero downtime. 4,6/5 satisfacao.

Isso nao e futuro. E hoje.


Quer a versao em ingles? Publiquei no dev.to com mais detalhes tecnicos: Building Production AI Agents - What Tutorials Won’t Tell You

Me acompanhe:

  • GitHub: @Richardmsbr
  • Construindo agentes de IA em escala
  • Solutions Architect focado em sistemas de IA em producao

Tem perguntas? Deixa nos comentarios - respondo todos!


Imagens: Unsplash

Gostou do conteudo?

Descubra como implementar IA no seu negocio com uma analise gratuita.

Agendar Analise Gratuita

Pronto para automatizar seu atendimento?

Agende uma analise gratuita e descubra como IA pode transformar seu negocio.

Agendar Analise Gratuita