AI Agents · درس

بروتوكولات الاتصال (حافلات الرسائل)

افصل الوكلاء باستخدام حافلة رسائل، مثل Redis أو NATS أو Kafka، حتى يتمكنوا من التوسع والتعطل بصورة مستقلة.

بروتوكولات الاتصال (حافلات الرسائل) درس مجاني في AI Agents على CoddyKit. هذا هو الدرس 4 من أصل 4. يمكنك قراءة الدرس كاملاً أدناه مجاناً — ثم تمرن عليه مباشرة في المتصفح باستخدام محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7. هذا الدرس جزء من مسار التعلم في AI Agents، وتقدمك يتزامن عبر الويب وتطبيق CoddyKit. تتضمن دورة AI Agents 4 دروس في المجموع.

بعض أجزاء هذا الدرس لم تُترجم بعد وتظهر باللغة الإنجليزية.

ما بعد المحادثات داخل العملية

في الأنظمة الصغيرة، يستدعي الوكلاء بعضهم بعضًا داخل العملية نفسها. أما في الأنظمة الأكبر، فافصلوا بينهم باستخدام ناقل رسائل — مثل Redis أو NATS أو Kafka.

لماذا نفصل بينهم؟

يمكن توسيع نطاق الوكلاء بشكل مستقل
عزل الأعطال لكل وكيل
سير العمل غير المتزامن، من دون الحاجة إلى انتظار الوكلاء البطيئين ضمن المسار المباشر
إعادة تشغيل الرسائل لتصحيح الأخطاء
التوافق بين اللغات: يمكن لوكيل Python ووكيل Go الاشتراك معًا

Simple Bus: Redis Pub/Sub

import redis
r = redis.Redis()

# Publisher
r.publish('agent.research.task', json.dumps({'task_id': 'abc', 'query': '...'}))

# Subscriber
p = r.pubsub()
p.subscribe('agent.research.task')
for msg in p.listen():
    if msg['type'] == 'message':
        handle_task(json.loads(msg['data']))

Pub/Sub للبث

استخدموه عندما قد يرغب العديد من الوكلاء في معرفة حدث ما، مثل "user-question-received".

قوائم الانتظار لتوزيع العمل

استخدموا قائمة انتظار، مثل Redis BLPOP أو RabbitMQ، عندما يجب أن يعالج عامل واحد بالضبط كل مهمة:

import json, queue

r = queue.Queue()

# Producer
r.put(json.dumps({'task_id': 'abc', 'kind': 'research'}))

def process(task):
    print('processing', task)

# Worker
while not r.empty():
    raw = r.get()
    task = json.loads(raw)
    process(task)

NATS للسرعة

إن NATS هو وسيط رسائل خفيف الوزن صُمم للخدمات المصغرة. ويوفر زمن استجابة أقل من ميلي ثانية، مع دعم مدمج لنمط الطلب/الرد:

import nats
nc = await nats.connect('nats://localhost:4222')

await nc.publish('agent.research', json.dumps(task).encode())

# Request/reply
response = await nc.request('agent.research', payload, timeout=10)

Kafka للمتانة

يضيف Kafka سجلات متينة قابلة لإعادة التشغيل. وهو مثالي للتدقيق وإعادة التشغيل:

from confluent_kafka import Producer, Consumer

producer.produce('agent-events', key=task_id, value=json.dumps(task))
producer.flush()

مخططات الأحداث

عرّفوا بنية كل رسالة باستخدام Pydantic أو Protobuf. فمن دون مخططات، سيتحول ناقل الرسائل إلى فوضى:

class ResearchTask(BaseModel):
    task_id: str
    user_id: str
    query: str
    deadline: datetime

class ResearchResult(BaseModel):
    task_id: str
    findings: list[str]
    duration_ms: int

معرّفات الارتباط

تحمل كل رسالة مرتبطة القيمة نفسها في correlation_id، حتى تتمكنوا من تتبع المهمة عبر الخدمات:

record = {'correlation_id': 'abc', 'task_id': 'def', 'span': 'llm_call'}
print(record)

المعالجات غير التكرارية

قد تُسلَّم الرسائل أكثر من مرة، وفق دلالات التسليم مرة واحدة على الأقل. اجعلوا المعالجات غير تكرارية — بحيث تتم معالجة task_id نفسه مرة واحدة.

قائمة انتظار الرسائل الميتة

عندما يفشل أحد المعالجات بشكل متكرر، انقلوا الرسالة إلى DLQ لمراجعتها يدويًا بدلًا من تكرار المحاولة إلى ما لا نهاية:

if attempts > MAX_RETRIES:
    r.rpush('queue:research:dlq', raw)
    log.error('Sent to DLQ', extra={'task_id': task_id})

قابلية المراقبة

تتبعوا كل رسالة:

تنتقل معرّفات التتبع عبر ناقل الرسائل
تتضمن السجلات task_id وcorrelation_id
المقاييس: معدل الإنتاجية، وزمن الاستجابة، ومعدل الأخطاء لكل موضوع

متى نعتمدها؟

ابدؤوا ببساطة — باستخدام وكلاء داخل العملية. وانتقلوا إلى ناقل رسائل فقط عندما يكون لديكم:

5 وكلاء أو أكثر في بيئة الإنتاج
حاجة إلى توسيع نطاق الوكلاء بشكل مستقل
سير عمل يمتد عبر خدمات متعددة

التسليم مرة واحدة على الأقل؟

ما الذي يتطلبه "التسليم مرة واحدة على الأقل" من معالجاتكم؟

خلاصة

افصلوا المكونات باستخدام Redis للحالات البسيطة، وNATS لزمن الاستجابة المنخفض، وKafka للمتانة. استخدموا المخططات، ومعرّفات الارتباط، والمعالجات غير التكرارية، وDLQ.

الأسئلة الشائعة

هل درس «بروتوكولات الاتصال (حافلات الرسائل)» مجاني؟

نعم — نص درس «بروتوكولات الاتصال (حافلات الرسائل)» كامل متاح مجاناً هنا على الويب. لتمرينه بشكل تفاعلي (محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7) وفتح باقي دورة AI Agents، انتقل إلى CoddyKit PRO. تتضمن دورة AI Agents 4 دروس في المجموع.

ماذا ستتعلم في «بروتوكولات الاتصال (حافلات الرسائل)»؟

افصل الوكلاء باستخدام حافلة رسائل، مثل Redis أو NATS أو Kafka، حتى يتمكنوا من التوسع والتعطل بصورة مستقلة. تتمرن على AI Agents مع أكواد عملية تشغلها مباشرة في المتصفح، ومدرس ذكاء اصطناعي متاح 24/7 يجيب على أسئلتك أثناء عملك.

هل أحتاج إلى خبرة سابقة لأبدأ AI Agents؟

لا تُشترط خبرة سابقة. AI Agents على CoddyKit منظم للمبتدئين حتى المتقدمين، لذا يمكنك البدء من هنا أو من البداية والتقدم بسرعتك الخاصة. هذا هو الدرس 4 من أصل 4.

كم من الوقت يستغرق درس «بروتوكولات الاتصال (حافلات الرسائل)»؟

معظم دروس CoddyKit تستغرق حوالي 5–10 دقائق. كل منها موجز وتفاعلي، لذا تحرز تقدماً مستمراً وتستأنف من حيث توقفت عبر الويب والتطبيق.

هل يمكنني كتابة وتشغيل أكواد في درس AI Agents هذا؟

نعم. كل درس في AI Agents يتضمن محرر أكواد مدمج، لذا تكتب وتشغل أكواداً حقيقية مباشرة في متصفحك وتحصل على تعليقات فورية من الذكاء الاصطناعي — بدون إعداد محلي.

جميع الدروس في هذه الدورة

وكلاء متعددون قائمون على المحادثة (AutoGen)
المشرفون الهرميون (المنسق والعمال)
أدوار الوكلاء وتخصصاتهم
بروتوكولات الاتصال (حافلات الرسائل)

← العودة إلى AI Agents