التنسيق متعدد الوكلاء باستخدام Temporal

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

تحدي إدارة الحالة في سير عمل الذكاء الاصطناعي

في سيناريو متعدد الوكلاء نموذجي، قد يقوم الوكيل A باسترداد البيانات، ويقوم الوكيل B بتحليلها، ويولد الوكيل C تقريرًا. إذا تعطل الوكيل B بعد عشر دقائق من المعالجة، فلا ترغب في إعادة تشغيل العملية بأكملها. تفقد أطر عمل الوكلاء التقليدية القائمة على الذاكرة (In-memory) حالتها عند إعادة التشغيل، مما يجعل النشر في بيئة الإنتاج محفوفًا بالمخاطر.

تحل Temporal هذه المشكلة من خلال معالجة تنفيذ سير العمل كآلة حالة دائمة وقابلة لإعادة التشغيل (Replayable). يتيح لك ذلك كتابة كود Python قياسي يعمل كمنسق (Orchestrator)، بينما تتولى Temporal مهمة الاستمرارية، وإعادة المحاولة، ومصادر الأحداث (Event Sourcing) خلف الكواليس.

تنفيذ سير عمل موثوق

لإظهار ذلك، دعنا ننشئ سير عمل بسيطًا متعدد الوكلاء حيث لدينا وكيل "باحث" ووكيل "كاتب". سنستخدم SDK الخاص بـ Temporal للغة Python لتعريف هذه المكونات.

1. تعريف الأنشطة (Activities)

الأنشطة هي وحدات العمل. في حالتنا، تمثل هذه الوحدات العمل الشاق الذي تقوم به وكلاء الذكاء الاصطناعي الأفراد.

import temporalio.activity
from temporalio import workflow

# Mocking AI agent interactions
@temporalio.activity.defn
async def research_topic(topic: str) -> dict:
    """Simulates the Researcher Agent"""
    await workflow.sleep(2)  # Simulate network latency
    return {"findings": f"Data for {topic} gathered."}

@temporalio.activity.defn
async def write_report(findings: dict) -> str:
    """Simulates the Writer Agent"""
    await workflow.sleep(2)  # Simulate LLM generation
    return f"Report based on: {findings['findings']}"

2. تنسيق سير العمل

تقوم دالة سير العمل بتنسيق هذه الأنشطة. لاحظ كيف نقوم باستدعاء الأنشطة بشكل تسلسلي ببساطة. تقوم Temporal تلقائيًا بتسجيل التقدم. إذا انقطع سير العمل أثناء `write_report`، فستقوم Temporal بإعادة تشغيل سير العمل، ولكن سيتم تخطي نشاط `research_topic` لأن نتيجته محفوظة بالفعل.

@workflow.defn
class MultiAgentPipeline:
    @workflow.run
    async def run(self, topic: str) -> str:
        # Step 1: Research
        findings = await workflow.execute_activity(
            research_topic,
            args=[topic],
            schedule_to_close_timeout=timedelta(minutes=10),
        )
        
        # Step 2: Write
        report = await workflow.execute_activity(
            write_report,
            args=[findings],
            schedule_to_close_timeout=timedelta(minutes=10),
        )
        
        return report

لماذا يهم ذلك في إنتاج الذكاء الاصطناعي

من خلال دمج Temporal في مجموعة أدوات الوكلاء المتعددة الخاصة بك، تحصل على عدة مزايا حرجة:

• المتانة (Durability): تبقى حالات الوكلاء الخاصة بك صامدة أمام إعادة تشغيل الخوادم وانقطاعات الشبكة.
• إعادة المحاولة مع تأخير (Retries with Backoff): يمكن التعامل مع فشل واجهات برمجة التطبيقات (API) المؤقتة لنماذج اللغات الكبيرة (LLM) تلقائيًا باستخدام استراتيجية التأخير الأسي.
• القابلية للملاحظة (Observability): يوفر واجهة مستخدم Temporal تتبعًا مرئيًا لتنفيذ كل وكيل، مما يتيح لك تصحيح أخطاء التبعيات المعقدة بين الوكلاء بسهولة.
• القابلية للتكوين (Composability): يمكنك تشعب سير العمل (Nest workflows)، مما يسمح لـ "وكيل مدير" بتنسيق عدة سير عمل فرعي.

الخاتمة

مع نمو تعقيد تطبيقات الذكاء الاصطناعي، يجب أن يكون الهيكل الهندسي الداعم لها بنفس القوة والموثوقية. يسمح الجمع بين مرونة أطر عمل وكلاء الذكاء الاصطناعي وموثوقية Temporal للمطورين ببناء أنظمة ليست فقط ذكية، بل أيضًا على مستوى المؤسسات (Enterprise-grade). ابدأ بدمج سير العمل الدائم في خطوط أنابيب الوكلاء الخاصة بك اليوم لضمان قدرة تطبيقات الذكاء الاصطناعي على تحمل عدم القدرة على التنبؤ بالاستخدام في العالم الحقيقي.