سير عمل AI موثوق في Node.js باستخدام BullMQ: شرح عملي

ما الذي ستبنيه

يستعرض هذا الشرح العملي خدمة صغيرة لاستقبال مهام العمل، مصممة بعقلية الإنتاج. يرسل نظام آخر عنصر عمل إلى POST /webhooks/work-items. تتحقق API من الـ payload، وتخزّنه في PostgreSQL، وتضيف مهمة BullMQ، ثم تعيد 202 Accepted من دون انتظار استجابة AI.

يتلقى Worker منفصل المهمة من Redis، ويحمّل السجل المرجعي من PostgreSQL، ويطلب من OpenAI تصنيف العنصر، ويتحقق من JSON المعاد باستخدام Zod، ثم يحفظ النتيجة. وتوفر API المسار GET /work-items/:id للاستعلام الدوري، والمسار GET /ready للتحقق من الاعتماديات.

هذا الفصل مهم. يمكن لـ LLM المساعدة في التفسير المحدود، مثل التصنيف والتلخيص، لكنه لا ينبغي أن يصبح نظام السجل أو محرك السياسات. تتولى PostgreSQL ملكية حالة الأعمال. بينما ينسق Redis وBullMQ التنفيذ في الخلفية. ويفرض كود التطبيق معالجة حتمية للفئات الحساسة أمنياً.

يناسب هذا النمط أيضاً المؤسسات في دول مجلس التعاون الخليجي التي تتلقى طلبات دعم أو هندسة أو امتثال أو عمليات عبر أنظمة متعددة. قبل النشر، قيّم متطلبات إقامة البيانات والاحتفاظ بها وتقييم اللغة العربية وضوابط الوصول والاستضافة الإقليمية التي تنطبق على مؤسستك.

المتطلبات المسبقة

• Node.js ١٨ أو إصدار أحدث، وnpm.
• Docker Compose، أو نسخ PostgreSQL وRedis يمكن الوصول إليها.
• مفتاح OpenAI API ومعرّف نموذج متاح لحسابك.
• معرفة أساسية بـ TypeScript وSQL وHTTP ومتغيرات البيئة.

يدعم سياق سير العمل المتحقق منه البنية العامة: تستخدم أنظمة تنسيق AI قوائم انتظار مدعومة بـ Redis، وWorkers في الخلفية، وAPIs، وحالات للمهام، ومزودي تذاكر خارجيين. يحافظ هذا الشرح العملي عمداً على حزمة تقنية مُدارة ذاتياً ومبنية على الكود أولاً. يمكن للفرق التي تفضل بنية تحتية مُدارة لسير عمل TypeScript تقييم هذا الخيار بشكل منفصل، لكن حدود الموثوقية الموضحة هنا تظل سارية.

١. إنشاء المشروع

mkdir node-ai-workflow
cd node-ai-workflow
npm init -y
npm install bullmq dotenv express ioredis openai pg pino pino-http zod
npm install -D @types/express @types/node @types/pg tsx typescript
mkdir -p src db

استبدل package.json بـ scripts لعمليتي API وWorker مستقلتين.

{
  "name": "node-ai-workflow",
  "private": true,
  "type": "module",
  "scripts": {
    "dev:api": "tsx watch src/api.ts",
    "dev:worker": "tsx watch src/worker.ts",
    "start:api": "tsx src/api.ts",
    "start:worker": "tsx src/worker.ts"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "outDir": "dist"
  },
  "include": ["src/**/*.ts"]
}

أنشئ خدمات PostgreSQL وRedis محلية. تمثل PostgreSQL التخزين الدائم لسير العمل، بينما Redis هي الاعتمادية الخاصة بمعالجة BullMQ.

cat > docker-compose.yml <<'EOF'
services:
  postgres:
    image: postgres:alpine
    environment:
      POSTGRES_DB: ai_workflow
      POSTGRES_USER: workflow_user
      POSTGRES_PASSWORD: workflow_password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
  redis:
    image: redis:alpine
    command: ["redis-server", "--appendonly", "yes"]
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
volumes:
  postgres_data:
  redis_data:
EOF

docker compose up -d

cat > .env <<'EOF'
PORT=3000
LOG_LEVEL=info
DATABASE_URL=postgresql://workflow_user:workflow_password@localhost:5432/ai_workflow
REDIS_URL=redis://localhost:6379
OPENAI_API_KEY=replace-with-your-key
OPENAI_MODEL=replace-with-a-model-available-to-your-account
WEBHOOK_SHARED_SECRET=local-development-secret-change-before-production
EOF

cat > .gitignore <<'EOF'
node_modules
dist
.env
*.log
EOF

٢. إنشاء مخطط قاعدة البيانات

يُعد idempotency_key الفريد ضرورياً. فقد يعيد مرسلو Webhook محاولة الإرسال بعد انتهاء المهلة أو حدوث فشل في الشبكة. يحوّل قيد قاعدة البيانات الفريد عملية الإرسال المكررة إلى بحث قابل للتكرار بدلاً من تشغيل ثانٍ لسير العمل.

cat > db/001_create_work_items.sql <<'EOF'
CREATE EXTENSION IF NOT EXISTS pgcrypto;

CREATE TABLE work_items (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  idempotency_key TEXT NOT NULL UNIQUE,
  source TEXT NOT NULL,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
  status TEXT NOT NULL DEFAULT 'queued'
    CHECK (status IN ('queued', 'processing', 'completed', 'failed')),
  attempt_count INTEGER NOT NULL DEFAULT 0,
  ai_result JSONB,
  failure_reason TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  completed_at TIMESTAMPTZ
);

CREATE INDEX work_items_status_created_at_idx
ON work_items (status, created_at DESC);
EOF

docker compose exec -T postgres psql -U workflow_user -d ai_workflow < db/001_create_work_items.sql

٣. إضافة الإعدادات والمخططات المشتركة

cat > src/config.ts <<'EOF'
import "dotenv/config";
import { z } from "zod";const schema = z.object({
PORT: z.coerce.number().int().min(1).max(65535).default(3000),
LOG_LEVEL: z.enum(["trace", "debug", "info", "warn", "error", "fatal"]).default("info"),
DATABASE_URL: z.string().url(),
REDIS_URL: z.string().url(),
OPENAI_API_KEY: z.string().min(1),
OPENAI_MODEL: z.string().min(1),
WEBHOOK_SHARED_SECRET: z.string().min(16)
});const parsed = schema.safeParse(process.env);
if (!parsed.success) {
console.error(parsed.error.flatten().fieldErrors);
process.exit(1);
}
export const config = parsed.data;
EOFcat > src/db.ts <<'EOF'
import pg from "pg";
import { config } from "./config.js";
export const pool = new pg.Pool({ connectionString: config.DATABASE_URL, max: 10 });
EOFcat > src/redis.ts <<'EOF'
import IORedis from "ioredis";
import { config } from "./config.js";
export const redis = new IORedis(config.REDIS_URL, { maxRetriesPerRequest: null });
EOFcat > src/schemas.ts <<'EOF'
import { z } from "zod";export const inputSchema = z.object({
idempotencyKey: z.string().min(8).max(200),
...
تابع القراءة
سجل دخولك مجاناً لقراءة المقال كاملاً والوصول إلى أدوات الذكاء الاصطناعي.
تسجيل الدخول / إنشاء حساب