Când serviciul dvs. monolit AI începe să se sufoceze pe documente Imaginați-vă un serviciu de procesare a documentelor care gestionează încărcările PDF, extrag textul, rulează analiza sentimentelor și generează rezumate. În timpul utilizării normale, funcționează bine până ... în ziua în care un utilizator încărcă 500 de documente la un moment dat! Acest tip de eșec ne învață o lecție comună: AI workloads should not run inside your web server. Distribuiți munca peste mai mulți lucrători AI, lăsați fiecare limbă să facă ceea ce face cel mai bine și utilizați o coadă de mesaje pentru a menține totul coordonat. Iată cum am construit un sistem care procesează documente în paralel folosind Node.js ca orchestrator și atât Node.js cât și Python ca lucrători AI, toate comunicând prin RabbitMQ și Celery. De ce are sens această arhitectură Înainte de a vă scufunda în cod, să vorbim despre motivul pentru care funcționează această combinație specifică: Rapid, bazat pe evenimente, perfect pentru gestionarea cererilor HTTP și coordonarea sarcinilor. serverul API rămâne receptiv chiar și atunci când prelucrează mii de documente. Node.js as the orchestrator Majoritatea bibliotecilor AI/ML (spaCy, transformatoare, scikit-learn) sunt native Python. Rularea procesării AI grele în lucrătorii Python menține API-ul Node.js ușor. Python AI workers : Pentru sarcinile în care JavaScript excelează (parsarea PDF, preprocesarea OCR, anumite sarcini NLP), puteți avea și lucrători Node. Node.js AI workers : Broker de mesaje de încredere care asigură că sarcinile nu se pierd. Dacă un lucrător se prăbușește, sarcina este reluată automat. RabbitMQ : Coada de sarcini distribuită testată în bătălie pentru Python. Gestionează retrișele, prioritățile și scalarea lucrătorilor din cutie. Celery Rezultatul este un sistem în care: 
 
 
 
 
 
 
 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 Arhitectura dintr-o privire Iată ce construim: Un utilizator încărcă un document prin intermediul API-ului REST. serverul Node.js îl salvează, publică o sarcină către RabbitMQ și returnează imediat un ID de sarcină. Lucrătorii Python și Node.js colectează sarcini în funcție de capacitățile lor. Fiecare lucrător procesează partea sa (extragerea textului, traducerea, analiza sentimentelor) și publică rezultatele înapoi. Stabilirea fundaţiei Folosesc Docker Compose pentru a rula întreaga stivă la nivel local. Acest lucru face trivial să adăugați lucrători sau să testați în condiții de producție. Iată setarea Docker Compose care aduce totul împreună: 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
 Începeți întregul stack cu și aveți un conductor AI distribuit care rulează pe mașina dvs. docker-compose up -d API-ul Node.js: păstrați-l rapid Serverul API are o singură sarcină: să primească solicitări, să trimită sarcini și să returneze rezultate. // 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ţi cum se întoarce imediat cu un ID de loc de muncă. procesarea actuală are loc în mod asincron în rândul lucrătorilor. /analyze Lucrătorii Python: ridicarea grea cu Celery Folosesc Celery pentru că se integrează perfect cu RabbitMQ și se ocupă de toată complexitatea cozii distribuite. # 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)
 Frumusețea Celery este că puteți scala acești lucrători în mod independent. Aveți nevoie de mai multă putere de analiză a sentimentelor? Doar adăugați mai multe containere pentru lucrători: . docker-compose up -d --scale python-worker=5 Lucrătorii Node.js: gestionarea a ceea ce face JavaScript cel mai bine Pentru sarcini precum parsing-ul PDF sau preprocesarea, lucrătorii Node.js sunt perfecți. Ei consumă din aceeași coadă RabbitMQ și lucrează alături de lucrătorii 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...');
 Rezultatele coordonării: modelul agregatorului Folosesc un model de agregator în care lucrătorii stochează rezultatele lor în Redis cu o structură de cheie coerentă, iar API le agregă la cerere. // 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
  };
}
 Când un client apelează , obțin o imagine completă a tuturor analizelor finalizate. /status/:jobId Gestionarea eșecurilor grațios În sistemele distribuite, lucrurile eșuează. Lucrătorii se prăbușesc, rețeaua se prăbușește, modelele se opresc. : 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
 pentru mesajele care nu reușesc în mod repetat: 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'
  }
});
 Pentru a evita spânzurarea permanentă a lucrătorilor: 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
 Monitorizarea conductei dvs. distribuite Cu mai multe angajați și cozi, vizibilitatea este critică. folosesc consola de management RabbitMQ ( pentru a monitoriza adâncimea cozii și ratele mesajelor. http://localhost:15672 http://localhost:15672 Pentru monitorizarea la nivel de aplicație, am adăugat un endpoint metric: 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();
});
 Pentru producție, ați dori metrice Prometheus și tablouri de bord Grafana, dar acest lucru vă oferă informații rapide în timpul dezvoltării. Iată tabloul de bord local de la RabbitMQ: Scalarea sistemului Lucrul frumos despre această arhitectură este scalarea independentă: Pentru mai multă putere de procesare: Scale Python workers docker-compose up -d --scale python-worker=10
 Pentru mai multe detalii despre procesarea PDF: Scale Node workers docker-compose up -d --scale node-worker=5
 Când aveți un volum mare de cereri: Scale the API docker-compose up -d --scale api=3
 Fiecare componentă scalează în mod independent în funcție de butelie. RabbitMQ se ocupă de echilibrarea sarcinii în mod automat. Performanță în practică Am testat acest sistem cu 1000 de documente PDF (în medie 10 pagini fiecare): 47 de minute, foc fără răspuns : 8 minute, API receptiv pe tot parcursul : 3,5 minute, API responsive Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) Scalele de debit sunt aproape liniare cu lucrătorii până când atingeți limitele I/O sau ale rețelei. Gothas din lumea reală pe care le-am învăţat : Încărcarea modelelor de transformator durează 10-20 de secunde. Încărcați-le o dată când începe lucrătorul, nu pe sarcină. Am pierdut ore de debugging "lucrători lenti" înainte de a realiza că se reîncărcă modelele de fiecare dată. Model loading time Pentru documente mari, stocați fișierul și treceți calea în mesaj, nu conținutul în sine. Message size limits Lucrătorii Python AI pot folosi 2-4 GB fiecare, în funcție de modele. Worker memory Creați o conexiune RabbitMQ pe lucrător și reutilizați-o. Connection pooling Când să folosiți această arhitectură Acest model are sens atunci când: 
 
 
 
 
 
 Procesezi sarcini care durează mai mult de câteva secunde Aveți nevoie de limbi diferite / timpi de rulare pentru diferite părți ale conductei dvs. Vrei să scali componentele în mod independent Fiabilitatea contează (RabbitMQ garantează că nu există sarcini pierdute) Puteți adăuga mai multe tipuri de lucrători mai târziu Este overkill pentru API-uri simple CRUD sau sarcini care se completează în milisecunde. dar pentru încărcături de lucru AI, conducte de prelucrare a datelor sau orice lucrări de fundal intensive cu CPU, această arhitectură m-a salvat de nenumărate ori. Codul Verificați codul complet de lucru pe GitHub: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer Depozitul include setarea completă Docker Compose, serverul API, ambele tipuri de lucrători, exemple de documente și o suită de teste. Pentru a vedea întregul sistem în acțiune. 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

Acest audio este produs în limba originală a poveștii!

Supercharging AI Workflows: Prelucrarea documentelor distribuite cu Node.js, Python și RabbitMQ

About Author

COMENTARII

Etichete suspendate

ACEST ARTICOL A FOST PREZENTAT IN

Related Stories

The Robots Are Taking Over

Elon Just Made Gambling Define the News - And It's More Honest Than Journalism Ever Was

The Spacecoin Writing Contest by Spacecoin and HackerNoon: Round 2 Results 🎉

The #bitcoin Writing Contest by Rootstock and HackerNoon: Round 2 Results 🎉

The Robots Are Taking Over

Elon Just Made Gambling Define the News - And It's More Honest Than Journalism Ever Was

The Spacecoin Writing Contest by Spacecoin and HackerNoon: Round 2 Results 🎉

The #bitcoin Writing Contest by Rootstock and HackerNoon: Round 2 Results 🎉

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps