När din monolitiska AI-tjänst börjar kväva på dokument Föreställ dig en dokumentbearbetningstjänst som hanterar PDF-uppladdningar, extraherar text, kör sentimentanalys och genererar sammanfattningar. Under normal användning fungerar den bra tills... den dag en användare laddar upp 500 dokument på en gång! Plötsligt stannar Node.js-servern i 20 minuter, blockerar andra förfrågningar och lämnar användare utan statusuppdateringar eller feedback om framsteg. Denna typ av misslyckande lär en gemensam läxa: AI workloads should not run inside your web server. Lösningen? Distribuera arbetet över flera AI-arbetare, låt varje språk göra vad det gör bäst, och använd en budskapskedja för att hålla allt samordnat.Här är hur jag byggde ett system som bearbetar dokument parallellt med Node.js som en orkestrator och både Node.js och Python som AI-arbetare, alla som kommunicerar genom RabbitMQ och Celery. Varför denna arkitektur är meningsfull Innan vi dyker in i koden, låt oss prata om varför denna specifika kombination fungerar: : Snabb, händelsedriven, perfekt för att hantera HTTP-förfrågningar och samordna uppgifter. Node.js as the orchestrator De flesta AI/ML-bibliotek (spaCy, transformatorer, scikit-learn) är Python-nativa. Att köra tung AI-bearbetning i Python-arbetare håller din Node.js API lätt. Python AI workers : För uppgifter där JavaScript utmärker sig (PDF-parsing, OCR-förbearbetning, vissa NLP-uppgifter) kan du också ha Node-arbetare. Node.js AI workers : Pålitlig meddelandemäklare som säkerställer att uppgifter inte går förlorade. Om en arbetare kraschar, tas uppgiften automatiskt tillbaka. RabbitMQ : Battle-testad distribuerad uppgiftskedja för Python. Hanterar retries, prioriteringar och arbetare skala ut ur lådan. Celery Resultatet är ett system där: 
 
 
 
 
 
 
 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 Arkitekturen på en blick Här är vad vi bygger: En användare laddar upp ett dokument via REST API. Node.js-servern sparar det, publicerar en uppgift till RabbitMQ och returnerar omedelbart ett jobb-ID. Python- och Node.js-arbetare plockar upp uppgifter baserat på sina förmågor. Varje arbetare bearbetar sin del (textutvinning, översättning, sentimentanalys) och publicerar resultaten tillbaka. Att sätta upp stiftelsen Jag använder Docker Compose för att köra hela stacken lokalt. Detta gör det trivialt att lägga till arbetare eller testa i produktionsliknande förhållanden. Här är inställningen Docker Compose som för samman allt: 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
 Starta hela stacken med och du har en distribuerad AI-rörledning som körs på din maskin. docker-compose up -d Node.js API: hålla det snabbt API-servern har en uppgift: ta emot förfrågningar, skicka uppgifter och returnera resultat. // 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'));
 Notera hur Returnerar omedelbart med ett jobb-ID. Den faktiska bearbetningen sker asynkront i arbetarna. /analyze Python-arbetare: tung lyftning med Celery Jag använder Celery eftersom det integreras perfekt med RabbitMQ och hanterar all distribuerad kökomplexitet. # 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)
 Skönheten i Celery är att du kan skala dessa arbetare självständigt. Behöver du mer känslomässig analysstyrka? Lägg bara till fler arbetare behållare: . docker-compose up -d --scale python-worker=5 Node.js-arbetare: hantera vad JavaScript gör bäst För uppgifter som PDF-parsing eller förbearbetning är Node.js-arbetare perfekta. De förbrukar från samma RabbitMQ-skede och arbetar tillsammans med Python-arbetare. // 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...');
 Samordnande resultat: aggregatorns mönster Jag använder ett aggregatormönster där arbetstagare lagrar sina resultat i Redis med en konsekvent nyckelstruktur, och API aggregerar dem på begäran. // 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
  };
}
 När en kund ringer De får en fullständig bild av alla slutförda analyser. /status/:jobId Att hantera misslyckanden på ett elegant sätt I distribuerade system misslyckas saker. Arbetstagare kraschar, nätverkshiccups, modeller timout. Här är hur man hanterar det: : 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
 För meddelanden som misslyckas upprepade gånger: 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'
  }
});
 för att förhindra att arbetstagare hängs för evigt: 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
 Övervaka din distribuerade rörledning Med flera arbetstagare och köer är synlighet avgörande. jag använder RabbitMQ management console ( ) för att övervaka ködjup och meddelandegrader. http://localhost:15672 http://localhost:15672 För övervakning på applikationsnivå har jag lagt till en metrisk slutpunkt: 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();
});
 För produktion vill du ha Prometheus-mätvärden och Grafana-dashboards, men detta ger dig snabba insikter under utvecklingen. Här är den lokala instrumentpanelen från RabbitMQ: Skala upp systemet Det vackra med denna arkitektur är oberoende skalning: För mer AI processorkraft: Scale Python workers docker-compose up -d --scale python-worker=10
 För mer PDF-bearbetning genomströmning: Scale Node workers docker-compose up -d --scale node-worker=5
 När du har hög volym: Scale the API docker-compose up -d --scale api=3
 Varje komponent skalar självständigt baserat på din flaskhals. RabbitMQ hanterar lastbalanseringen automatiskt. Prestanda i praktiken Jag testade detta system med 1000 PDF-dokument (i genomsnitt 10 sidor vardera): : 47 minuter, API svarar inte : 8 minuter, API responsiv hela tiden : 3,5 minuter, API responsiv Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) Genomströmningen skalar nästan linjärt med arbetare tills du träffar I/O eller nätverksgränser. Real-world gotchas jag lärt mig : Ladda transformatormodeller tar 10-20 sekunder. ladda dem en gång när arbetaren startar, inte per uppgift. jag slösade timmar med att debugga "slow workers" innan jag insåg att de laddar om modeller varje gång. Model loading time RabbitMQ har en standardmeddelande storlek gräns. För stora dokument, spara filen och passera vägen i meddelandet, inte innehållet självt. Message size limits Python AI-arbetare kan använda 2-4 GB vardera beroende på modeller. Worker memory : Skapa en RabbitMQ-anslutning per arbetare och återanvänd den. Connection pooling När du använder denna arkitektur Detta mönster är meningsfullt när: 
 
 
 
 
 
 Du bearbetar uppgifter som tar mer än några sekunder Du behöver olika språk/körtider för olika delar av din pipeline Du vill skala komponenterna självständigt Tillförlitlighet spelar roll (RabbitMQ garanterar inga förlorade uppgifter) Du kan lägga till fler typer av arbetare senare Det är överkill för enkla CRUD-API:er eller uppgifter som slutförs på millisekunder.Men för AI-arbetsbelastningar, databehandlingsrörledningar eller några CPU-intensiva bakgrundsarbeten har denna arkitektur sparat mig otaliga gånger. Koden är Kolla in den fullständiga koden på GitHub: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer Repositoryn innehåller den fullständiga Docker Compose-installationen, API-servern, båda arbetstyperna, exempeldokumenten och en testpaket. Att se hela systemet i aktion. 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

Det här ljudet är producerat på berättelsens originalspråk!

Supercharging AI Workflows: Distributed Document Processing med Node.js, Python och RabbitMQ

About Author

KOMMENTARER

HÄNG TAGGAR

DENNA ARTIKEL PRESENTERAS I

Related Stories

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps