Cuando su servicio de IA monolítico comienza a ahogarse en documentos Imagínese un servicio de procesamiento de documentos que maneja cargas de PDF, extrae texto, ejecuta análisis de sentimientos y genera resúmenes. En el uso normal, funciona bien hasta... el día en que un usuario carga 500 documentos a la vez! De repente, el servidor Node.js se detiene durante 20 minutos, bloqueando otras solicitudes y dejando a los usuarios sin actualizaciones de estado o comentarios sobre el progreso. Este tipo de fracaso enseña una lección común: AI workloads should not run inside your web server. Distribuir el trabajo a través de múltiples trabajadores de IA, dejar que cada lenguaje haga lo que hace mejor, y usar una cola de mensajes para mantener todo coordinado.Así es como construí un sistema que procesa documentos en paralelo utilizando Node.js como orquestador y tanto Node.js como Python como trabajadores de IA, todo comunicando a través de RabbitMQ y Celery. ¿Por qué esta arquitectura tiene sentido? Antes de sumergirse en el código, hablemos de por qué funciona esta combinación específica: Rápido, orientado a eventos, perfecto para el manejo de solicitudes HTTP y la coordinación de tareas. su servidor de API permanece receptivo incluso cuando procesa miles de documentos. Node.js as the orchestrator La mayoría de las bibliotecas de IA/ML (spaCy, transformadores, scikit-learn) son nativas de Python. Ejecutar el procesamiento de IA pesado en los trabajadores de Python mantiene su API Node.js ligera. Python AI workers : Para tareas en las que JavaScript excele (parsamiento de PDF, preprocesamiento de OCR, ciertas tareas de NLP), también puede tener trabajadores de nodo. Node.js AI workers Broker de mensajes fiable que asegura que las tareas no se pierdan.Si un trabajador se estrella, la tarea se repite automáticamente. RabbitMQ : La cola de tareas distribuida para Python, probada por la batalla, maneja los retiros, las prioridades y el escalado de los trabajadores fuera de la caja. Celery El resultado es un sistema en el 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 La arquitectura a simple vista Aquí está lo que estamos construyendo: Un usuario carga un documento a través de la API REST. El servidor Node.js lo guarda, publica una tarea a RabbitMQ, y inmediatamente devuelve un ID de trabajo. Los trabajadores de Python y Node.js recogen tareas basadas en sus capacidades. Cada trabajador procesa su parte (extracción de texto, traducción, análisis de sentimientos) y publica los resultados de vuelta. Creación de la Fundación Estoy usando Docker Compose para ejecutar toda la pila localmente. Esto hace que sea trivial agregar trabajadores o probar en condiciones similares a la producción. Aquí está la configuración de Docker Compose que lo une todo: 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
 Comience todo el proceso con y tienes un tubo de IA distribuido que se ejecuta en tu máquina. docker-compose up -d La API de Node.js: manteniéndolo rápido El servidor de API tiene una tarea: recibir solicitudes, enviar tareas y 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'));
 Observa cómo Retorna inmediatamente con un ID de trabajo.El procesamiento real ocurre de forma asíncrona en los trabajadores. /analyze Trabajadores de Python: Levantamiento pesado con Celery Estoy usando Celery porque se integra perfectamente con RabbitMQ y maneja toda la complejidad de la cola distribuida. # 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)
 La belleza de Celery es que puede escalar estos trabajadores de forma independiente. ¿Necesita más poder de análisis de sentimientos? simplemente añade más contenedores de trabajadores: . docker-compose up -d --scale python-worker=5 Trabajadores de Node.js: manejo de lo que JavaScript hace mejor Para tareas como el análisis de PDF o el preprocesamiento, los trabajadores de Node.js son perfectos. Consumen de la misma cola de RabbitMQ y trabajan junto a los trabajadores de 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 coordinación: el patrón de agregador Utilizo un patrón de agregador donde los trabajadores almacenan sus resultados en Redis con una estructura de clave consistente, y la API los agrega a petición. // 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
  };
}
 Cuando un cliente llama , obtienen una imagen completa de todos los análisis completos. /status/:jobId Tratar los fracasos con gracia En los sistemas distribuidos, las cosas fallan. los trabajadores se colisionan, las redes se rompen, los modelos se agotan. Aquí está cómo manejarlo: : 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 los mensajes que fallan 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 los trabajadores se colguen para siempre: 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
 Control de tu tubería distribuida En la mayoría de los casos, el uso de la herramienta de control es muy importante, ya que se utiliza la herramienta de control ( ) para controlar la profundidad de la cola y las tasas de mensajes. http://localhost:15672 http://localhost:15672 Para el seguimiento a nivel de aplicación, agregé un punto final de métricas: 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 la producción, querrías las métricas de Prometheus y los dashboards de Grafana, pero esto te da una visión rápida durante el desarrollo. Aquí está el dashboard local de RabbitMQ: Escalar el sistema Lo bello de esta arquitectura es el escalado independiente: Para más potencia de procesamiento: Scale Python workers docker-compose up -d --scale python-worker=10
 Para obtener más datos PDF: Scale Node workers docker-compose up -d --scale node-worker=5
 Si tienes un alto volumen de solicitudes: Scale the API docker-compose up -d --scale api=3
 Cada componente escala de forma independiente en función de su botella. RabbitMQ maneja el equilibrio de carga automáticamente. El rendimiento en la práctica He probado este sistema con 1.000 documentos PDF (una media de 10 páginas cada uno): : 47 minutos, API sin respuesta : 8 minutos, API responsive durante todo el tiempo 3,5 minutos, API Responder Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) Las escalas de rendimiento son casi lineales con los trabajadores hasta que llegue a los límites de I/O o de red. Gothas del mundo real que aprendí Cargar modelos de transformadores toma 10-20 segundos. cargarlos una vez cuando el trabajador comienza, no por tarea. He perdido horas debugando "trabajadores lentos" antes de darse cuenta de que estaban cargando modelos cada vez. Model loading time RabbitMQ tiene un límite de tamaño de mensaje por defecto. Para documentos grandes, almacene el archivo y pasa el camino en el mensaje, no el contenido en sí. Message size limits : Los trabajadores de Python AI pueden usar 2-4 GB cada uno dependiendo de los modelos. Worker memory Crear una conexión RabbitMQ por trabajador y reutilizarla. Connection pooling Cuándo utilizar esta arquitectura Este patrón tiene sentido cuando: 
 
 
 
 
 
 Estás procesando tareas que tardan más de unos segundos Usted necesita diferentes idiomas / tiempos de ejecución para diferentes partes de su tubería ¿Quieres escalar componentes de forma independiente La fiabilidad importa (RabbitMQ garantiza que no se pierdan tareas) Puede agregar más tipos de trabajadores más tarde Es overkill para APIs CRUD simples o tareas que se completan en milisegundos. pero para cargas de trabajo de IA, tuberías de procesamiento de datos o cualquier trabajo de fondo intensivo en CPU, esta arquitectura me ha salvado innumerables veces. Código Consulte el código de trabajo completo en GitHub: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer El repositorio incluye la configuración completa de Docker Compose, el servidor de API, ambos tipos de trabajadores, documentos de ejemplo y una suite de pruebas. Ver todo el sistema en acción. 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 audio es producido en el idioma original de la historia!

Supercharger flujos de trabajo de IA: procesamiento de documentos distribuidos con Node.js, Python y RabbitMQ

About Author

COMENTARIOS

ETIQUETAS

ESTE ARTÍCULO FUE PRESENTADO EN

Related Stories

Nómadas digitales escuchen: lo que necesitan saber sobre la nueva visa DTV de Tailandia

Navegando por las aguas: desarrollo de aplicaciones RAG de nivel de producción con lagos de datos

Aumente su productividad con estas 18 herramientas para desarrolladores 🚀🔥

La fuga rápida del sistema Claude Sonnet 3.5: un análisis forense

Nómadas digitales escuchen: lo que necesitan saber sobre la nueva visa DTV de Tailandia

Navegando por las aguas: desarrollo de aplicaciones RAG de nivel de producción con lagos de datos

Aumente su productividad con estas 18 herramientas para desarrolladores 🚀🔥

La fuga rápida del sistema Claude Sonnet 3.5: un análisis forense

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps