بناء خلفية ذكاء اصطناعي للإنتاج باستخدام Motia: واجهات API والمهام الخلفية والوكلاء في إطار واحد

الخلفيات الحديثة مجزّأة. واجهة الـ API لديك تعيش في إطار، والمهام الخلفية في نظام طوابير مثل BullMQ أو Celery، والمهام المجدولة في مشغّل cron، ووكلاء الذكاء الاصطناعي في خدمة Python ما، أما المراقبة فتُضاف لاحقًا. كل قطعة لها قصة نشر خاصة، ونموذج ذهني خاص، وطريقة فشل خاصة.

يتبنّى Motia منهجًا مختلفًا. فهو يوحّد واجهات API والمهام الخلفية والمهام المجدولة والبث اللحظي وإدارة الحالة ووكلاء الذكاء الاصطناعي في إطار واحد مبني حول وحدة أساسية واحدة: الـ Step. إذا كان React قد جعل كل شيء في الواجهة الأمامية مكوّنًا (component)، فإن Motia يجعل كل شيء في الخلفية خطوة (Step) — ويتيح لك مزج TypeScript وJavaScript وPython داخل سير العمل نفسه.

في هذا الدرس ستبني خلفية كاملة لفرز تذاكر الدعم بالذكاء الاصطناعي: نقطة نهاية HTTP تستقبل تذكرة، وخطوة حدث خلفية تستدعي نموذجًا لغويًا لتصنيف الأولوية وصياغة ردّ، وتُبثّ النتائج إلى العميل لحظيًا، وخطوة Python تضيف تحليل الكلمات المفتاحية، وخطوة cron يومية تُولّد ملخصًا. في النهاية ستفهم كل مفهوم أساسي في Motia من خلال كود فعّال.

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

قبل البدء، تأكد من توفّر:

• Node.js 20+ وnpm مثبّتين
• Python 3.10+ (مطلوب فقط للخطوة متعددة اللغات في الخطوة 7)
• معرفة أساسية بـ TypeScript وبنمط async/await
• مفتاح OpenAI API (أو أي مزوّد متوافق) لخطوة الذكاء الاصطناعي
• محرر أكواد (يُنصح بـ VS Code)

لا تحتاج إلى خبرة سابقة في طوابير الرسائل أو مكتبات cron أو خوادم WebSocket. يوفّر Motia كل ذلك.

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

خط معالجة قائم على الأحداث من أربع مراحل يُبرز كل أنواع خطوات Motia:

1. خطوة API — يتحقّق POST /tickets من الطلب باستخدام Zod ثم يُطلق حدثًا ويعود فورًا.
2. خطوة حدث — تشترك في الحدث، وتستدعي نموذجًا لغويًا لتصنيف التذكرة وصياغة ردّ، وتحفظ النتيجة في الحالة.
3. بث (Stream) — يدفع تحديثات الحالة اللحظية ("استُلمت" ← "قيد التحليل" ← "تمّت") إلى العميل عبر WebSocket.
4. خطوة Python — تستخرج الكلمات المفتاحية من نص التذكرة باستخدام Python الأصلية.
5. خطوة Cron — تعمل كل صباح لتلخيص تذاكر اليوم.

جمال هذه البنية أن كل مرحلة منفصلة، ويُعاد تنفيذها تلقائيًا عند الفشل، وقابلة للمراقبة في مُنقّح بصري مدمج.

الخطوة 1: إعداد المشروع

يأتي Motia بأداة إنشاء تفاعلية. أنشئ مشروعًا جديدًا:

npx motia@latest create

تسألك الأداة عن اسم المشروع وقالب ولغة. اختر قالب TypeScript وسمِّ المشروع ticket-triage. ثم شغّل خادم التطوير:

cd ticket-triage
npm run dev

يُطلق هذا أمرين معًا: خلفيتك على http://localhost:3000 وMotia Workbench، وهي وحدة تحكم بصرية لفحص خطواتك وتشغيلها وتتبّعها — على http://localhost:3000 أيضًا. افتحها في المتصفح؛ ستعود إليها طوال هذا الدرس.

يبدو المشروع الجديد هكذا:

ticket-triage/
├── steps/                 # كل Step يعيش هنا، يُكتشف تلقائيًا
│   └── hello-world.step.ts
├── .env                   # متغيرات البيئة
├── package.json
├── tsconfig.json
└── motia-workbench.json   # تخطيط الـ workbench (يُدار تلقائيًا)

الفكرة الأساسية: أي ملف يطابق *.step.ts أو *.step.js أو *_step.py داخل steps/ يُكتشف تلقائيًا ويُربط في وقت التشغيل. لا يوجد موجّه مركزي، ولا تسجيل يدوي، ولا app.use(). تكتب ملفًا، فيجده Motia.

أضف مفتاح الـ API إلى .env:

OPENAI_API_KEY=sk-your-key-here

الخطوة 2: فهم وحدة الـ Step

كل Step هو ملف واحد يُصدّر شيئين بالضبط:

• config — كائن بسيط يصف ماهية الخطوة: نوعها، وكيف تُطلَق، وما الأحداث التي تُطلقها أو تشترك فيها، ومخططات التحقق الخاصة بها.
• handler — دالة async تحتوي منطق عملك.

هناك ثلاثة أنواع من الخطوات ستستخدمها:

يستقبل كل handler كائن context كوسيطه الثاني. الـ context هو مكمن قوة Motia:

handler = async (input, { emit, logger, state, streams }) => { /* ... */ }

• emit — يُطلق حدثًا لتفعيل خطوات الأحداث اللاحقة.
• logger — تسجيل مُهيكَل يظهر في الـ Workbench، مترابط لكل طلب.
• state — مخزن مفتاح-قيمة مدمج، مُجمّع ومستمر، دون أي إعداد لقاعدة بيانات.
• streams — قنوات لحظية مُسمّاة تدفع البيانات إليها؛ ويشترك العملاء فيها عبر WebSocket.

أنت لا تستورد عميل طوابير، ولا اتصال Redis، ولا خادم WebSocket. الـ context يسلّمها لك جاهزة ومتتبَّعة.

الخطوة 3: إنشاء خطوة الـ API

احذف ملف hello-world.step.ts المُولَّد وأنشئ steps/submit-ticket.step.ts. هذه هي البوابة الأمامية لخط معالجتنا. تتحقق من التذكرة الواردة باستخدام Zod، وتزرع حالة أولية في بثّ، وتُطلق حدثًا للمعالجة الخلفية — ثم تعود فورًا كي لا ينتظر العميل النموذج اللغوي أبدًا.

import { ApiRouteConfig, Handlers } from 'motia'
import { z } from 'zod'
import { randomUUID } from 'crypto'
 
export const config: ApiRouteConfig = {
  type: 'api',
  name: 'SubmitTicket',
  description: 'يستقبل تذكرة دعم ويُدرجها في طابور الفرز بالذكاء الاصطناعي',
  path: '/tickets',
  method: 'POST',
  // تحقّق من جسم الطلب قبل أن يعمل الـ handler إطلاقًا
  bodySchema: z.object({
    subject: z.string().min(1),
    body: z.string().min(1),
    email: z.string().email(),
  }),
  responseSchema: {
    200: z.object({ ticketId: z.string(), status: z.string() }),
  },
  // هذه الخطوة تُطلق خط المعالجة الخلفي
  emits: ['ticket.submitted'],
  flows: ['ticket-triage'],
}
 
export const handler: Handlers['SubmitTicket'] = async (req, { emit, logger, streams }) => {
  const ticketId = randomUUID()
  const { subject, body, email } = req.body
 
  logger.info('Ticket received', { ticketId, email })
 
  // ازرع حالة لحظية يمكن للعميل الاشتراك فيها فورًا
  await streams.ticketStatus.set(ticketId, 'status', {
    ticketId,
    stage: 'received',
    priority: null,
    draftReply: null,
  })
 
  // سلّم المعالجة لخط المعالجة الخلفي وعُد فورًا
  await emit({
    topic: 'ticket.submitted',
    data: { ticketId, subject, body, email },
  })
 
  return {
    status: 200,
    body: { ticketId, status: 'received' },
  }
}

لاحظ ثلاثة أمور. حقل bodySchema يعني أن الطلبات المشوّهة تُرفض بالرمز 400 قبل أن يعمل كودك — دون تحقّق يدوي. ومصفوفة emits تُعلن العقد: تُنتج هذه الخطوة حدث ticket.submitted. ويعود الـ handler خلال أجزاء من الثانية لأن كل العمل البطيء يحدث لاحقًا في السلسلة.

يجمع حقل flows الخطوات المترابطة معًا كي يرسمها الـ Workbench كمخطط واحد متصل.

الخطوة 4: خطوة الحدث — تصنيف بالذكاء الاصطناعي

الآن العامل الخلفي. أنشئ steps/triage-ticket.step.ts. يشترك في ticket.submitted، ويستدعي نموذجًا لغويًا لتصنيف الأولوية وصياغة ردّ، ويحفظ النتيجة في الحالة، ويحدّث البثّ.

import { EventConfig, Handlers } from 'motia'
import { z } from 'zod'
import OpenAI from 'openai'
 
const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY })
 
export const config: EventConfig = {
  type: 'event',
  name: 'TriageTicket',
  description: 'يُصنّف التذكرة ويصوغ ردًّا باستخدام نموذج لغوي',
  subscribes: ['ticket.submitted'],
  // بعد الفرز، سلّم إلى استخراج الكلمات المفتاحية (خطوة Python)
  emits: ['ticket.triaged'],
  input: z.object({
    ticketId: z.string(),
    subject: z.string(),
    body: z.string(),
    email: z.string(),
  }),
  flows: ['ticket-triage'],
}
 
export const handler: Handlers['TriageTicket'] = async (input, { emit, logger, state, streams }) => {
  const { ticketId, subject, body } = input
 
  // ادفع حالة وسيطة إلى العميل
  await streams.ticketStatus.set(ticketId, 'status', {
    ticketId,
    stage: 'analyzing',
    priority: null,
    draftReply: null,
  })
 
  const prompt = `You are a support triage assistant. Read the ticket and reply with ONLY JSON:
{"priority":"low|medium|high|urgent","category":"string","draftReply":"a polite 2-sentence reply"}
 
Subject: ${subject}
Body: ${body}`
 
  const completion = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    messages: [{ role: 'user', content: prompt }],
    response_format: { type: 'json_object' },
  })
 
  const result = JSON.parse(completion.choices[0]?.message?.content || '{}')
  logger.info('Ticket triaged', { ticketId, priority: result.priority })
 
  // احفظ النتيجة المُهيكلة. state.set(groupId, key, value)
  await state.set('tickets', ticketId, {
    ...input,
    ...result,
    triagedAt: new Date().toISOString(),
  })
 
  // حدّث البثّ اللحظي كي يرى العميل الإجابة النهائية
  await streams.ticketStatus.set(ticketId, 'status', {
    ticketId,
    stage: 'triaged',
    priority: result.priority,
    draftReply: result.draftReply,
  })
 
  // تابع خط المعالجة: استخراج الكلمات المفتاحية يحدث في Python
  await emit({
    topic: 'ticket.triaged',
    data: { ticketId, text: `${subject} ${body}` },
  })
}

أمران يجعلان هذا جاهزًا للإنتاج دون عمل إضافي. أولًا، input هو مخطط Zod، لذا تُتحقَّق حمولة الحدث وتُكتَب أنواعها بالكامل داخل الـ handler. ثانيًا، إذا فشل استدعاء OpenAI، يُعيد Motia تنفيذ الحدث تلقائيًا بناءً على سياسة إعادة المحاولة للخطوة — فلا تخسر التذاكر بسبب خطأ 503 عابر.

تكتب state.set('tickets', ticketId, value) إلى مخزن مفتاح-قيمة مُجمّع ومستمر. الوسيط الأول هو المجموعة، والثاني هو المفتاح. سنقرأ المجموعة كاملة في خطوة cron.

الخطوة 5: البث اللحظي

كنا نكتب إلى streams.ticketStatus دون تعريفها. الـ stream قناة لحظية مُسمّاة ومُتحقَّق منها عبر مخطط. عرّفها في ملف .stream.ts: أنشئ steps/ticket-status.stream.ts.

import { StateStreamConfig } from 'motia'
import { z } from 'zod'
 
export const config: StateStreamConfig = {
  name: 'ticketStatus',
  schema: z.object({
    ticketId: z.string(),
    stage: z.string(),
    priority: z.string().nullable(),
    draftReply: z.string().nullable(),
  }),
}

هذا هو الإعداد بكامله. يمكن لأي خطوة الآن استدعاء streams.ticketStatus.set(groupId, itemId, data) لبث تحديث، ويمكن لأي عميل الاشتراك عبر WebSocket لتلقّي التغييرات لحظيًا. في خط معالجتنا، يرى المتصفّح received ← analyzing ← triaged تُدفع تلقائيًا مع تشغيل كل خطوة — دون استطلاع، ودون أن تكتب سطرًا واحدًا من كود WebSocket.

تُطابق واجهة الـ streams واجهة الـ state: set(groupId, itemId, data) للكتابة، وdelete(groupId, itemId) لإزالة عنصر.

الخطوة 6: أضف خطوة Python (متعددة اللغات)

هنا يتميّز Motia عن الأطر أحادية اللغة. يمكن لسير العمل نفسه مزج اللغات: أبقِ واجهة API والتنسيق في TypeScript، وانتقل إلى Python حيث تكون منظومتها أقوى — معالجة اللغة الطبيعية، علم البيانات، التعلّم الآلي.

أنشئ steps/extract_keywords_step.py. لاحظ اصطلاح تسمية Python: ينتهي اسم الملف بـ _step.py.

import re
from collections import Counter
 
config = {
    "type": "event",
    "name": "ExtractKeywords",
    "description": "يستخرج أبرز الكلمات المفتاحية من تذكرة باستخدام Python الأصلية",
    "subscribes": ["ticket.triaged"],
    "emits": [],
    "flows": ["ticket-triage"],
}
 
STOPWORDS = {"the", "a", "an", "to", "is", "it", "and", "i", "my", "of", "for"}
 
async def handler(input_data, context):
    ticket_id = input_data.get("ticketId")
    text = input_data.get("text", "").lower()
 
    words = [w for w in re.findall(r"[a-z]{3,}", text) if w not in STOPWORDS]
    top = [word for word, _ in Counter(words).most_common(5)]
 
    context.logger.info("Keywords extracted", {"ticketId": ticket_id, "keywords": top})
 
    # ادمج الكلمات المفتاحية في سجل التذكرة الموجود في الحالة المشتركة
    existing = await context.state.get("tickets", ticket_id) or {}
    existing["keywords"] = top
    await context.state.set("tickets", ticket_id, existing)

عند تشغيل npm run dev، يكتشف Motia خطوة Python، ويُهيّئ لها بيئة معزولة، ويربطها في السير نفسه. يقرأ ويكتب handler الخاص بـ Python في نفس مخزن الحالة state بالضبط الذي استخدمته خطوة TypeScript — يُرجع state.get("tickets", ticket_id) ما كتبته خطوة الفرز في TypeScript. مشاركة البيانات عبر اللغات تأتي مجانًا لأن الحالة جزء من الإطار، لا من كودك.

إذا احتاجت خطوة Python إلى حزم خارجية، أضف ملف requirements.txt إلى جذر المشروع وسيثبّتها Motia في بيئة الخطوة.

الخطوة 7: خطوة Cron مجدولة

أخيرًا، ملخص يومي. أنشئ steps/daily-digest.step.ts. لا تحتاج خطوة cron إلى حدث إطلاق — تعمل على جدول تحدّده بصيغة cron القياسية.

import { CronConfig, Handlers } from 'motia'
 
export const config: CronConfig = {
  type: 'cron',
  name: 'DailyDigest',
  description: 'يُلخّص تذاكر اليوم كل صباح في التاسعة',
  cron: '0 9 * * *', // كل يوم في الساعة 09:00
  emits: ['digest.ready'],
  flows: ['ticket-triage'],
}
 
export const handler: Handlers['DailyDigest'] = async ({ emit, state, logger }) => {
  // اقرأ كل تذكرة من مجموعة 'tickets'
  const tickets = await state.getGroup('tickets')
 
  const byPriority = tickets.reduce((acc: Record<string, number>, t: any) => {
    const p = t.priority || 'unknown'
    acc[p] = (acc[p] || 0) + 1
    return acc
  }, {})
 
  const digest = {
    date: new Date().toISOString().slice(0, 10),
    total: tickets.length,
    byPriority,
  }
 
  logger.info('Daily digest generated', digest)
  await emit({ topic: 'digest.ready', data: digest })
}

يُرجع state.getGroup('tickets') كل قيمة مخزّنة تحت مجموعة tickets كمصفوفة — السجلات التي كتبتها خطوتا TypeScript وPython معًا. من هنا يمكنك إطلاق digest.ready إلى خطوة حدث تُرسل الملخص بالبريد أو تنشره في Slack. كل قدرة جديدة هي مجرد Step صغير آخر.

اختبار التنفيذ

مع تشغيل npm run dev، أرسل تذكرة من طرفية أخرى:

curl -X POST http://localhost:3000/tickets \
  -H "Content-Type: application/json" \
  -d '{
    "subject": "Cannot log in after password reset",
    "body": "I reset my password but the login page keeps rejecting it. This is blocking my work.",
    "email": "user@example.com"
  }'

ينبغي أن تتلقّى فورًا:

{ "ticketId": "a1b2c3d4-...", "status": "received" }

افتح الآن الـ Workbench على http://localhost:3000. سترى سير ticket-triage مرسومًا كرسم بياني متصل: خطوة API تتدفّق إلى خطوة حدث الفرز، التي تتفرّع إلى خطوة الكلمات المفتاحية في Python. انقر على أي تنفيذ لترى سجلات كل خطوة، وحمولات الأحداث، وزمن استجابة النموذج اللغوي، وعمليات كتابة الحالة — جميعها مترابطة بنفس الأثر. هذه هي المراقبة التي تجمعها عادةً من ثلاث أدوات منفصلة.

للتأكد من تشغيل النموذج اللغوي، تحقّق من السجلات بحثًا عن إدخال Ticket triaged مع الأولوية المُسندة، وافحص مجموعة الحالة tickets في عارض الحالة بالـ Workbench.

النشر

عندما تكون جاهزًا للإطلاق، ابنِ وانشر إلى Motia Cloud:

npm run build
npx motia cloud deploy \
  --api-key "YOUR_MOTIA_API_KEY" \
  --version-name "v1.0.0" \
  --env-file .env

كما يُحاوَى Motia في الحاويات بسلاسة — يمكنك نشر المشروع نفسه على أي منصة تشغّل Docker، ما دام مجلد steps/ و(لـ Python) ملف requirements.txt أو pyproject.toml يُشحَن معه. وقت التشغيل الذي يُشغّل npm run dev هو نفسه الذي يعمل في الإنتاج، فلا مفاجآت بين البيئات.

استكشاف الأخطاء وإصلاحها

خطوتي لا تظهر في الـ Workbench. تحقّق من اسم الملف. يجب أن تنتهي خطوات TypeScript بـ .step.ts، وخطوات Python بـ _step.py، ويجب أن تعيش داخل مجلد steps/. راقب طرفية خادم التطوير بحثًا عن سطر [CREATED] Step يؤكّد الاكتشاف.

خطوة الحدث لا تُطلَق أبدًا. تأكّد من أن موضوع emits في المُنتِج يطابق تمامًا موضوع subscribes في المُستهلِك — فهي سلاسل نصية وينبغي أن تكون متطابقة. سيُظهر رسم الـ Workbench عقدة غير متصلة إذا لم يكن للموضوع مُشترِك.

تحقّق Zod يرفض طلبات صحيحة. تأكّد من أن العميل يرسل Content-Type: application/json. يعمل bodySchema قبل الـ handler، لذا يُرجع عدم التطابق رمز 400 مع خطأ التحقق في جسم الاستجابة.

خطوة Python لا تستطيع استيراد حزمة. أضفها إلى requirements.txt في جذر المشروع وأعد تشغيل npm run dev كي يعيد Motia بناء بيئة Python.

قراءات الحالة تُرجع null عبر اللغات. تحقّق مرة أخرى من استخدامك نفس اسم المجموعة والمفتاح في كلتا اللغتين — 'tickets' والـ ticketId. الحالة مشتركة، لكن فقط عندما تتطابق المعرّفات تمامًا.

الخطوات التالية

أصبح لديك الآن خلفية ذكاء اصطناعي عاملة، متعددة اللغات، قائمة على الأحداث، بطوابير مدمجة وإعادة محاولة وبث لحظي وجدولة ومراقبة — وكل ذلك مجرد مجلد من ملفات Step صغيرة. لتوسيعها:

• أضف خطوة حدث ثانية تشترك في ticket.triaged وتردّ تلقائيًا على التذاكر urgent.
• استبدل استدعاء OpenAI بنموذج محلي واقرأ دليلنا حول تشغيل النماذج اللغوية المحلية في الإنتاج باستخدام vLLM.
• أضف موافقة بشرية قبل إرسال الردود المصاغة بالذكاء الاصطناعي.
• استكشف نوع خطوة noop في Motia لنمذجة الخطوات الخارجية/اليدوية في مخطط السير.

لأنماط وكلاء أعمق، قارن هذا المنهج القائم على الأحداث مع نمط ReAct لوكلاء الذكاء الاصطناعي باستخدام Vercel AI SDK.

الخلاصة

رهان Motia هو أن تجزّؤ الخلفية حادثة تاريخية، لا ضرورة. فبدمج واجهات API والمهام والجداول والبث ووكلاء الذكاء الاصطناعي في وحدة Step واحدة — والسماح بكتابة هذه الخطوات بأي لغة تناسب المهمة — يزيل كمًّا هائلًا من الكود الرابط ومن السطح التشغيلي. لقد بنيت خط فرز ذكاء اصطناعي كاملًا كان سيتطلّب في المكدّس التقليدي خادم Express، وعامل BullMQ، وعملية node-cron، وبوابة WebSocket، وخدمة Python مصغّرة، كلٌّ منها يُنشر ويُراقب على حدة. هنا كان خمسة ملفات في مجلد واحد.

الـ Step بالنسبة للخلفية هو ما صار المكوّن (component) بالنسبة للواجهة الأمامية: وحدة صغيرة بما يكفي للتفكير فيها، قابلة للتركيب بما يكفي لبناء أي شيء. ابدأ صغيرًا، أطلِق حدثًا، ودع الإطار يتولّى الأجزاء الصعبة.