Supercharging AI Workflows: Разпределена обработка на документи с Node.js, Python и RabbitMQ

Когато вашата монолитна AI услуга започне да задушава документи Представете си услуга за обработка на документи, която обработва PDF качвания, извлича текст, изпълнява анализ на настроението и генерира обобщения.При нормална употреба тя работи добре до... деня, в който потребителят качва 500 документа наведнъж! Този вид провал ни учи на един общ урок: AI workloads should not run inside your web server. Решението? Разпределете работата между няколко служители на AI, позволете на всеки език да направи това, което прави най-добре, и използвайте опашка за съобщения, за да поддържате всичко координирано. Ето как изградих система, която обработва документи паралелно, като използва Node.js като оркестър и както Node.js, така и Python като служители на AI, всички комуникиращи чрез RabbitMQ и Celery. Защо тази архитектура има смисъл Преди да се потопите в кода, нека да поговорим за това защо тази конкретна комбинация работи: : Бърз, ориентиран към събития, идеален за обработка на HTTP заявки и координиране на задачи. Node.js as the orchestrator : Повечето AI/ML библиотеки (spaCy, трансформатори, scikit-learn) са Python-native. Running тежко AI обработка в Python работници поддържа вашия Node.js API леко. Python AI workers : За задачи, в които JavaScript се отличава (PDF анализиране, OCR предварителна обработка, определени NLP задачи), можете да имате и работници на възли. Node.js AI workers : Надежден брокер за съобщения, който гарантира, че задачите не се губят. RabbitMQ Разпределена опашка за задачи, тествана в битка за Python. Управлява ретриите, приоритетите и мащабирането на работниците извън кутията. Celery Резултатът е система, в която: 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 Архитектурата на един поглед Ето какво изграждаме: Потребителят качва документ чрез REST API. Node.js сървърът го записва, публикува задача на RabbitMQ и незабавно връща Job ID. Работниците на Python и Node.js вземат задачи въз основа на техните възможности. Всеки работник обработва своята част (текстово извличане, превод, анализ на настроението) и публикува резултатите обратно. Създаване на фондация Използвам Docker Compose, за да изпълнявам целия стак локално.Това прави тривиално да добавяте работници или да тествате в условия, подобни на производството. Ето и настройката на Docker Compose, която обединява всичко: 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 Започнете целия стак с и имате разпределен AI тръбопровод, който работи на вашата машина. docker-compose up -d Node.js API: поддържане на бързината API сървърът има една задача: получаване на заявки, изпращане на задачи и връщане на резултати. // 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')); Забележете как се връща незабавно с ID за работа. Действителното обработване се извършва асинхронно в работниците. /analyze Работници в Python: повдигане на тежести с Celery Използвам Celery, защото той се интегрира перфектно с RabbitMQ и се справя с цялата сложност на разпределената опашка. # 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 е, че можете да скалирате тези работници самостоятелно. Нуждаете се от повече сила за анализ на настроението? Просто добавете повече контейнери за работници: . docker-compose up -d --scale python-worker=5 Node.js работници: справяне с това, което JavaScript прави най-добре За задачи като PDF анализиране или предварителна обработка, работниците на Node.js са перфектни. Те консумират от същата опашка RabbitMQ и работят заедно с работниците на 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...'); Координационни резултати: агрегаторният модел Използвам модел на агрегатор, където работниците съхраняват резултатите си в Redis с последователна структура на ключа, а API ги агрегира при поискване. // 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 }; } Когато клиентът се обади , те получават пълна картина на всички завършени анализи. /status/:jobId Да се справяме с неуспехите грациозно В разпределени системи нещата се провалят. Работниците се сриват, мрежовите хикапи, моделите се изчерпват. Ето как да се справите с това: : 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 за съобщения, които се провалят многократно: 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' } }); За да се предотврати завиването на работници завинаги: 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 Проследяване на вашия разпределен тръбопровод При многобройни работници и опашки видимостта е критична. използвам конзолата за управление на RabbitMQ ( За да следите дълбочината на опашката и скоростта на съобщенията. http://localhost:15672 http://localhost:15672 За мониторинг на ниво приложение добавих метрична крайна точка: 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(); }); За производството бихте искали метриките на Prometheus и таблата Grafana, но това ви дава бързи прозрения по време на разработването. Ето и локалната табела от RabbitMQ: Скалиране на системата Красивото в тази архитектура е независимото скалиране: за повече AI обработваща мощност: Scale Python workers docker-compose up -d --scale python-worker=10 За повече PDF обработка пропуск: Scale Node workers docker-compose up -d --scale node-worker=5 Когато имате голям обем на заявката: Scale the API docker-compose up -d --scale api=3 Всеки компонент скалира самостоятелно въз основа на вашата бутилка. RabbitMQ се справя с балансирането на товара автоматично. Изпълнение на практика Тествах тази система с 1000 PDF документа (средно по 10 страници): : 47 минути, API не отговаря : 8 минути, API отзивчив през цялото време 3.5 минути, API отзивчив Monolith (single Node.js process) Distributed (2 Node workers, 4 Python workers) Distributed (5 Node workers, 10 Python workers) Пропускната способност скалира почти линейно с работниците, докато не достигнете I/O или мрежови ограничения. Реалният свят gotchas аз научих : Натоварването на моделите на трансформаторите отнема 10-20 секунди.Натоварвайте ги веднъж, когато работникът започне, а не по задача.Потребих часове за дебютиране на "бавни работници", преди да осъзная, че те претоварват модели всеки път. Model loading time RabbitMQ има ограничение за размера на съобщението по подразбиране.За големи документи съхранявайте файла и предавайте пътя в съобщението, а не самото съдържание. Message size limits : Работниците на Python AI могат да използват 2-4GB всеки в зависимост от моделите. Worker memory : Създайте една RabbitMQ връзка на работник и я използвайте отново. Connection pooling Кога да използвате тази архитектура Този модел има смисъл, когато: Вие обработвате задачи, които отнемат повече от няколко секунди Имате нужда от различни езици / време за изпълнение за различни части на вашия тръбопровод Искате да скалирате компонентите самостоятелно Надеждността е важна (RabbitMQ гарантира, че няма загубени задачи) Можете да добавите повече работни видове по-късно Това е излишно за прости CRUD API или задачи, които завършват в милисекунди. но за AI работни натоварвания, тръбопроводи за обработка на данни или всякакви CPU-интензивни фонови задачи, тази архитектура ме спаси безброй пъти. кодът Вижте пълния работен код на GitHub: https://github.com/ivmarcos/distributed-ai-document-analyzer https://github.com/ivmarcos/distributed-ai-document-analyzer Репозиторията включва пълната настройка на Docker Compose, API сървъра, двата типа работници, примерните документи и тестовия пакет. Виждам цялата система в действие. docker-compose up