بناء AI Pipeline في Laravel: من Demo إلى نظام إنتاج

أول Demo للذكاء الاصطناعي داخل Laravel يكون عادةً سطرين، ويعمل بشكل ممتاز. المشكلة تبدأ عندما يدخل المشروع إلى الإنتاج ويتحوّل العدد من عشرة طلبات إلى عشرة آلاف: طلبات بطيئة، أخطاء 429، عمّال معلّقون، Jobs مكررة، وفاتورة API تتضاعف بلا تفسير واضح. في هذه المرحلة لم يعد السؤال «كيف أرسل Prompt؟» بل «كيف أبني Pipeline تتحمّل آلاف العمليات دون أن ينهار التطبيق أو تنفجر التكلفة؟».

هذا المقال دليل عملي لبناء تلك الطبقة: من نقل المعالجة إلى الخلفية، إلى التحكم في التدفّق والتكلفة والفشل والمراقبة.

الخلاصة في ثلاثة أسطر: عامِل معالجة الذكاء الاصطناعي كـ Background Workload لا كجزء من الـ HTTP Request. اجعل الـ Queue هي نقطة التحكم في التزامن والتكلفة والفشل. لا تعتبر أي Job جاهزة للإنتاج قبل أن تكون idempotent، ومحدودة المعدّل، ومحدودة الميزانية، وقابلة للمراقبة.

• لمن هذا المقال: مطوّرو Laravel الذين يشغّلون — أو على وشك أن يشغّلوا — ميزات AI أمام مستخدمين حقيقيين.
• النسخ المستخدمة: Laravel 12/13، وLaravel AI SDK (laravel/ai) الذي صدر في فبراير 2026 واستقر مع Laravel 13. بعض التفاصيل — خصوصًا وحدات القياس في وسائط الـ Middleware — تغيّرت بين الإصدارات، لذا تحقّق من توثيق نسختك.

لماذا تفشل AI Calls داخل HTTP Request

تخيّل هذا الـ Endpoint:

Route::post('/analyze', function (Request $request) {
    return (new LeadAnalyzer)->prompt($request->input('text'));
});

المسار هنا خطّي: المتصفح ينتظر Laravel، وLaravel ينتظر مزوّد الذكاء الاصطناعي، والـ PHP worker يبقى محجوزًا طوال الوقت. إذا استغرق التوليد 15 أو 30 ثانية، فهذه 30 ثانية من العامل مشغولة بلا عمل فعلي — مجرد انتظار على الشبكة.

المشكلة ليست في السرعة فقط، بل في التزامن: مئة طلب متزامن تعني مئة عامل PHP معلّق ومئة استدعاء متزامن للمزوّد. حتى لو كان السيرفر قويًا، أنت هنا سمحت لحركة المرور الخارجية بأن تقرّر نيابةً عنك كم استدعاء AI يحدث في اللحظة نفسها. هذا هو جوهر الخلل المعماري.

الأساس: نقل المعالجة إلى الخلفية

بدل تنفيذ العملية داخل الطلب، ننشئ سجلًا يمثّل المهمة، نرسل Job إلى الطابور، ونعيد استجابة فورية:

flowchart TD
    A[User Request] --> B[Laravel API]
    B --> C[Create AI Task Record]
    C --> D[Dispatch Job]
    D --> E[202 Accepted]
    D --> F[Queue]
    F --> G[Rate Limiter]
    G --> H[AI Provider]
    H --> I[Validation]
    I --> J[Database]
    J --> K[Notify / Broadcast]

$analysis = AiAnalysis::create([
    'user_id' => $request->user()->id,
    'status'  => 'pending',
    'input'   => $request->input('text'),
]);

AnalyzeText::dispatch($analysis->id)->afterCommit();

return response()->json([
    'id'     => $analysis->id,
    'status' => 'processing',
], 202);

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

إنشاء الـ Job

php artisan make:job AnalyzeText

class AnalyzeText implements ShouldQueue
{
    use Queueable;

    public $tries = 5;
    public $timeout = 120;
    public $maxExceptions = 3;

    public function __construct(public int $analysisId) {}

    public function handle(): void
    {
        $analysis = AiAnalysis::findOrFail($this->analysisId);

        $analysis->update(['status' => 'processing']);

        $response = (new TextAnalyzer)->prompt($analysis->input);

        $analysis->update([
            'status' => 'completed',
            'result' => (string) $response,
        ]);
    }
}

لاحظ أننا نمرّر المعرّف لا الكائن الكامل ولا النص الخام. السبب عملي: الـ payload يُخزَّن في Redis أو قاعدة البيانات، وتمرير مستند بحجم ميغابايت داخل مئة ألف Job يعني عبئًا حقيقيًا على الذاكرة والشبكة. المعرّف يبقى بضعة بايتات، والبيانات تُقرأ عند التنفيذ.

فخّ الـ Transaction: لماذا afterCommit() ليست اختيارية

هذا أشهر خطأ في الطوابير، ويظهر عادةً كـ ModelNotFoundException عشوائية في الإنتاج فقط:

DB::transaction(function () use ($request) {
    $analysis = AiAnalysis::create([...]);

    AnalyzeText::dispatch($analysis->id); // ← قد يفشل
});

العامل قد يلتقط الـ Job وينفّذها قبل أن تُثبَّت المعاملة، فلا يجد السجل. الحل إمّا ->afterCommit() عند كل إرسال، أو تفعيله عالميًا في config/queue.php:

'redis' => [
    'driver'       => 'redis',
    'queue'        => env('REDIS_QUEUE', 'default'),
    'retry_after'  => 300,
    'after_commit' => true,
],

على الأجهزة المحلية لن تلاحظ المشكلة إطلاقًا، لأن العامل غالبًا أبطأ من الـ commit. في الإنتاج تحت الضغط، ستلاحظها.

الطريق المختصر: queue() المدمجة في AI SDK

قبل أن تبني Job يدويًا لكل عملية، انتبه إلى أن الـ SDK يوفّر تشغيلًا في الخلفية جاهزًا مع معالجة النجاح والفشل:

(new SalesCoach)
    ->queue($request->input('transcript'))
    ->then(function (AgentResponse $response) {
        // تخزين النتيجة
    })
    ->catch(function (Throwable $e) {
        // تسجيل الفشل
    });

نفس الأمر متاح لتوليد الصور والصوت والتفريغ والـ embeddings عبر ->queue().

متى تستخدم كلًا منهما؟ استخدم queue() المدمجة للعمليات البسيطة ذات الخطوة الواحدة. اكتب Job مخصّصة عندما تحتاج سلوكًا لا يوفّره الـ SDK: مفتاح ShouldBeUnique، أو middleware مخصّص، أو حالات وسيطة متعددة، أو انتماء إلى Bus::batch. القاعدة العملية: ابدأ بالمدمج، وانتقل إلى Job مخصّصة عند أول متطلّب لا يغطّيه.

إيصال النتيجة إلى المستخدم

نقل العمل إلى الخلفية يحلّ مشكلة ويفتح أخرى: كيف يعرف المستخدم أن النتيجة جاهزة؟ أمامك ثلاثة خيارات:

الخيار الثالث يستحق الانتباه، لأنه يعالج مباشرة الشعور بالبطء:

(new SalesCoach)->broadcastOnQueue(
    'Analyze this sales transcript...',
    new Channel("analysis.{$analysis->id}"),
);

هنا تُنفَّذ العملية في الخلفية، وتُبَثّ أحداث الـ stream إلى الواجهة فور توفّرها. المستخدم يرى النص يُكتب بدل أن يرى مؤشّر تحميل صامتًا لأربعين ثانية — والفرق في الإحساس أكبر بكثير من الفرق في الزمن الفعلي.

إذا اخترت Polling، اجعل الفاصل الزمني تصاعديًا (ثانية، ثم ثانيتان، ثم خمس) بدل استعلام كل ثانية بلا نهاية.

تصميم الـ Queues

افصل حسب نوع المهمة

وضع كل شيء في default يعني أن ملف PDF ضخم قد يؤخّر رسالة محادثة يفترض أن تُجاب خلال ثوانٍ. الأفضل تقسيم واضح:

AnalyzeLead::dispatch($leadId)->onQueue('ai-analysis');
ProcessDocument::dispatch($documentId)->onQueue('ai-documents');

تقسيم منطقي شائع: ai-realtime للمحادثة، ai-analysis للتحليل، ai-documents للمستندات، ai-audio للتفريغ، ai-embeddings للتضمينات، ai-bulk للدفعات الكبيرة.

الأولوية

عند تشغيل العامل، ترتيب الطوابير هو ترتيب الأولوية:

php artisan queue:work --queue=ai-realtime,ai-analysis,ai-bulk

سيستنزف Laravel ai-realtime أولًا، ثم ينتقل إلى ما بعدها. هذا وحده يمنع دفعة خلفية من 20,000 سجل من تدمير تجربة المستخدم الحيّ.

العدالة بين المستخدمين

فصل الطوابير حسب نوع المهمة لا يكفي في تطبيقات SaaS متعدّدة المستأجرين. عميل واحد يرفع 5,000 ملف سيملأ ai-documents ويجمّد بقية العملاء خلف دفعته. الحلّ حصر المعدّل لكل مستأجر لا لكل نوع مهمة فقط:

RateLimiter::for('ai-per-tenant', function (object $job) {
    return Limit::perMinute(20)->by('tenant:'.$job->tenantId);
});

بدون هذا، أول عميل ينشر استخدامًا كثيفًا سيبدو للجميع وكأنه تعطّل في النظام.

التحكم في التدفّق

التزامن ليس «كلما زاد كان أفضل»

خمسون عاملًا يعني خمسين استدعاءً متزامنًا. إذا كان المزوّد يسمح بأقل من ذلك، فالنتيجة 429، ثم إعادة محاولة، ثم حمل إضافي، ثم مزيد من 429. في هذه الحالة تقليل العمّال يجعل النظام أسرع فعليًا.

flowchart LR
    A[More Workers] --> B[More 429s]
    B --> C[More Retries]
    C --> D[More Load]
    D --> A

Rate Limiting بشكل صحيح

تعريف الحدّ وحده لا يفعل شيئًا للـ Jobs — يجب ربطه بـ Job Middleware، وهذه نقطة يغفلها كثير من الأمثلة:

// AppServiceProvider::boot()
RateLimiter::for('ai-openai', fn () => Limit::perMinute(100)->by('openai'));
RateLimiter::for('ai-anthropic', fn () => Limit::perMinute(50)->by('anthropic'));

use Illuminate\Queue\Middleware\RateLimitedWithRedis;

public function middleware(): array
{
    return [
        (new RateLimitedWithRedis('ai-openai'))->releaseAfter(30),
    ];
}

ثلاث ملاحظات مهمة:

• RateLimitedWithRedis أدقّ من RateLimited تحت الضغط العالي لأنه يعتمد على عمليات Redis الذرّية.
• releaseAfter() يمنع الـ hot loop الذي يستهلك محاولات الـ Job على لا شيء.
• بشكل افتراضي، إعادة الـ Job إلى الطابور تُحتسب ضمن $tries. إن أردت ألا يُعاقَب الـ Job على ازدحام خارج سيطرته، استخدم ->dontRelease() مع منطق تأجيل خاص بك، أو ارفع $tries مع maxExceptions أقل.

لا تستخدم Limiter واحدًا لكل المزوّدين. لكل مزوّد حدود مختلفة، وقد تختلف داخل المزوّد نفسه حسب النموذج ومستوى الحساب. الشيء نفسه ينطبق على نوع العملية: حدود المحادثة تختلف عن حدود الـ embeddings.

إيقاف النزيف: ThrottlesExceptions

عندما يبدأ المزوّد بالفشل المتكرر، لا معنى لأن تواصل آلاف الـ Jobs ضرب الـ API في اللحظة نفسها:

use Illuminate\Queue\Middleware\ThrottlesExceptions;

public function middleware(): array
{
    return [
        (new ThrottlesExceptions(10, 5 * 60))
            ->by('openai')          // throttle مشترك لكل Jobs نفس المزوّد
            ->backoff(5),
    ];
}

انتبه إلى الوسيط الثاني: وحدته تغيّرت بين إصدارات Laravel (من دقائق إلى ثوانٍ). اكتبها بصيغة 5 * 60 صراحةً لتفادي الالتباس، وتحقّق من توثيق نسختك. الجزء الأهم هو ->by('openai'): بدونه يكون الـ throttle خاصًا بكل Job على حدة، وهو عكس المطلوب تمامًا.

نمط Circuit Breaker

يمكن الذهاب أبعد: إذا أعاد المزوّد 503 عدة مرات متتالية، سجّل حالته كـ degraded مؤقتًا في الـ Cache، ووجّه الـ Jobs التالية إلى مزوّد بديل مباشرة بدل أن يكتشف كل Job العطل بنفسه.

هذه ليست ميزة في الطابور، بل نمط تبنيه في طبقة المزوّد لديك — وهو المكان الطبيعي لـ Agent Middleware كما سنرى لاحقًا.

Backpressure

هذا مفهوم يُهمَل كثيرًا. إذا كنت تستقبل 1,000 مهمة في الدقيقة بينما يستطيع المزوّد معالجة 100 فقط، فالطابور سينمو بمقدار 900 كل دقيقة. بعد عشر دقائق لديك 9,000 مهمة منتظرة، وبعد ساعة لديك مشكلة مختلفة تمامًا.

إضافة عمّال لن تحلّ شيئًا لأن الحدّ عند المزوّد لا عندك. الخيارات الحقيقية:

• رفض أو تأجيل المهام منخفضة الأولوية عند تجاوز عمق معيّن للطابور.
• إبلاغ المستخدم بزمن انتظار تقديري بدل تركه أمام «جارٍ المعالجة».
• توزيع الحمل على مزوّد ثانٍ.
• رفع الحصّة لدى المزوّد.

إذا كان الطابور ينمو باستمرار، فأنت تقبل عملًا أكثر مما تستطيع تسليمه. هذه مشكلة قبول لا مشكلة سعة.

الموثوقية

إعادة المحاولة مع تأخير تصاعدي

فشل المزوّد كثيرًا ما يكون مؤقتًا: مهلة، أو 429، أو 503، أو خطأ شبكة. لكن إعادة المحاولة الفورية تزيد الطين بلّة:

public $tries = 5;

public function backoff(): array
{
    return [5, 30, 120, 300];
}

التأخير التصاعدي يمنح المزوّد وقتًا للتعافي بدل أن يشارك في إغراقه.

ليست كل الأخطاء قابلة لإعادة المحاولة

مفتاح API غير صالح، أو مدخل غير صحيح، أو ملف غير مدعوم — كلها لن تُصلَح بإعادة المحاولة عشر مرات. صنّف الأخطاء بوضوح:

وطبّق ذلك في الكود بدل ترك الطابور يخمّن:

public function handle(): void
{
    try {
        // ...
    } catch (InvalidInputException $e) {
        $this->fail($e); // فشل نهائي، لا إعادة محاولة
    }
}

Failover ليست Retry

الفرق جوهري: إعادة المحاولة تجرّب نفس المزوّد مرة أخرى، بينما الـ Failover ينتقل إلى مزوّد آخر. الـ SDK يدعم الثاني بتمرير مصفوفة مرتّبة:

use Laravel\Ai\Enums\Lab;

$response = (new LeadAnalyzer)->prompt(
    $text,
    provider: [Lab::OpenAI, Lab::Anthropic],
);

في الإنتاج ستستخدم الاثنين معًا، وهنا يكمن الخطر: ثلاثة مزوّدين × خمس محاولات = خمسة عشر استدعاءً محتملًا لطلب واحد. إذا كانت العملية غالية، هذه كارثة تكلفة صامتة.

كلما زاد عدد المزوّدين في السلسلة، قلّل $tries. سلسلة من ثلاثة مزوّدين مع محاولتين للـ Job كافية في معظم الحالات.

انتبه أيضًا إلى أن لكل مزوّد بديل خصائصه: استخدم providerOptions() لتمرير إعدادات مختلفة لكل مزوّد بدل افتراض أنها متطابقة.

Idempotency

أي Job مهمة يجب أن تتحمّل التنفيذ مرتين دون كارثة، لأن التنفيذ المزدوج سيحدث — عند انتهاء مهلة، أو إعادة تشغيل عامل، أو انقطاع شبكة بعد نجاح الاستدعاء وقبل حفظ النتيجة.

// خطأ: ينتج سجلات مكررة عند كل إعادة محاولة
InvoiceAnalysis::create([...]);

// صحيح
InvoiceAnalysis::updateOrCreate(
    ['document_id' => $document->id, 'prompt_version' => 'invoice-v5'],
    ['result' => $result],
);

للعمليات الغالية تحديدًا (توليد الصور مثلًا)، الحماية الحقيقية أعمق من updateOrCreate: سجّل معرّف العملية لدى المزوّد قبل بدء التنفيذ، وتحقّق منه عند إعادة المحاولة. صورة وُلِّدت بنجاح ثم انقطع الاتصال قبل الحفظ ستُولَّد مرة ثانية وتُدفَع مرة ثانية إن لم تفعل ذلك.

Unique Jobs

المستخدم الذي يضغط زر «تحليل» خمس مرات لا يجب أن ينتج خمس عمليات:

class AnalyzeDocument implements ShouldQueue, ShouldBeUniqueUntilProcessing
{
    public $uniqueFor = 3600; // ضروري

    public function __construct(public int $documentId) {}

    public function uniqueId(): string
    {
        return (string) $this->documentId;
    }
}

ثلاث نقاط حاسمة:

• $uniqueFor ليست اختيارية. إن مات العامل وبقي القفل، لن تعمل أي Job لنفس المفتاح مرة أخرى — إلى الأبد. هذه أعطال إنتاج يصعب تشخيصها.
• ShouldBeUniqueUntilProcessing عادةً هي الصحيحة لحالة زر يضغطه المستخدم: يُحرَّر القفل عند بدء المعالجة، فيستطيع طلب تحليل جديد بعد اكتمال السابق. أما ShouldBeUnique فتحتفظ بالقفل حتى نهاية التنفيذ.
• القفل يحتاج مخزن Cache مشترك (Redis). مع file أو array في بيئة متعدّدة الخوادم، الضمان وهمي.

الدفعات: Bus::batch

عند معالجة عشرات الآلاف من السجلات، لا تحمّل كل شيء في الذاكرة:

$jobs = [];

Lead::query()->chunkById(500, function ($leads) use (&$jobs) {
    foreach ($leads as $lead) {
        $jobs[] = new AnalyzeLead($lead->id);
    }
});

Bus::batch($jobs)
    ->name('lead-analysis-v3')
    ->onQueue('ai-bulk')
    ->allowFailures()
    ->then(fn (Batch $batch) => Log::info('اكتملت الدفعة'))
    ->catch(fn (Batch $batch, Throwable $e) => $batch->cancel())
    ->dispatch();

الـ Batch يمنحك ما لا يمنحه الإرسال الفردي: نسبة تقدّم قابلة للعرض ($batch->progress())، وإلغاء جماعي بأمر واحد، ومعالجة موحّدة للنجاح والفشل. والأهم أن cancel() هو أداة حارس التكلفة الحقيقية — عند تجاوز الميزانية تُلغى الدفعة كاملة بدل فحص الميزانية داخل كل Job على حدة.

السلاسل: Bus::chain

عندما تكون العملية متعدّدة المراحل (استخراج ← تضمين ← تحليل ← حفظ)، فهي سلسلة لا مجموعة Jobs مستقلة:

Bus::chain([
    new ExtractDocumentText($documentId),
    new GenerateEmbeddings($documentId),
    new AnalyzeDocument($documentId),
])->onQueue('ai-documents')->dispatch();

الميزة أن فشل أي حلقة يوقف ما بعدها تلقائيًا — بدل أن تحاول مرحلة التحليل العمل على نصّ لم يُستخرج أصلًا.

Failed Jobs

بعد استنفاد المحاولات لا تختفي المشكلة، بل تنتقل إلى failed_jobs:

php artisan queue:failed
php artisan queue:retry {id}
php artisan queue:prune-failed --hours=168

لا تشغّل queue:retry all بشكل أعمى في الإنتاج. إذا كان السبب مدخلًا غير صالح أو ملفًا مفقودًا، ستفشل مرة أخرى — وستدفع مقابل المحاولة.

الأهم أن تُحدِّث حالة الكيان في تطبيقك لا في جدول الطوابير فقط:

public function failed(Throwable $exception): void
{
    AiAnalysis::find($this->analysisId)?->update([
        'status'      => 'failed',
        'error_class' => $this->classifyError($exception),
        'error'       => $exception->getMessage(),
    ]);
}

بدون هذا تبقى الواجهة عالقة على «جارٍ المعالجة» إلى الأبد. مع هذا تعرض «فشل التحليل» مع زر إعادة محاولة.

تصنيف أسباب الفشل

معرفة أن 182 Job فشلت اليوم لا تفيد. معرفة أن 120 منها بسبب حدود المعدّل و34 بسبب ملفات تالفة و8 بسبب الميزانية — هذه معلومة قابلة للتنفيذ. صنّف الأسباب في عمود مستقل: provider_rate_limit، provider_unavailable، invalid_input، budget_exceeded، file_missing، validation_error، unknown.

اعتبر failed_jobs بمثابة Dead Letter Queue: عمليات لم تعد مناسبة لإعادة المحاولة التلقائية، وتحتاج مراجعة بشرية. وفّر زر «إعادة المحاولة» في لوحة الإدارة بدل أن يتطلّب كل حالة اتصال SSH بالخادم.

التحقّق من الناتج قبل اعتباره ناجحًا

استجابة HTTP 200 من المزوّد لا تعني أن النتيجة صالحة. النموذج قد يعيد بنية ناقصة أو تصنيفًا غير موجود في قائمتك أو قيمة خارج المدى المتوقّع.

flowchart LR
    A[AI Response] --> B[Schema Validation]
    B --> C[Application Validation]
    C --> D[Business Rules]
    D --> E[Completed]
    B -.fail.-> F[Retry / Failed]
    C -.fail.-> F
    D -.fail.-> F

استخدام Structured Output عبر HasStructuredOutput يغطّي الطبقة الأولى، لكنه لا يغطّي المنطق: درجة تقييم بين 1 و10 قد تكون صحيحة بنيويًا وخاطئة تجاريًا. تحقّق من الطبقات الثلاث، وقرّر بناءً على نوع الخطأ هل تعيد المحاولة أم تعلن الفشل.

ضبط التكلفة

أكبر خطأ في هذا الباب جملة واحدة: «سنراجع فاتورة الـ AI لاحقًا». التكلفة يجب أن تكون مرئية داخل التطبيق قبل وصول الفاتورة.

سجّل الاستهلاك عبر Events لا يدويًا

بدل استدعاء AiUsage::create() في كل Job — وهو ما ستنساه في مكان ما حتمًا — استمع إلى أحداث الـ SDK مرة واحدة:

// EventServiceProvider
Event::listen(AgentPrompted::class, function ($event) {
    AiUsage::create([
        'user_id'        => auth()->id(),
        'agent'          => $event->agent::class,
        'provider'       => $event->provider,
        'model'          => $event->model,
        'input_tokens'   => $event->response->usage->inputTokens,
        'output_tokens'  => $event->response->usage->outputTokens,
        'estimated_cost' => $this->estimateCost($event),
    ]);
});

الـ SDK يُطلق أحداثًا لكل العمليات: AgentPrompted، AgentStreamed، ImageGenerated، EmbeddingsGenerated، ToolInvoked وغيرها. سطر واحد في مزوّد الخدمة يغطّي التطبيق كله، ويستحيل أن «تنسى» تسجيل عملية جديدة. (تحقّق من أسماء خصائص الحدث في نسختك، فقد تختلف.)

ميزانيات منفصلة لا ميزانية واحدة

عندما تقترب من الحدّ، الإيقاف الشامل هو أسوأ استجابة ممكنة. الأفضل تعطيل ما يمكن تأجيله والإبقاء على ما يراه العميل:

وعند فحص الميزانية داخل Job، لا تستخدم fail(). تجاوز الميزانية ليس خطأً تقنيًا، واستخدام fail() سيملأ failed_jobs بآلاف السجلات التي ليست أعطالًا:

if ($user->aiSpendThisMonth() >= $user->aiBudget()) {
    $this->release(now()->addHour());
    $analysis->update(['status' => 'budget_paused']);

    return;
}

توجيه النماذج حسب التعقيد

تصنيف مشاعر جملة قصيرة لا يحتاج النموذج نفسه الذي يحلّل عقدًا من 70 صفحة. الـ SDK يوفّر هذا كإعداد على مستوى الـ Agent:

#[UseCheapestModel]
class SentimentClassifier implements Agent
{
    use Promptable;
}

#[UseSmartestModel]
class ContractAnalyzer implements Agent
{
    use Promptable;
}

للتوجيه الديناميكي حسب حالة وقت التشغيل — خطة المستخدم مثلًا، أو حجم المستند — عرّف دالة model() أو provider() على الـ Agent، فهي تسبق الـ Attribute في الأولوية.

الرافعة الأكبر: Batch APIs وPrompt Caching

هذه أهم فقرة في القسم كله. تحليل 100,000 سجل بتكلفة سنتين للسجل يساوي 2,000 دولار — لكن هذا الرقم قابل للخفض بأكثر من النصف قبل أن تكتب سطر تحسين واحد:

• Batch APIs: معظم المزوّدين يوفّرون واجهة دفعات بخصم يقارب 50% مقابل زمن تسليم أطول (ساعات بدل ثوانٍ). الدفعات الليلية والتحليل التاريخي هي بالضبط الحالة المستهدفة. إذا كان العمل غير آنيّ ولم تستخدم Batch API، فأنت تدفع ضعف ما يلزم.
• Prompt Caching: نفس تعليمات النظام تتكرر في 100,000 استدعاء. تفعيل التخزين المؤقت للجزء الثابت من الـ Prompt يخفض تكلفة الإدخال بنسبة كبيرة، وهو مدعوم لدى المزوّدين الرئيسيين.
• تخزين الـ Embeddings مؤقتًا: مدمج في الـ SDK، ويكفي تفعيله.

// config/ai.php
'caching' => [
    'embeddings' => ['cache' => true, 'store' => env('CACHE_STORE', 'redis')],
],

تخزين النتائج مؤقتًا وتنسيخ الـ Prompts

إذا طلب المستخدم تلخيص المستند نفسه مرتين دون تغييره، لا داعي لاستدعاء جديد. لكن مفتاح الـ Cache يجب ألّا يعتمد على المحتوى وحده:

$key = hash('sha256', implode('|', [
    $document->content_hash,
    'invoice-analyzer-v5',   // نسخة الـ prompt
    $model,
    $analysisType,
]));

تنسيخ الـ Prompts (lead-score-v3، invoice-analyzer-v5) ليس ترفًا تنظيميًا. عندما تعدّل الـ Prompt، تحتاج أن تعرف أي النتائج وُلِّدت بأي نسخة، وأن تعيد تحليل القديم فقط بدل إعادة كل شيء.

ملاحظة على النصّ العربي

قاعدة «عدد الأحرف ÷ 4 = عدد التوكنز» شائعة، لكنها مشتقّة من الإنجليزية ولا تنطبق على العربية. النص العربي يستهلك توكنز أكثر لكل حرف بفارق ملحوظ، ما يعني أن تقديرات التكلفة المبنية على تلك القاعدة ستكون أقل من الواقع بشكل منهجي.

المخرج العملي: عايِر التقدير على بياناتك الفعلية. سجّل input_tokens الحقيقية من استجابة المزوّد لعيّنة من نصوصك، واحسب المعامل الخاص بك، ثم استخدمه في حسابات الميزانية.

التشغيل: Horizon والمهل والنشر

سلسلة المهل الزمنية

هذه أكثر إعدادات الطوابير تسبّبًا للأعطال الصامتة، ولا بد أن تُضبط كسلسلة مرتّبة:

HTTP timeout (AI SDK، افتراضيًا 60s)
        <
Job timeout ($timeout)
        <
retry_after (config/queue.php)

إذا انعكس الترتيب — أي إذا صار retry_after أقصر من زمن تنفيذ الـ Job — سيعتبر الطابور المهمة منتهية الصلاحية بينما لا يزال العامل يعالجها، وسيلتقطها عامل آخر. النتيجة استدعاء AI مزدوج وتكلفة مضاعفة، وقد تستمر أسابيع دون أن تلاحظ.

اضبط مهلة الـ SDK صراحةً بدل الاعتماد على الافتراضي:

#[Timeout(90)]
class DocumentAnalyzer implements Agent
{
    use Promptable;
}

ثم اجعل $timeout = 120 على الـ Job وretry_after = 300 على الاتصال. عند استخدام Horizon، يجب أن تكون مهلة الـ supervisor أكبر من مهلة الـ Job.

إعداد Horizon

'environments' => [
    'production' => [
        'ai-realtime' => [
            'connection'      => 'redis',
            'queue'           => ['ai-realtime'],
            'balance'         => 'auto',
            'maxProcesses'    => 20,
            'balanceMaxShift' => 3,
            'balanceCooldown' => 3,
            'timeout'         => 90,
            'tries'           => 3,
            'memory'          => 256,
        ],

        'ai-analysis' => [
            'connection'   => 'redis',
            'queue'        => ['ai-analysis'],
            'balance'      => 'auto',
            'maxProcesses' => 10,
            'timeout'      => 180,
            'tries'        => 5,
            'memory'       => 512,
        ],

        'ai-bulk' => [
            'connection'   => 'redis',
            'queue'        => ['ai-bulk'],
            'balance'      => 'simple',
            'maxProcesses' => 3,
            'timeout'      => 300,
            'tries'        => 2,
            'memory'       => 512,
        ],
    ],
],

عشرون عاملًا للطلبات الآنية مقابل ثلاثة للدفعات — لأن الهدف ألّا تبتلع الأعمال الخلفية موارد العملاء. لاحظ أن maxProcesses مع balance: auto هو سقف للـ supervisor لا رقم ثابت، وأن memory مهمّ لأن معالجة المستندات الكبيرة تسرّب ذاكرة بمرور الوقت.

النشر: queue:restart

العمّال عمليات طويلة العمر تحتفظ بالكود في الذاكرة. بعد أي نشر، ستستمر في تشغيل النسخة القديمة حتى تُعاد تشغيلها:

php artisan queue:restart

اجعل هذا جزءًا ثابتًا من سكربت النشر. الأعطال الناتجة عن إهماله من أصعب ما يُشخَّص، لأن الكود المنشور صحيح فعلًا.

التوسّع التلقائي ليس توسّعًا لا نهائيًا

عمق طابور بمقدار 100,000 لا يعني تشغيل 100,000 عامل. maxProcesses ضرورية، لأن الاختناق الحقيقي سيكون عند حدود المزوّد أو قاعدة البيانات أو Redis — لا عند معالج الخادم. أي قرار توسّع يجب أن يحترم أربعة قيود: حدّ المزوّد، والميزانية، وسعة قاعدة البيانات، وسعة Redis.

المراقبة

يوفّر Laravel Pulse مراقبة جاهزة للطوابير والمهام البطيئة والاستثناءات والطلبات الخارجية البطيئة. راقب عبره: عمق الطابور، ومتوسط زمن التنفيذ، والمهام الفاشلة، والمهام البطيئة، والاستدعاءات الخارجية.

لكن هذه مقاييس عامة. المقاييس التي تخصّ الذكاء الاصطناعي تحتاج بناءً خاصًا: عدد الاستدعاءات في الدقيقة، ونسبة 429، ونسبة أخطاء المزوّد، ومتوسط زمن الاستجابة، والتوكنز لكل مهمة، والتكلفة لكل مهمة، والتكلفة لكل مستخدم، ونسبة الانتقال إلى المزوّد البديل، ونسبة إعادة المحاولة.

ارتفاع نسبة الـ Failover تحديدًا هو أفضل مؤشّر مبكّر على أن المزوّد الأساسي يعاني — قبل أن يشتكي المستخدمون بوقت كافٍ للتصرّف.

زمن الانتظار أهم من زمن المعالجة

استدعاء AI يستغرق ثانيتين، لكن المهمة انتظرت في الطابور 45 ثانية. المستخدم يشعر بـ 47 ثانية، لا بثانيتين. لوحات المراقبة التي تعرض زمن التنفيذ فقط ستُظهر لك نظامًا صحّيًا بينما المستخدمون يرون العكس.

Queue Wait + Processing Time = Total Latency

اتفاقيات مستوى خدمة حسب نوع المهمة

ليست كل مهمة تحتاج السرعة نفسها، وتحديد ذلك صراحةً يسهّل تصميم الطوابير:

الأعمال التاريخية بالذات لا داعي لتشغيلها في ذروة الاستخدام:

Schedule::command('ai:process-historical-leads')->dailyAt('02:00');

على أن يقتصر الأمر على إنشاء الـ Batch وإرسالها، لا تنفيذ الاستدعاءات مباشرة.

الأمان والخصوصية

هذا القسم غالبًا ما يُؤجَّل حتى يطرحه العميل أو القسم القانوني — والأفضل حسمه في التصميم:

• تنقية البيانات الحسّاسة قبل الإرسال. أرقام الهوية، وبيانات البطاقات، وأرقام الهواتف، وعناوين البريد: قرّر صراحةً ما يخرج من نظامك. Agent Middleware هو المكان الطبيعي لذلك، لأنه يعترض الـ Prompt قبل وصوله إلى المزوّد.
• لا تسجّل الـ Prompt كاملًا في اللوجات. سجّل بصمة (hash) وطولًا ومعرّفًا. لوجات التطبيق عادةً أقل حماية من قاعدة البيانات، وتُصدَّر إلى خدمات خارجية.
• مدّة الاحتفاظ. حدّد متى تُحذف نتائج التحليل والمحادثات المخزّنة في agent_conversations. النموّ التلقائي بلا سياسة حذف مشكلة تخزين وامتثال معًا.
• صلاحية الاطّلاع. نتيجة تحليل عميل محتمل بيانات تجارية حسّاسة؛ يجب أن تخضع لنفس ضوابط الوصول التي تخضع لها بياناته الأصلية.

عند العمل تحت متطلّبات تنظيمية، وجّه الطلبات عبر بوّابة مؤسسية باستخدام إعداد url في config/ai.php بدل الاتصال المباشر بالمزوّد.

الاختبار

قائمة تحقّق «جاهز للإنتاج» بلا اختبارات ليست مقنعة. المشكلة العملية معروفة: كيف تختبر Pipeline دون دفع فاتورة API عند كل تشغيل؟ الـ SDK يوفّر طبقة تزييف كاملة:

public function test_analysis_is_queued_and_stored(): void
{
    TextAnalyzer::fake(['{"score": 8, "intent": "high"}']);

    $this->postJson('/analyze', ['text' => 'نصّ تجريبي'])
         ->assertStatus(202);

    TextAnalyzer::assertQueued(fn ($prompt) => $prompt->contains('نصّ تجريبي'));
}

ثلاث ممارسات تستحق التبنّي:

1. preventStrayPrompts() في TestCase الأساسي، حتى يفشل الاختبار فورًا إذا حاول أي كود الاتصال بمزوّد حقيقي.
2. عند استخدام Structured Output، يولّد fake() بيانات مطابقة للمخطّط تلقائيًا — ما يجعل اختبار مسارات التحقّق سهلًا.
3. اختبر مسار الفشل لا مسار النجاح فقط: ماذا يحدث عند 429؟ عند تجاوز الميزانية؟ عند نتيجة غير صالحة؟ هذه هي المسارات التي ستعمل في الإنتاج ولم تُجرَّب أبدًا.

مثال متكامل: تحليل 10,000 عميل محتمل

المطلوب استخراج النيّة والمشاعر ودرجة التقييم والإجراء التالي لكل سجل.

flowchart TD
    A[10,000 Leads] --> B[chunkById 500]
    B --> C[Bus::batch]
    C --> D[ai-bulk Queue]
    D --> E[Rate Limiter: 100/min]
    E --> F[Cost Guard]
    F --> G[Primary Provider]
    G -.fail.-> H[Fallback Provider]
    G --> I[Structured Output]
    H --> I
    I --> J[Validation]
    J --> K[CRM]

الحدود المعلَنة مسبقًا:

الفارق الجوهري بين هذا وبين حلقة foreach هو أن كل رقم في هذا الجدول قرار اتخذته أنت. في الحلقة، كل هذه الأرقام تقرّرها حركة المرور والصدفة.

عند بلوغ 95 من أصل 100 دولار، لا يتوقّف النظام كله: تُعلَّق الدفعة التاريخية، ويبقى الذكاء الاصطناعي الموجّه للعملاء يعمل. هذا أفضل بكثير من تعطيل شامل.

المعمارية النهائية

flowchart TD
    A[Client] --> B[Laravel API]
    B --> C[Create AI Task]
    C --> D[Dispatch Job afterCommit]
    D --> E[202 Accepted]
    D --> F[(Redis)]
    F --> G{Queue Layer}
    G --> H[Realtime Workers]
    G --> I[Analysis Workers]
    G --> J[Bulk Workers]
    H --> K[Rate Limiter]
    I --> K
    J --> K
    K --> L[Cost Guard]
    L --> M[Agent Middleware]
    M --> N[Primary Provider]
    N -.fail.-> O[Fallback Provider]
    N --> P[Validation]
    O --> P
    P --> Q[(Database)]
    Q --> R[Broadcast / Notify]
    N -.error.-> S{Retryable?}
    S -.yes.-> T[Backoff & Retry]
    T --> K
    S -.no.-> U[Failed Job]
    U --> V[Admin Review]

Production Checklist

قبل إطلاق أي عبء عمل AI حقيقي، تأكّد من وجود:

• طوابير منفصلة حسب نوع المهمة والتكلفة
• حدود واضحة لعدد العمّال (maxProcesses)
• Rate Limiting مربوط بـ Job Middleware، لكل مزوّد على حدة
• حصر معدّل لكل مستأجر لضمان العدالة
• سلسلة مهل صحيحة: HTTP < Job < retry_after
• afterCommit مفعّل
• إعادة محاولة مع تأخير تصاعدي، وتصنيف للأخطاء القابلة وغير القابلة
• ShouldBeUniqueUntilProcessing مع $uniqueFor
• Idempotency في كل عملية كتابة
• Bus::batch للدفعات، مع إمكانية cancel()
• failed() يُحدّث حالة الكيان، مع تصنيف السبب
• Provider Failover مع تقليل $tries بما يتناسب
• تسجيل الاستهلاك عبر Events
• ميزانيات منفصلة، وتعليق تدريجي لا إيقاف شامل
• Batch API وPrompt Caching للأعمال غير الآنية
• تنسيخ الـ Prompts وربطه بمفاتيح الـ Cache
• تحقّق من الناتج على ثلاث طبقات
• تنقية البيانات الحسّاسة قبل الإرسال
• مراقبة تشمل زمن الانتظار لا زمن التنفيذ فقط
• اختبارات مع fake() وpreventStrayPrompts()
• queue:restart ضمن سكربت النشر
• زر إعادة محاولة يدوية في لوحة الإدارة

إن غابت هذه العناصر، فما لديك ليس AI Pipeline — بل استدعاء API داخل Job.

أسئلة شائعة

هل أحتاج Horizon أم يكفي queue:work؟

queue:work كافٍ للبداية ولحالات الحمل البسيط. Horizon يصبح ضروريًا عندما تحتاج طوابير متعدّدة بأولويات مختلفة، وتوزيعًا تلقائيًا للعمّال، ولوحة تعرض ما يجري فعلًا. مع أعباء AI، ستحتاجه أسرع مما تتوقّع.

كم عاملًا يجب أن أشغّل لمهام الذكاء الاصطناعي؟

ابدأ من حدّ المزوّد لا من موارد الخادم. إذا كان الحدّ 100 استدعاء في الدقيقة ومتوسط الاستدعاء ثلاث ثوانٍ، فخمسة عمّال قد تكون كافية. زيادة العدد فوق ما يسمح به المزوّد تنتج 429 وإعادة محاولات، أي حملًا أكبر بإنتاجية أقل.

لماذا تُنفَّذ بعض الـ Jobs مرتين؟

السبب الأشيع أن retry_after أقصر من زمن تنفيذ الـ Job، فيعتبر الطابور المهمة منتهية ويسلّمها لعامل آخر. تحقّق من سلسلة المهل، واجعل كل عملية idempotent كخط دفاع ثانٍ.

كيف أمنع فاتورة مفاجئة؟

بثلاث طبقات: تسجيل الاستهلاك لحظيًا عبر Events، وميزانيات منفصلة لكل نوع عمل، وتقدير التكلفة قبل الدفعات الكبيرة. ولا تعتمد على فحص الميزانية داخل كل Job وحده — $batch->cancel() أسرع وأرخص.

هل أستخدم queue() من الـ SDK أم Job مخصّصة؟

queue() للعمليات البسيطة ذات الخطوة الواحدة. Job مخصّصة عندما تحتاج مفتاح تفرّد، أو Middleware، أو حالات وسيطة، أو انتماء إلى Batch أو Chain.

كيف أختبر ميزات الذكاء الاصطناعي دون تكلفة؟

Agent::fake() مع assertPrompted وassertQueued، إضافة إلى preventStrayPrompts() لضمان عدم تسرّب أي استدعاء حقيقي في بيئة الاختبار.

ما الفرق بين Retry وFailover؟

Retry يعيد المحاولة على المزوّد نفسه، وFailover ينتقل إلى مزوّد آخر. يمكن استخدامهما معًا، لكن حاصل ضربهما هو ما يحدّد أقصى عدد استدعاءات لطلب واحد — وهذا ما يجب ضبطه بوعي.

الخلاصة

مع عشرة طلبات، يمكنك تشغيل كل شيء مباشرة. مع عشرة آلاف، تصبح المعمارية نفسها هي الميزة أو المشكلة.

الـ Queue ليست أداة لتسريع الاستجابة، بل الطبقة التي تتحكّم عبرها في الحمل والموثوقية والتكلفة والتوفّر.

بدونها، عندما يكبر عبء الذكاء الاصطناعي، لن تكون المشكلة أن Laravel بطيئة — بل أنك سمحت لآلاف الطلبات بأن تقرّر بنفسها متى وكيف وكم مرة تعمل. في الإنتاج، أنت من يجب أن يقرّر ذلك.

مراجع

• Laravel Queues
• Laravel Horizon
• Laravel AI SDK
• Laravel Broadcasting
• Laravel Pulse