Quando seu serviço de IA monolítico começa a sufocar em documentos Imagine um serviço de processamento de documentos que gerencia carregamentos de PDF, extrai texto, executa análise de sentimentos e gera resumos. Em uso normal, ele funciona bem até... o dia em que um usuário carrega 500 documentos de uma só vez! De repente, o servidor Node.js pára por 20 minutos, bloqueando outras solicitações e deixando os usuários sem atualizações de status ou feedback sobre o progresso. Este tipo de fracasso ensina uma lição comum: AI workloads should not run inside your web server. A solução? Distribua o trabalho entre vários trabalhadores de IA, deixe que cada idioma faça o que faz melhor e use uma fila de mensagens para manter tudo coordenado.Eis como construí um sistema que processa documentos em paralelo usando o Node.js como orquestrador e tanto o Node.js quanto o Python como trabalhadores de IA, todos comunicando através do RabbitMQ e do Celery. Por que essa arquitetura faz sentido Antes de mergulhar no código, vamos falar sobre por que essa combinação específica funciona: Rápido, orientado por eventos, perfeito para lidar com solicitações HTTP e coordenar tarefas.O seu servidor de API permanece responsivo mesmo quando processa milhares de documentos. Node.js as the orchestrator A maioria das bibliotecas de AI/ML (spaCy, transformadores, scikit-learn) são nativas de Python. Executar processamento de IA pesado em trabalhadores de Python mantém sua API Node.js leve. Python AI workers : Para tarefas onde o JavaScript se destaca (parsagem de PDF, pré-processamento de OCR, certas tarefas NLP), você pode ter trabalhadores de nó também. Node.js AI workers Corretor de mensagens confiável que garante que as tarefas não sejam perdidas.Se um trabalhador falhar, a tarefa é recuada automaticamente. RabbitMQ : Faixa de tarefas distribuída testada por batalha para Python. Gerencia retries, prioridades e escalagem de trabalhadores fora da caixa. Celery O resultado é um sistema em que: 
 
 
 
 
 
 
 Your API responds in milliseconds, not minutes 
 
 Workers can scale independently (add 10 Python workers if you need more NLP power) 
 
 Failed tasks automatically retry 
 
 You can monitor everything in real-time A arquitetura a um olhar Aqui está o que estamos construindo: Um usuário carrega um documento através da API REST. O servidor Node.js salva-o, publica uma tarefa para RabbitMQ e imediatamente retorna um ID de trabalho. Trabalhadores do Python e do Node.js recolhem tarefas com base em suas capacidades. Cada trabalhador processa sua parte (extracção de texto, tradução, análise de sentimentos) e publica os resultados de volta. Criando a Fundação Eu estou usando o Docker Compose para executar toda a pilha localmente.Isso torna trivial adicionar trabalhadores ou testes em condições semelhantes à produção. Aqui está a configuração do Docker Compose que reúne tudo: version: '3.8'

services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  api:
    build:
      context: ./api
    ports:
      - "3000:3000"
    depends_on:
      - rabbitmq
      - redis
    environment:
      RABBITMQ_URL: amqp://admin:admin123@rabbitmq:5672
      REDIS_URL: redis://redis:6379
    volumes:
      - ./uploads:/app/uploads

  python-worker:
    build:
      context: ./workers/python
    depends_on:
      - rabbitmq
      - redis
    environment:
      CELERY_BROKER_URL: amqp://admin:admin123@rabbitmq:5672
      CELERY_RESULT_BACKEND: redis://redis:6379
    deploy:
      replicas: 2

  node-worker:
    build:
      context: ./workers/node
    depends_on:
      - rabbitmq
      - redis
    environment:
      RABBITMQ_URL: amqp://admin:admin123@rabbitmq:5672
      REDIS_URL: redis://redis:6379
    deploy:
      replicas: 2
 Comece todo o pacote com e você tem um pipeline de IA distribuído em execução em sua máquina. docker-compose up -d O Node.js API: mantendo-o rápido O servidor de API tem uma tarefa: receber solicitações, enviar tarefas e devolver resultados. // api/src/server.js
import express from 'express';
import amqp from 'amqplib';
import { createClient } from 'redis';
import multer from 'multer';
import { v4 as uuidv4 } from 'uuid';

const app = express();
const upload = multer({ dest: 'uploads/' });

let channel, redisClient;

async function initializeConnections() {
  const connection = await amqp.connect(process.env.RABBITMQ_URL);
  channel = await connection.createChannel();
  await channel.assertQueue('document_analysis', { durable: true });

  redisClient = createClient({ url: process.env.REDIS_URL });
  await redisClient.connect();
}

app.post('/analyze', upload.single('document'), async (req, res) => {
  const jobId = uuidv4();
  const { analysisType = 'full' } = req.body;

  const task = {
    jobId,
    filePath: req.file.path,
    fileName: req.file.originalname,
    analysisType,
    createdAt: new Date().toISOString()
  };

  // Publish to RabbitMQ
  channel.sendToQueue('document_analysis',
    Buffer.from(JSON.stringify(task)),
    { persistent: true }
  );

  // Store initial status in Redis
  await redisClient.set(`job:${jobId}`, JSON.stringify({
    status: 'queued',
    progress: 0,
    createdAt: task.createdAt
  }), { EX: 86400 }); // 24h expiry

  res.json({ jobId, status: 'queued' });
});

app.get('/status/:jobId', async (req, res) => {
  const data = await redisClient.get(`job:${req.params.jobId}`);

  if (!data) {
    return res.status(404).json({ error: 'Job not found' });
  }

  res.json(JSON.parse(data));
});

await initializeConnections();
app.listen(3000, () => console.log('API running on port 3000'));
 Observe como Retorna imediatamente com um ID de trabalho. O processamento real ocorre de forma assíncrona nos trabalhadores. /analyze Trabalhadores de Python: levantamento pesado com Celery Eu estou usando o Celery porque ele se integra perfeitamente com o RabbitMQ e lida com toda a complexidade da fila distribuída. # workers/python/tasks.py
from celery import Celery
import redis
import json
from transformers import pipeline
import spacy

# Initialize Celery
app = Celery('tasks',
             broker=os.getenv('CELERY_BROKER_URL'),
             backend=os.getenv('CELERY_RESULT_BACKEND'))

# Load models once at startup
nlp = spacy.load('en_core_web_sm')
sentiment_analyzer = pipeline('sentiment-analysis')

redis_client = redis.from_url(os.getenv('CELERY_RESULT_BACKEND'))

@app.task(bind=True)
def analyze_sentiment(self, job_id, text):
    """Analyze sentiment of document text"""
    try:
        update_progress(job_id, 30, 'Analyzing sentiment')

        # Process in chunks if text is large
        chunk_size = 512
        chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]

        sentiments = []
        for chunk in chunks:
            result = sentiment_analyzer(chunk)[0]
            sentiments.append(result)

        # Aggregate results
        avg_score = sum(s['score'] for s in sentiments) / len(sentiments)
        dominant_label = max(set(s['label'] for s in sentiments),
                            key=lambda x: sum(1 for s in sentiments if s['label'] == x))

        return {
            'sentiment': dominant_label,
            'confidence': avg_score,
            'details': sentiments[:5]  # First 5 chunks for detail
        }
    except Exception as e:
        self.retry(exc=e, countdown=60, max_retries=3)

@app.task(bind=True)
def extract_entities(self, job_id, text):
    """Extract named entities using spaCy"""
    try:
        update_progress(job_id, 50, 'Extracting entities')

        doc = nlp(text)

        entities = {}
        for ent in doc.ents:
            entity_type = ent.label_
            if entity_type not in entities:
                entities[entity_type] = []
            entities[entity_type].append({
                'text': ent.text,
                'start': ent.start_char,
                'end': ent.end_char
            })

        return entities
    except Exception as e:
        self.retry(exc=e, countdown=60, max_retries=3)

def update_progress(job_id, progress, message):
    """Update job progress in Redis"""
    job_key = f'job:{job_id}'
    data = redis_client.get(job_key)
    if data:
        job_data = json.loads(data)
        job_data['progress'] = progress
        job_data['message'] = message
        redis_client.set(job_key, json.dumps(job_data), ex=86400)
 A beleza de Celery é que você pode escalar esses trabalhadores de forma independente. Precisa de mais poder de análise de sentimentos? basta adicionar mais contêineres de trabalhadores: . docker-compose up -d --scale python-worker=5 Trabalhadores do Node.js: lidar com o que o JavaScript faz melhor Para tarefas como arquivo de PDF ou pré-processamento, os trabalhadores do Node.js são perfeitos. Eles consomem da mesma fila RabbitMQ e trabalham ao lado dos trabalhadores do Python. // workers/node/worker.js
import amqp from 'amqplib';
import { createClient } from 'redis';
import pdf from 'pdf-parse';
import fs from 'fs/promises';

const redisClient = createClient({ url: process.env.REDIS_URL });
await redisClient.connect();

async function processDocument(task) {
  const { jobId, filePath, analysisType } = task;

  try {
    await updateProgress(jobId, 10, 'Extracting text from PDF');

    // Read and parse PDF
    const dataBuffer = await fs.readFile(filePath);
    const pdfData = await pdf(dataBuffer);

    const result = {
      text: pdfData.text,
      pages: pdfData.numpages,
      metadata: pdfData.info,
      wordCount: pdfData.text.split(/\s+/).length
    };

    await updateProgress(jobId, 20, 'Text extracted, queuing analysis tasks');

    // Publish extracted text to Python workers for AI analysis
    if (analysisType === 'full' || analysisType === 'sentiment') {
      await publishTask('sentiment_analysis', {
        jobId,
        text: result.text
      });
    }

    if (analysisType === 'full' || analysisType === 'entities') {
      await publishTask('entity_extraction', {
        jobId,
        text: result.text
      });
    }

    // Store extraction results
    await storeResult(jobId, 'extraction', result);

  } catch (error) {
    console.error(`Error processing document ${jobId}:`, error);
    await updateProgress(jobId, -1, `Error: ${error.message}`);
  }
}

async function updateProgress(jobId, progress, message) {
  const jobKey = `job:${jobId}`;
  const data = await redisClient.get(jobKey);

  if (data) {
    const jobData = JSON.parse(data);
    jobData.progress = progress;
    jobData.message = message;
    jobData.status = progress < 0 ? 'failed' :
                     progress === 100 ? 'completed' : 'processing';

    await redisClient.set(jobKey, JSON.stringify(jobData), { EX: 86400 });
  }
}

// Connect to RabbitMQ and start consuming
const connection = await amqp.connect(process.env.RABBITMQ_URL);
const channel = await connection.createChannel();
await channel.assertQueue('document_analysis', { durable: true });
channel.prefetch(1);

channel.consume('document_analysis', async (msg) => {
  if (msg) {
    const task = JSON.parse(msg.content.toString());
    await processDocument(task);
    channel.ack(msg);
  }
});

console.log('Node.js worker waiting for documents...');
 Resultados de coordenação: o padrão de agregador Eu uso um padrão de agregador onde os trabalhadores armazenam seus resultados no Redis com uma estrutura de chave consistente, e a API os agrega a pedido. // api/src/aggregator.js
export async function getJobResults(jobId) {
  const jobData = await redisClient.get(`job:${jobId}`);
  if (!jobData) return null;

  const job = JSON.parse(jobData);

  // Gather all result keys for this job
  const resultKeys = await redisClient.keys(`job:${jobId}:*`);

  const results = {};
  for (const key of resultKeys) {
    const [, , resultType] = key.split(':');
    const data = await redisClient.get(key);
    results[resultType] = JSON.parse(data);
  }

  return {
    jobId,
    status: job.status,
    progress: job.progress,
    message: job.message,
    createdAt: job.createdAt,
    results
  };
}
 Quando um cliente chama , eles obtêm uma imagem completa de todas as análises concluídas. /status/:jobId Gerencie os fracassos com graça Em sistemas distribuídos, as coisas falham. Colisões de trabalhadores, colisões de rede, timeout de modelos. Aqui está como lidar com isso: : Celery automatic retries @app.task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3, 'countdown': 5})
def risky_task(self, data):
    # If this fails, Celery will retry 3 times with 5-second delays
    pass
 Para mensagens que falham repetidamente: Dead letter queues await channel.assertQueue('document_analysis_dlq', { durable: true });
await channel.assertQueue('document_analysis', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': '',
    'x-dead-letter-routing-key': 'document_analysis_dlq'
  }
});
 Para evitar que os trabalhadores sejam enforcados para sempre: Timeouts @app.task(time_limit=300, soft_time_limit=270)
def long_running_task(data):
    # Hard limit at 5 minutes, soft limit warning at 4.5 minutes
    pass
 Acompanhe o seu pipeline distribuído Com vários trabalhadores e filas, a visibilidade é crítica. uso o console de gerenciamento RabbitMQ ( Para monitorar as profundidades de fila e as taxas de mensagens. http://localhost:15672 http://localhost:15672 Para monitoramento de nível de aplicação, adicionei um endpoint métrico: app.get('/metrics', async (req, res) => {
  const connection = await amqp.connect(process.env.RABBITMQ_URL);
  const channel = await connection.createChannel();

  const queue = await channel.checkQueue('document_analysis');

  res.json({
    queueDepth: queue.messageCount,
    consumers: queue.consumerCount,
    // Could add more metrics from Redis, worker health checks, etc.
  });

  await connection.close();
});
 Para a produção, você gostaria de métricas Prometheus e painéis Grafana, mas isso lhe dá insights rápidos durante o desenvolvimento. Aqui está o painel de controle local do RabbitMQ: Escalar o sistema A coisa bonita sobre esta arquitetura é a escalabilidade independente: Para mais poder de processamento: Scale Python workers docker-compose up -d --scale python-worker=10
 Para mais informações sobre o PDF: Scale Node workers docker-compose up -d --scale node-worker=5
 Quando você tem um grande volume de solicitações: Scale the API docker-compose up -d --scale api=3
 Cada componente escala de forma independente com base no seu ponto de engarrafamento. RabbitMQ lida com o balanço de carga automaticamente. Desempenho na prática Eu testei este sistema com 1000 documentos PDF (em média 10 páginas cada): : 47 minutos, API não responde : 8 minutos, API responsivo ao longo do tempo : 3,5 minutos, API responsivo Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) As escalas de transmissão são quase lineares com os trabalhadores até atingir os limites de I/O ou de rede. Gotchas do mundo real que aprendi Carregá-los uma vez quando o trabalhador começa, não por tarefa. eu perdi horas debugando "trabalhadores lentos" antes de perceber que eles estavam recarregando modelos cada vez. Model loading time RabbitMQ tem um limite de tamanho de mensagem padrão.Para documentos grandes, armazene o arquivo e passe o caminho na mensagem, não o conteúdo em si. Message size limits : Os trabalhadores de Python AI podem usar 2-4GB cada, dependendo dos modelos. Worker memory Criar uma conexão RabbitMQ por trabalhador e reutilizá-la. Connection pooling Quando usar essa arquitetura Este padrão faz sentido quando: 
 
 
 
 
 
 Você está processando tarefas que levam mais do que alguns segundos Você precisa de diferentes idiomas / tempos de execução para diferentes partes do seu pipeline Você quer escalar componentes de forma independente Confiabilidade é importante (RabbitMQ não garante tarefas perdidas) Você pode adicionar mais tipos de trabalhadores mais tarde É um overkill para APIs CRUD simples ou tarefas que terminam em milissegundos.Mas para cargas de trabalho de IA, pipelines de processamento de dados ou qualquer trabalho de fundo intensivo com CPU, essa arquitetura me salvou inúmeras vezes. Código Veja o código completo no GitHub: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer O repositório inclui a configuração completa do Docker Compose, o servidor API, ambos os tipos de trabalhadores, documentos de exemplo e um conjunto de testes. Ver todo o sistema em ação. docker-compose up

This story contains new, firsthand information uncovered by the writer.

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

Read My Stories

Este áudio é produzido no idioma original da história!

Superchargeando fluxos de trabalho de IA: processamento de documentos distribuídos com Node.js, Python e RabbitMQ

About Author

COMENTARIOS

Rótulos

ESTE ARTIGO FOI APRESENTADO EM

Related Stories

Telegram: a ponte da Crypto Island para o continente

Digital Nomads Ouçam: O que você precisa saber sobre o novo visto DTV da Tailândia

Liberando o poder da IA. Uma revisão sistemática de técnicas de ponta: resumo e introdução

Vazamento do prompt do sistema Claude Sonnet 3.5: uma análise forense

Telegram: a ponte da Crypto Island para o continente

Digital Nomads Ouçam: O que você precisa saber sobre o novo visto DTV da Tailândia

Liberando o poder da IA. Uma revisão sistemática de técnicas de ponta: resumo e introdução

Vazamento do prompt do sistema Claude Sonnet 3.5: uma análise forense

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps