Monolit AI hizmetiniz belgelerde boğulmaya başladığında PDF yüklemelerini yöneten, metni çıkaran, duyguların analizini yürüten ve özetleri oluşturan bir belge işleme hizmeti hayal edin. Normal kullanımda, bir kullanıcı bir kez 500 belge yüklerken... gününe kadar iyi performans gösterir! Aniden, Node.js sunucusu 20 dakika boyunca durur, diğer talepleri engeller ve kullanıcıların durum güncellemeleri veya ilerleme geribildirimleri olmadan bırakır. Bu tür bir başarısızlık ortak bir ders öğretir: AI workloads should not run inside your web server. Çözüm? Çalışmayı birden fazla AI çalışanına dağıtın, her dilin en iyi yaptığı şeyi yapmasına izin verin ve her şeyi koordine etmek için bir mesaj sırası kullanın. RabbitMQ ve Celery aracılığıyla iletişim kuran Node.js, Node.js ve Python gibi bir orkestratör olarak belgeleri paralel olarak işleyen bir sistem oluşturdum. Neden bu mimarlık mantıklı Kod içine dalmadan önce, bu özel kombinasyonun neden çalıştığını konuşalım: Hızlı, olay odaklı, HTTP istekleri ve koordinasyon görevleri işlemek için mükemmel. API sunucunuz binlerce belge işlerken bile duyarlı kalır. Node.js as the orchestrator : Çoğu AI/ML kütüphanesi (spaCy, transformatörler, scikit-learn) Python'a özgüdür. Python çalışanlarında ağır AI işlemlerini çalıştırmak Node.js API'nizi hafif tutar. Python AI workers : JavaScript'in mükemmel olduğunu gösteren görevler için (PDF analizi, OCR ön işleme, belirli NLP görevleri), Node çalışanlarına da sahip olabilirsiniz. Node.js AI workers : Görevleri kaybedemeyeceğinden emin olan güvenilir mesaj broker. bir işçi çökerse, görevi otomatik olarak tekrarlanır. RabbitMQ : Battle-tested dağıtılmış görev sırası Python için. Retries, öncelikleri ve işçi boyutlandırma kutusundan yönetir. Celery Sonuç olarak bir sistem var: 
 
 
 
 
 
 
 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 Bir bakışta mimarlık İşte inşa ettiğimiz şey: Bir kullanıcı bir belgeyi REST API aracılığıyla yükler. Node.js sunucusu onu kaydeder, RabbitMQ'ye bir görevi yayınlar ve hemen bir iş kimliği iade eder. Python ve Node.js çalışanları kendi yeteneklerine göre görevleri alır. Her çalışan parçalarını işler (tekst çıkarma, çeviri, duygu analizi) ve sonuçları geri yayınlar. Vakfı kurmak Docker Compose'ı, tüm seti yerel olarak çalıştırmak için kullanıyorum.Bu, işçileri eklemek veya üretim benzeri koşullarda test etmek zorlaştırıyor. İşte her şeyi bir araya getiren Docker Compose kurulumu: 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
 Tüm stack ile başlatın ve makinenizde çalışan dağıtılmış bir AI boru hattınız var. docker-compose up -d Node.js API: Hızlı tutun API sunucusunun tek bir görevi vardır: istekleri almak, görevleri göndermek ve sonuçları iade etmek. // 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'));
 Bakın nasıl iş kimliği ile hemen geri döner. gerçek işleme işçilerde eşzamanlı olarak gerçekleşir. /analyze Python Çalışanları: Celery ile Ağır Yükseltme Celery kullanıyorum çünkü RabbitMQ ile mükemmel bir şekilde entegre olur ve dağıtılan sıra karmaşıklığını ele alır. # 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)
 Celery'nin güzelliği, bu işçilerin bağımsız olarak ölçeklenebileceğidir. Daha fazla duygusal analiz gücüne ihtiyacınız var mı? Sadece daha fazla işçi konteynerini ekleyin: . docker-compose up -d --scale python-worker=5 Node.js Çalışanları: JavaScript'in En İyi Yaptığını Yönetme PDF analizi veya ön işleme gibi görevler için, Node.js işçileri mükemmel. Aynı RabbitMQ sırasından tüketir ve Python işçileri ile birlikte çalışır. // 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...');
 Koordinasyon sonuçları: aggregator model Dağıtılmış çalışanlarla ilgili bir zorluk, sonuçların toplanmasıdır. işçilerin sonuçlarını tutarlı bir anahtar yapısı ile Redis'te depoladıkları ve API'nin talep üzerine bunları topladığı bir agregatör şablonu kullanıyorum. // 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
  };
}
 Bir müşteri çağırdığında Tüm tamamlanan analizlerin tam bir resmini alırsınız. /status/:jobId Başarısızlıkları kolayca halletmek Dağıtılmış sistemlerde işler başarısız olur. işçilerin çöküşü, ağ çöküşleri, modellerin zaman kaybı. : 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
 Tekrar tekrar başarısız olan mesajlar için: 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'
  }
});
 işçilerin sonsuza kadar asılmasını engellemek için: 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
 dağıtılmış boru hattını kontrol etmek Çeşitli işçi ve sıralarda görünürlük kritiktir. ben RabbitMQ yönetim konsolunu kullanıyorum ( ) sıra derinliği ve mesaj hızı izlemek için. http://localhost:15672 http://localhost:15672 Uygulama düzeyinde izleme için, bir metrik son noktası ekledim: 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();
});
 Üretim için, Prometheus metrikleri ve Grafana panolarını istersiniz, ancak bu size geliştirme sırasında hızlı bir bilgi verir. İşte RabbitMQ'nin yerel kontrol paneli: Sistemi genişletmek Bu mimari hakkında güzel şey bağımsız ölçeklendirme: Daha fazla işlem gücü için: Scale Python workers docker-compose up -d --scale python-worker=10
 Daha fazla PDF dosyası için: Scale Node workers docker-compose up -d --scale node-worker=5
 Yüksek talep hacmine sahipseniz: Scale the API docker-compose up -d --scale api=3
 Her bir bileşen, şişenize bağlı olarak bağımsız olarak ölçeklenir. RabbitMQ yük dengesini otomatik olarak ele alır. Pratik performans Bu sistemi 1000 PDF belgesiyle test ettim (her biri ortalama 10 sayfa): : 47 dakika, yanıtlanmamış API : 8 dakika, tüm API yanıtlı 3.5 dakika, API yanıtlı Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) I/O veya ağ sınırlarına ulaşana kadar işçilerle neredeyse lineer ölçümler. Öğrendiğim Gerçek Dünya Gotchas : transformatör modellerinin yüklenmesi 10-20 saniye alır. işçi başlarken bir kez yükleme, görev başına değil. her seferinde modelleri yeniden yüklediklerini fark etmeden önce "yavaş işçiler" deneme saatlerini boşa harcadım. Model loading time RabbitMQ varsayılan bir mesaj boyut sınırına sahiptir. büyük belgeler için dosyayı depolayın ve içeriğin kendisi değil mesajda yol geçirin. Message size limits : Python AI çalışanları, modellerine bağlı olarak her birini 2-4 GB kullanabilir. Worker memory Kullanıcı başına bir RabbitMQ bağlantısı oluşturun ve yeniden kullanın. Connection pooling Bu mimariyi ne zaman kullanmalıyız Bu çerçeve anlamlıdır: 
 
 
 
 
 
 Birkaç saniyeden fazla süren görevleri işliyor Pipelinizin farklı bölümleri için farklı diller / çalıştırma zamanları gerektirir Bağımsız olarak bileşenleri ölçmek istiyorsunuz Güvenilirlik önemlidir (RabbitMQ kayıp görevleri garanti etmez) Daha sonra daha fazla işçi türü ekleyebilirsiniz Basit CRUD API'ler veya milisekundaki görevler için aşırı derecede.Ama AI iş yükleri, veri işleme boruları veya herhangi bir CPU yoğun arka plan işleri için, bu mimari beni sayısız kez kurtardı. Kodlama Github'da tam çalışma kodunu kontrol edin: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer Depo, Docker Compose'ın tam kurulumunu, API sunucusunu, her iki işçi türünü, örnek belgeleri ve bir test paketi içerir. Tüm sistemin işleyişini görmek. 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

Bu ses hikayenin orijinal dilinde üretilmiştir!

Supercharging AI Workflows: Node.js, Python ve RabbitMQ ile dağıtılmış belge işleme

About Author

YORUMLAR

ETİKETLERİ ASIN

BU YAZI

Related Stories

Bu 18 Geliştirici Aracıyla Üretkenliğinizi Artırın 🚀🔥

Kullanıcı Odaklı Kripto Ürünleri Oluşturma: Müşteri Geri Bildiriminin Önemi

Yapay Zekanın Gücünü Ortaya Çıkarıyoruz. En Son Tekniklerin Sistematik Bir İncelemesi: Özet ve Giriş

Dijital Göçebeler Dinleyin: Tayland'ın Yeni DTV Vizesi Hakkında Bilmeniz Gerekenler

Bu 18 Geliştirici Aracıyla Üretkenliğinizi Artırın 🚀🔥

Kullanıcı Odaklı Kripto Ürünleri Oluşturma: Müşteri Geri Bildiriminin Önemi

Yapay Zekanın Gücünü Ortaya Çıkarıyoruz. En Son Tekniklerin Sistematik Bir İncelemesi: Özet ve Giriş

Dijital Göçebeler Dinleyin: Tayland'ın Yeni DTV Vizesi Hakkında Bilmeniz Gerekenler

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps