Technical Tutorials

مع تطور تطبيقات نماذج اللغات الكبيرة (LLM) من روبوتات الإجابة البسيطة إلى أنظمة معقدة لتوليد المعلومات المعززة بالاسترجاع متعدد الوكلاء (RAG)، أصبحت إمكانية المراقبة عن بُعد عقبة حرجة. في إعداد RAG ذو الوكيل الواحد، تتدفق الطلبات بشكل خطي: استعلام، استرجاع، توليد. ومع ذلك، في البنى متعددة الوكلاء، تتفرع الطلبات إلى وكلاء فرعيين يعملون بالتوازي، حيث يؤدي كل منها مهامًا مميزة مثل تحليل المستندات، أو البحث الدلالي، أو التحقق من الحقائق. يؤدي هذا التعقيد إلى تجزئة التتبع، حيث يتشتت طلب المستخدم الواحد عبر عشرات الحلقات غير ذات الصلة، مما يجعل تصحيح الأخطاء شبه مستحيل.

تثبت سجلات البيانات التقليدية (Logging) عدم كفايتها لأنها تفتقر إلى السياق. هنا يبرز دور OpenTelemetry (OTel). من خلال تنفيذ أنماط تتبع مخصصة، يمكنك ربط هذه التتبعات المجزأة معًا، مما يوفر رؤية موحدة للبنية التحتية للذكاء الاصطناعي الخاص بك. في هذا المنشور، نستكشف كيفية هيكلة هذه الأنماط باستخدام لغة Python.

تحديات سير العمل الموزع للذكاء الاصطناعي

تخيل نظام دعم العملاء الذي يتكون من ثلاثة وكلاء: وكيل الفرز (Triage Agent)، ووكيل استرجاع المعرفة، ومولد الاستجابة. عندما يقدم المستخدم استعلامًا، يحدد وكيل الفرز النية ويوكل المهام. إذا كانت النية تقنية، يبحث وكيل المعرفة في قاعدة بيانات متجهة (vector database)؛ وإذا كانت متعلقة بالفواتير، فإنه يستعلم من قاعدة بيانات SQL.

بدون ارتباط صحيح، تظهر الحلقات (Spans) التي ينشئها هؤلاء الوكلاء كجزر معزولة في لوحة مراقبة أداء التطبيق (APM). لا يمكنك معرفة أن زمن الاستجابة الطويل في مولد الاستجابة كان ناتجًا عن انقطاع في الاتصال (timeout) في البحث المتجه لوكيل المعرفة. لحل هذه المشكلة، نحتاج إلى نشر السياق بشكل صريح عبر حدود الوكلاء.

تنفيذ التتبع المخصص باستخدام OpenTelemetry

تتضمن الاستراتيجية الأساسية إنشاء غلاف مخصص يلتقط سياق التنفيذ لكل وكيل. نستخدم مكتبة opentelemetry-api لإدارة الحلقات يدويًا، مما يضمن تجميع المكالمات المتداخلة داخل منطق الوكيل الداخلي تحت حلقة أبوية واحدة. هذه الطريقة مفيدة بشكل خاص عندما تستخدم وكلاؤك مكتبات متعددة (مثل LangChain، أو LlamaIndex، أو عملاء HTTP مخصصين) قد لا تعمل معها أدوات التتبع التلقائي بشكل جيد.

يُعد المثال العملي التالي غلافًا مخصصًا (decorator) يلتف حول طريقة تنفيذ الوكيل، مما يضمن بنية تتبع متسقة:

import functools
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer(__name__)

def agent_span(agent_name: str):
    """
    غلاف يلتف حول طريقة الوكيل في حلقة OpenTelemetry مخصصة.
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with tracer.start_as_current_span(f"agent.{agent_name}") as span:
                try:
                    # تعيين سمات لتحسين التصفية في أدوات مراقبة الأداء
                    span.set_attribute("agent.name", agent_name)
                    span.set_attribute("args", str(args)[:100]) # تطهير المدخلات
                    
                    # تنفيذ منطق الوكيل
                    result = func(*args, **kwargs)
                    
                    # تحديد الحالة كنجحة
                    span.set_status(Status(StatusCode.OK))
                    return result
                except Exception as e:
                    # تحديد الحالة كفشل وتسجيل الخطأ
                    span.set_status(Status(StatusCode.ERROR, str(e)))
                    span.record_exception(e)
                    raise
        return wrapper
    return decorator

# مثال على الاستخدام
class KnowledgeRetrievalAgent:
    @agent_span("knowledge_retrieval")
    def search_docs(self, query: str):
        # محاكاة بحث مكلف في قاعدة البيانات المتجهة
        return {"doc_id": "123", "content": "الإجابة هي 42."}

في هذا النمط، يعمل الغلاف @agent_span كحاوية لجميع العمليات الداخلية التي يؤديها ذلك الوكيل المحدد. عند تسلسل عدة وكلاء معًا، يتم تشعب الحلقات الفرعية (من المنطق الداخلي) تلقائيًا داخل حلقة الوكيل، مما يخلق عرضًا هرميًا بدلاً من قائمة مسطحة من الأحداث.

ربط الحلقات عبر حدود الخدمات

بالنسبة لأنظمة الوكلاء المتعددة التي يتم نشرها كخدمات مصغرة (microservices)، يعد نشر السياق أمرًا حيويًا. يجب أن تضمن مرور TraceId و SpanId عبر الرؤوس (headers) (مثل رؤوس HTTP في واجهات برمجة التطبيقات REST أو رؤوس الرسائل في Kafka). تتعامل أدوات نشر السياق في OpenTelemetry مع هذا الأمر تلقائيًا إذا كانت أدوات تتبع عميل أو خادم HTTP الخاصة بك مفعّلة. ومع ذلك، بالنسبة لمقاييس الرسائل المخصصة أو حافلات الأحداث الداخلية، قد تحتاج إلى حقن حامل السياق يدويًا:

from opentelemetry.propagate import inject

def send_message_to_agent(queue, message, agent_name):
    # حقن سياق التتبع في بيانات الرسالة الوصفية
    headers = {}
    inject(headers)
    
    queue.publish({
        "agent": agent_name,
        "payload": message,
        "trace_headers": headers
    })

الخاتمة

حل مشكلة تجزئة التتبع في أنظمة RAG متعددة الوكلاء لا يتعلق فقط بجمع المزيد من البيانات؛ بل يتعلق بهيكلة تلك البيانات بذكاء. من خلال اعتماد أنماط تتبع مخصصة باستخدام OpenTelemetry، يمكن للمطورين اكتساب رؤية دقيقة لتفاعلات الوكلاء، وتحديد الاختناقات عبر سير العمل المتوازي، وبناء تطبيقات ذكاء اصطناعي أكثر موثوقية في النهاية. مع نمو تعقيد هذه الأنظمة، ستنتقل إمكانية المراقبة عن بُعد من ميزة مرغوبة إلى العمود الفقري للتميز التشغيلي.