Amikor a monolit AI-szolgáltatás elkezd fulladni a dokumentumokban Képzeljünk el egy dokumentumfeldolgozási szolgáltatást, amely feldolgozza a PDF-letöltéseket, kivonja a szöveget, futtatja a hangulatelemzéseket és összefoglalókat készít. Normál használat esetén jól működik... egészen a napig, amikor egy felhasználó egyszerre 500 dokumentumot tölt fel! Ez a fajta kudarc egy közös leckét tanít: AI workloads should not run inside your web server. A megoldás? Ossza el a munkát több AI-munkás között, hagyja, hogy mindegyik nyelv a legjobban csinálja, és használjon üzenetsorozatot, hogy mindent összehangoljon.Íme, hogyan építettem egy olyan rendszert, amely párhuzamosan feldolgozza a dokumentumokat a Node.js-szel, valamint a Node.js-szel és a Python-szel mint AI-munkásokkal, amelyek mindegyike a RabbitMQ és a Celery segítségével kommunikál. Miért van értelme ennek az építészetnek Mielőtt belemerülnénk a kódba, beszéljünk arról, hogy miért működik ez a konkrét kombináció: Gyors, eseményvezérelt, tökéletes a HTTP-kérelmek feldolgozásához és a feladatok koordinálásához.Az API-kiszolgáló még több ezer dokumentum feldolgozása esetén is reagál. Node.js as the orchestrator : A legtöbb AI/ML könyvtár (spaCy, transzformátorok, scikit-learn) Python-alapú. Python AI workers : Olyan feladatok esetében, ahol a JavaScript kiemelkedik (PDF-elemzés, OCR-előfeldolgozás, bizonyos NLP-tevékenységek), Node-munkások is lehetnek. Node.js AI workers : Megbízható üzenetközvetítő, amely biztosítja, hogy a feladatok ne veszjenek el. Ha egy munkavállaló összeomlik, a feladat automatikusan újraindul. RabbitMQ : Harcban tesztelt elosztott feladatsor a Python számára. Kezeli a retries, a prioritások és a dolgozók skálázását a dobozból. Celery Az eredmény egy olyan rendszer, amelyben: 
 
 
 
 
 
 
 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 Az építészet egy pillantásra Itt van, amit építünk: A felhasználó feltölti a dokumentumot a REST API-n keresztül. A Node.js kiszolgáló mentette, közzétett egy feladatot a RabbitMQ-nek, és azonnal visszaadta a munkakódot. A Python és a Node.js munkatársak felveszik a feladatokat képességeik alapján. Mindegyik munkatárs feldolgozza a részét (szövegkivonat, fordítás, érzelmi elemzés) és közzéteszi az eredményeket. Alapítvány létrehozása A Docker Compose-t használom a teljes halom helyi futtatására, ami triviálisvá teszi a dolgozók hozzáadását vagy a termeléshez hasonló körülmények között történő tesztelést. Íme a Docker Compose beállítása, amely mindent összehoz: 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
 Indítsa el az egész csomagot a és van egy elosztott AI csővezeték, amely a gépen fut. docker-compose up -d A Node.js API: tartsa gyors Az API-kiszolgálónak egyetlen feladata van: kérések fogadása, feladatok küldése és eredmények visszaküldése. // 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'));
 Figyeljük meg, hogyan A munkahelyi azonosítóval azonnal visszatér, a tényleges feldolgozás a munkavállalókban aszinkron módon történik. /analyze Python munkavállalók: nehéz emelés Celery-vel A Celery-t azért használom, mert tökéletesen integrálódik a RabbitMQ-vel, és kezeli az összes elosztott sor összetettségét. # 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 Celery szépsége az, hogy önállóan skálázhatja ezeket a munkavállalókat. Több érzéselemzési erőre van szüksége? Csak adjon hozzá több munkavállalói tartályt: . docker-compose up -d --scale python-worker=5 Node.js munkavállalók: kezeli, hogy mi javítja a JavaScript Olyan feladatokhoz, mint például a PDF-elemzés vagy az előfeldolgozás, a Node.js munkások tökéletesek. Ugyanabból a RabbitMQ sorból fogyasztanak, és a Python munkásokkal együtt dolgoznak. // 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...');
 Koordinációs eredmények: az aggregátor mintája Az egyik kihívás az elosztott munkavállalókkal az eredmények összegyűjtése.Egy aggregátor mintát használok, ahol a munkavállalók eredményeiket következetes kulcsstruktúrával tárolják a Redisben, és az API kérésre aggregálja őket. // 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
  };
}
 Amikor az ügyfél felhívja Teljes képet kap az összes elvégzett elemzésről. /status/:jobId A kudarcok tisztességes kezelése Az elosztott rendszerekben a dolgok meghibásodnak. A munkavállalók összeomlanak, a hálózatok meghibásodnak, a modellek eltűnnek. Itt van, hogyan kell kezelni: : 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
 olyan üzenetek esetében, amelyek ismételten sikertelenek: 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'
  }
});
 hogy megakadályozzák a munkavállalók örökre felakasztása: 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
 Az Ön elosztott csővezetékének ellenőrzése A többszörös munkavállalók és sorok esetében a láthatóság kritikus. a RabbitMQ menedzsment konzolt használom ( ) figyelemmel kíséri a sorok mélységét és az üzenetek sebességét. http://localhost:15672 http://localhost:15672 Az alkalmazásszintű monitorozáshoz hozzáadtam egy metrikus végpontot: 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();
});
 A termeléshez Prometheus metrikákat és Grafana táblákat szeretne, de ez gyors betekintést nyújt a fejlesztés során. Íme a RabbitMQ helyi eszköztár: A rendszer skálázása Ennek az építészetnek a gyönyörűsége a független skálázás: A feldolgozóerő növelése érdekében: Scale Python workers docker-compose up -d --scale python-worker=10
 További PDF feldolgozási eredmények: Scale Node workers docker-compose up -d --scale node-worker=5
 Ha nagy mennyiségű kérésed van: Scale the API docker-compose up -d --scale api=3
 Mindegyik alkatrész önállóan méretezi a palackozást.A RabbitMQ automatikusan kezeli a terhelés kiegyensúlyozását. Teljesítmény a gyakorlatban Ezt a rendszert 1000 PDF dokumentummal teszteltem (átlagosan 10 oldal): : 47 perc, tűz nem reagál : 8 perc, API-reaktív egész idő alatt 3,5 perc, API responsive Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) Az áramlási skálák szinte lineárisak a munkavállalókkal, amíg el nem éri az I/O vagy a hálózati korlátokat. Valós világ gotchas megtanultam : A transzformátormodellek betöltése 10-20 másodpercet vesz igénybe. töltse fel őket egyszer, amikor a munkavállaló elindul, nem feladatonként. órákat pazaroltam a „lassú munkavállalók” hibakeresésével, mielőtt rájöttem, hogy minden alkalommal újratöltötték a modelleket. Model loading time A RabbitMQ alapértelmezett üzenetméret-korlátozással rendelkezik.A nagyméretű dokumentumok esetében tárolja a fájlt, és adja át az utat az üzenetben, nem pedig a tartalmat. Message size limits : A Python AI munkatársai a modellektől függően 2-4 GB-ot használhatnak. Worker memory Hozzon létre egy RabbitMQ-kapcsolatot munkavállalónként, és használja újra. Connection pooling Mikor kell használni az építészetet Ez a módszer akkor van értelme, ha: 
 
 
 
 
 
 Olyan feladatok feldolgozása, amelyek több mint néhány másodpercet vesz igénybe Különböző nyelvekre / futási időre van szüksége a csővezeték különböző részeihez Szeretné önállóan méretezni az alkatrészeket A megbízhatóság fontos (a RabbitMQ nem garantálja az elveszett feladatokat) Később további munkavállalói típusokat is hozzáadhat Az egyszerű CRUD API-k vagy a milliszekundumokban elvégzendő feladatok túlzás.De az AI-munkaterhelések, az adatfeldolgozási csővezetékek vagy bármilyen CPU-intenzív háttérmunka esetében ez az architektúra számtalan alkalommal mentett meg. Kód Nézze meg a teljes munkakódot a GitHub-on: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer A raktár tartalmazza a teljes Docker Compose beállítást, az API-kiszolgálót, mindkét munkatípust, a példadokumentumokat és egy tesztcsomagot. Az egész rendszert cselekvésben látni. 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

Ez a hanganyag a történet eredeti nyelvén készült!

Supercharging AI munkafolyamatok: Elosztott dokumentumfeldolgozás Node.js, Python és RabbitMQ segítségével

About Author

HOZZÁSZÓLÁSOK

HANG TAGOK

EZT A CIKKET BEMUTATTA

Related Stories

HackerNoon Social Media Presence: A Complete Guide

A SECOND ENIGMA

INTRODUCTION

HackerNoon Contributor Denisa Ganea on HackerNoon, Human Existence, and Everything in Between

HackerNoon Social Media Presence: A Complete Guide

A SECOND ENIGMA

INTRODUCTION

HackerNoon Contributor Denisa Ganea on HackerNoon, Human Existence, and Everything in Between

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps