البروتوكول الذي نظم بنية وكيلنا

البروتوكول الذي نظّم بنية وكلائنا الذكيين

غالبًا ما يبدو بناء أنظمة متعددة الوكلاء مثل محاولة جمع القطط. لكل وكيل طريقته الخاصة في التواصل، وتنسيق بياناته الخاص، وفكرته الخاصة عن ماهية "الطلب". ومع نمو بنية وكلائنا من حفنة من البوتات المتخصصة إلى نظام بيئي مترامي يضم أكثر من 50 عاملاً مستقلاً، اصطدمنا بجدار. أصبح تعقيد التواصل بين الوكلاء يطغى على نظامنا. تحول تصحيح الأخطاء إلى كابوس، وارتفعت زمن الاستجابة، وأصبحت إضافة وكيل جديد تتطلب أيامًا من العمل على التكامل.

لم يأتِ الحل من بناء وكيل أذكى، بل من اعتماد بروتوكول بسيط وموحد. يشرح هذا المقال البروتوكول الدقيق الذي نظّم بنيتنا، وعملية التثبيت خطوة بخطوة، وأمثلة عملية يمكنك تطبيقها اليوم.

المشكلة: انتشار الوكلاء دون لغة مشتركة

في أوائل عام 2024، كان فريقنا يدير وكلاء لاستخراج البيانات، والتلخيص، ومراجعة الكود، وفرز دعم العملاء، وتوليد المحتوى. كل وكيل بُني بواسطة فريق مختلف باستخدام أطر عمل مختلفة. استخدم البعض استدعاء الدوال من OpenAI، وآخرون استخدام الأدوات من Anthropic، وقليلون استخدموا نقاط نهاية REST مخصصة.

النتيجة؟ شبكة متشابكة من التكاملات النقطية. كان الوكيل (أ) يرسل JSON إلى الوكيل (ب)، لكن الوكيل (ب) كان يتوقع XML. الوكيل (ج) كان ينتظر استدعاءً من الوكيل (د)، لكن المهلة الزمنية كانت مضبوطة بشكل خاطئ. كل وكيل جديد كان يتطلب محولاً مخصصًا. كانت بنيتنا هشة، غير شفافة، وبطيئة.

كنا بحاجة إلى لغة عالمية يمكن لكل وكيل التحدث بها. كنا بحاجة إلى بروتوكول.

البروتوكول الذي غيّر كل شيء

البروتوكول الذي اعتمدناه مبني على ثلاثة مبادئ أساسية:

1. **غلاف رسالة موحد** – كل رسالة، بغض النظر عن محتواها، تُغلف في رأس قياسي مع بيانات وصفية (المرسل، المستلم، النوع، الطابع الزمني، معرف التتبع). 2. **حمولات مصنفة** – يستخدم محتوى الجسم سجل مخططات مشتركًا بحيث يعرف كل وكيل بالضبط الحقول المتوقعة. 3. **توجيه غير متزامن** – يتواصل الوكلاء عبر وسيط رسائل خفيف الوزن، وليس عبر استدعاءات HTTP مباشرة. هذا يفصل المرسلين عن المستقبلين ويتيح إعادة المحاولة، والطوابير، وتوزيع الأحمال.

هذا النهج مستوحى من أنماط مستخدمة في الأنظمة الموزعة واسعة النطاق، لكنه مُكيَّف للعالم الوكيلي. البروتوكول الذي طبقناه يسمى **بروتوكول التواصل بين الوكلاء (ACP)**، وهو معيار مفتوح بنيناه فوق بنية تحتية موثوقة موجودة.

المتطلبات

قبل البدء، تأكد من أن بيئتك تلبي هذه المتطلبات:

• **Python 3.10+** – يستخدم تطبيق البروتوكول ميزات غير متزامنة حديثة.
• **Redis 7+** – يُستخدم كوسيط رسائل للتوجيه والطوابير.
• **pip** – لتثبيت التبعيات.
• **وكيل واحد على الأقل** – سكريبت أو خدمة بسيطة تريد توصيلها.
• **إلمام أساسي بـ async Python** – نستخدم `asyncio` و `aiohttp`.

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

1. تثبيت مكتبة البروتوكول

المكتبة الأساسية هي `acp-py`. قم بتثبيتها باستخدام pip:

pip install acp-py

هذا يثبت غلاف الرسالة، عميل سجل المخططات، والموجه المدعوم بـ Redis.

2. تثبيت وتشغيل Redis

إذا لم يكن Redis قيد التشغيل، قم بتثبيته. على Ubuntu:

sudo apt update
sudo apt install redis-server
sudo systemctl start redis-server

تحقق من تشغيله:

redis-cli ping
# يجب أن يعيد PONG

3. تكوين إعدادات البروتوكول

أنشئ ملف تكوين `acp_config.yaml` في جذر مشروعك:

broker:
  type: redis
  host: localhost
  port: 6379
  queue_prefix: "acp:queue:"

schema_registry:
  type: local
  path: "./schemas/"

agent:
  name: "my-agent"
  version: "1.0.0"
  capabilities:
    - text-processing
    - data-extraction

هذا يخبر البروتوكول بمكان العثور على الوسيط، ومن أين يتم تحميل المخططات، وما يمكن لوكيلك فعله.

4. إنشاء ملف مخطط

حدد مخطط رسالة بسيط في `./schemas/extract_request.json`:

{
  "type": "object",
  "properties": {
    "text": {"type": "string"},
    "max_tokens": {"type": "integer", "default": 500}
  },
  "required": ["text"]
}

يضمن هذا المخطط أن كل رسالة `extract_request` تحتوي على حقل `text` واختياريًا حقل `max_tokens`.

5. تهيئة البروتوكول في وكيلك

أنشئ سكريبت Python `agent.py`:

import asyncio
from acp import Agent, Message

async def handle_extract(msg: Message):
    """معالجة طلب استخراج."""
    text = msg.payload["text"]
    max_tokens = msg.payload.get("max_tokens", 500)
    result = f"تم استخراج {len(text.split())} كلمة، محدودة بـ {max_tokens} رمز."
    
    # إرسال رد باستخدام نفس تتبع الرسالة
    reply = msg.reply(payload={"result": result, "status": "ok"})
    await agent.send(reply)

async def main():
    global agent
    agent = Agent.from_config("acp_config.yaml")
    agent.register_handler("extract_request", handle_extract)
    await agent.start()
    print("الوكيل يستمع على الطابور...")
    await asyncio.Event().wait()  # التشغيل للأبد

if __name__ == "__main__":
    asyncio.run(main())

قم بتشغيله:

python agent.py

وكيلك الآن يستمع للرسائل على طابور Redis.

أمثلة الاستخدام

مثال 1: إرسال طلب من وكيل آخر

أنشئ وكيلًا ثانيًا `requester.py` يرسل رسالة إلى الوكيل الأول:

import asyncio
from acp import Agent, Message

async def main():
    requester = Agent.from_config("acp_config.yaml", name="requester-agent")
    await requester.start()
    
    # بناء رسالة صالحة باستخدام المخطط
    msg = Message(
        target="my-agent",
        type="extract_request",
        payload={"text": "مرحبًا بالعالم، هذا اختبار استخراج.", "max_tokens": 100}
    )
    
    # إرسال وانتظار الرد
    reply = await requester.send_and_wait(msg, timeout=10.0)
    print(f"تم استلام الرد: {reply.payload}")

if __name__ == "__main__":
    asyncio.run(main())

قم بتشغيله:

python requester.py

سترى الرد مطبوعًا: `تم استلام الرد: {'result': 'تم استخراج 4 كلمات، محدودة بـ 100 رمز.', 'status': 'ok'}`

مثال 2: بث رسالة إلى وكلاء متعددين

أحيانًا تريد أن يستقبل جميع الوكلاء ذوي قدرة معينة رسالة. استخدم طريقة `broadcast`:

from acp import Agent, Message

async def broadcast_example():
    broadcaster = Agent.from_config("acp_config.yaml", name="broadcaster")
    await broadcaster.start()
    
    # إرسال إلى جميع الوكلاء ذوي قدرة "text-processing"
    msg = Message(
        target="*",  # بطاقة بدل للبث
        type="status_request",
        payload={"query": "health"}
    )
    # البث يعيد قاموسًا باسم_الوكيل -> رد
    replies = await broadcaster.broadcast(msg, capability_filter="text-processing", timeout=5.0)
    for agent_name, reply in replies.items():
        print(f"{agent_name}: {reply.payload}")

مثال 3: إضافة معالج أخطاء التحقق من المخطط

يقوم البروتوكول تلقائيًا بالتحقق من الحمولات مقابل المخططات. إذا فشل التحقق، يتلقى المرسل رسالة خطأ. يمكنك تخصيص هذا:

from acp import SchemaValidationError

async def handle_validation_error(msg: Message, error: SchemaValidationError):
    error_reply = msg.reply(
        payload={"error": f"فشل التحقق: {error}", "status": "error"}
    )
    await agent.send(error_reply)

agent.register_validation_error_handler(handle_validation_error)

هذا يحول انتهاكات المخطط إلى استجابات خطأ منظمة بدلاً من الإخفاقات الصامتة.

كيف نظف هذا البروتوكول بنيتنا

قبل البروتوكول، كانت بنيتنا تحتوي على:

• 15 تنسيق رسالة مختلف
• أكثر من 30 محولاً مخصصًا
• 40% من كود الوكيل مخصص للتحليل والتنسيق
• متوسط زمن استجابة 2.3 ثانية لكل استدعاء بين الوكلاء
• تصحيح الأخطاء يتطلب تتبع 6 تنسيقات سجلات مختلفة

بعد اعتماد ACP:

• جميع الوكلاء يتحدثون تنسيقًا واحدًا
• صفر محولات بين الوكلاء
• كود الوكيل أصغر بنسبة 60% (لا يوجد منطق تحليل)
• انخفض متوسط زمن الاستجابة إلى 0.4 ثانية (وسيط غير متزامن مع طوابير)
• كل رسالة لها معرف تتبع، مما يجعل تصحيح الأخطاء تافهًا

البروتوكول أيضًا مكّن ميزات لم نخطط لها:

• **إعادة تشغيل الرسائل** – يمكننا إعادة تشغيل الرسائل الفاشلة لتصحيح الأخطاء أو إعادة التدريب.
• **إصدار المخططات** – الوكلاء القدامى ما زالوا يعملون مع المخططات الجديدة عبر حقول متوافقة مع الإصدارات السابقة.
• **المراقبة** – كل رسالة تحمل بيانات وصفية تغذي لوحات مراقبتنا.

الدروس المستفادة

1. **ابدأ بسجل مخططات مبكرًا.** حتى لو كان لديك وكيلان فقط، حدد مخططات مشتركة. يمنع الانحراف. 2. **استخدم التواصل غير المتزامن.** الاستدعاءات المتزامنة بين الوكلاء تخلق اقترانًا وثيقًا. الوسيط المدعوم بـ Redis أعطانا المرونة. 3. **لا تفرط في هندسة البروتوكول.** حاولت نسختنا الأولى دعم كل حالة حافة. النسخة المبسطة (غلاف + حمولات مصنفة + توجيه غير متزامن) غطت 95% من الاحتياجات. 4. **اختبر مع تفعيل التحقق من المخطط.** يكتشف الأخطاء قبل أن تصبح إخفاقات في وقت التشغيل.

الخاتمة

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

إذا كان نظامك البيئي للوكلاء ينمو أسرع من قدرتك على إدارته، فكر في اعتماد بروتوكول مماثل. التثبيت يستغرق 15 دقيقة، لكن العائد في تقليل التعقيد وزيادة السرعة هائل. وكلاؤنا الآن يقضون وقتهم في فعل ما يجيدونه—معالجة البيانات—بدلاً من التصارع على تنسيقات رسائل بعضهم البعض.

البروتوكول لم ينظف بنيتنا فحسب. بل جعل وكلاءنا يعملون معًا كفريق منسق جيدًا. وهذا، في النهاية، هو ما يجب أن تدور حوله الأنظمة متعددة الوكلاء.