Como construir um agente de IA local com streaming em tempo real (Ollama + Wikipédia)

“Streaming” que confunde muita gente

Quando se fala em agentes de IA, a palavra “streaming” aparece com dois significados diferentes, e a maioria dos tutoriais só constrói um deles. Ora significa que o agente consome um fluxo contínuo de eventos, em vez de esperar alguém digitar; ora significa que a saída do agente é transmitida token a token. Este guia — do especialista Shittu Olumide, da KDnuggets — constrói os dois de propósito, porque resolvem problemas distintos e um agente “sempre ligado” de verdade precisa de ambos.

O cenário é concreto: um agente local que vigia o feed público de edições ao vivo da Wikipédia, sem chave de API, e raciocina sobre quais edições parecem vandalismo — tudo rodando na sua própria máquina via Ollama.

A decisão de design que importa

O feed de edições da Wikipédia não é um fio d’água: em dia movimentado, empurra várias edições por segundo. Entregar cada uma a um modelo de linguagem queima o processamento da sua máquina em edições que nunca foram interessantes — e o agente fica para trás do fluxo ao vivo, derrotando o propósito de ser “sempre ligado”.

A correção é um funil de dois estágios:

• Estágio 1 — filtro barato: matemática simples em Python, sem modelo, em todo evento (quantos bytes a edição removeu, quantas edições o usuário fez nos últimos minutos). A maioria esmagadora é chata, e chato é barato de detectar.
• Estágio 2 — o LLM local: só acorda para a pequena fração que dispara um limiar. É o mesmo princípio de qualquer bom sistema de monitoramento.

As quatro peças do build

1. Consumidor do stream

O serviço EventStreams da Wikipédia empurra edições como Server-Sent Events por HTTP simples — sem chave, sem handshake além de um GET. Um detalhe importante: a Wikipédia não envia um campo explícito “é anônimo”. Edições anônimas são atribuídas ao endereço IP como nome de usuário, então a detecção de anonimato verifica se o username tem formato de IPv4 ou IPv6.

def is_anonymous_user(username: str) -> bool:
    return bool(_IPV4_RE.match(username) or _IPV6_RE.match(username))

O gerador assíncrono ainda se reconecta automaticamente em caso de queda, com espera de 5 segundos — um serviço “sempre ligado” que morre na primeira conexão perdida não é sempre ligado.

2. Filtro barato (Estágio 1)

Um rastreador de velocidade mantém, por usuário, uma fila de timestamps de edições recentes numa janela deslizante. Isso transforma “5 edições em 2 minutos” num número contínuo e preciso. Há ainda um limite de memória: sem ele, o dicionário cresceria para sempre num stream que nunca para.

3. Raciocinador local (Estágio 2)

Aqui, um schema estrito importa. O modelo é obrigado a responder um JSON válido — is_likely_vandalism, severity (1 a 5) e reasoning — porque o Ollama aplica o schema diretamente na geração. O resultado é um objeto tipado usável no código, não “JSON geralmente válido” que você precisa parsear defensivamente. E o agente ainda transmite cada token conforme chega, para exibição em tempo real.

4. Broadcast para clientes

Cada cliente conectado recebe sua própria fila asyncio.Queue. Um assinante lento descarta mensagens com elegância em vez de bloquear o loop que processa edições ao vivo — sem essa separação, uma única aba de navegador lenta poderia travar o agente inteiro.

Como rodar

ollama pull llama3.1:8b
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
uvicorn src.main:app --reload

Depois, em outro terminal, curl -N http://localhost:8000/events para ver as mensagens flagged, token e verdict chegarem ao vivo. As edições chatas — a vasta maioria — nunca aparecem, o que é exatamente o ponto.

Escalando de verdade

Para produção real — múltiplas fontes, múltiplos processos, sobreviver a reinício sem perder eventos — a evolução natural é trocar o stream direto e o broadcast em memória por um barramento de mensagens como Kafka entre o produtor e o estágio de raciocínio.

A lição por baixo de todo o código não é sobre Wikipédia, Ollama ou FastAPI: eficiência deixa de ser uma otimização que você adiciona depois no momento em que o agente passa de “responde quando perguntado” para “sempre ligado”. Um agente de chat ocioso não custa nada; um agente de streaming está, por definição, sempre consumindo algo.