Airbyte: بنية تحتية مفتوحة المصدر لنقل البيانات

Airbyte هي بنية تحتية مفتوحة المصدر لنقل البيانات لبناء خطوط أنابيب استخراج وتحميل البيانات (EL). وهي مصممة لتعدد الاستخدامات، وقابلية التوسع، وسهولة الاستخدام. يأتي كتالوج موصل Airbyte "جاهزًا" مع أكثر من 350 موصلًا مدمجًا مسبقًا. يمكن استخدام هذه الموصلات لبدء نسخ البيانات من مصدر إلى وجهة في بضع دقائق فقط.

المكونات الرئيسية ل Airbyte

1. كتالوج الموصلات

• أكثر من 350 موصل مدمج مسبقاً: يأتي كتالوج موصل Airbyte "جاهزاً" مع أكثر من 350 موصل مدمج مسبقاً. يمكن استخدام هذه الموصلات لبدء نسخ البيانات من مصدر إلى وجهة في بضع دقائق فقط.
• منشئ موصل بدون كود: يمكنك بسهولة توسيع وظائف Airbyte لدعم حالات الاستخدام المخصصة الخاصة بك من خلال أدوات مثل أداة إنشاء الموصلات بدون كود.

2. المنصة

توفر منصة Airbyte جميع الخدمات الأفقية المطلوبة لتهيئة عمليات نقل البيانات وتوسيع نطاقها، وهي متاحة كخدمة مُدارة سحابياً أو مُدارة ذاتياً.

3. واجهة المستخدم

تتميز Airbyte بواجهة مستخدم، وPyAirbyte (مكتبة Python)، وواجهة برمجة التطبيقات، ومزود Terraform للتكامل مع الأدوات والنهج المفضل لديك لإدارة البنية التحتية.

مع قدرة Airbyte، يمكن للمستخدمين دمج مصادر البيانات في مجموعة Milvus للبحث عن التشابه.

قبل أن تبدأ

ستحتاج إلى

• حساب Zendesk (أو مصدر بيانات آخر تريد مزامنة البيانات منه)
• حساب Airbyte أو مثيل محلي
• مفتاح OpenAI API
• مجموعة ميلفوس العنقودية
• بايثون 3.10 مثبت محلياً

إعداد مجموعة ميلفوس العنقودية

إذا كنت قد قمت بالفعل بنشر مجموعة K8s للإنتاج، يمكنك تخطي هذه الخطوة والمتابعة مباشرةً لنشر Milvus Operator. إذا لم يكن الأمر كذلك، يمكنك اتباع الخطوات لنشر مجموعة ميلفوس العنقودية مع مشغل ميلفوس.

يتم تخزين الكيانات الفردية (في حالتنا، تذاكر الدعم ومقالات القاعدة المعرفية) في "مجموعة" - بعد إعداد مجموعتك، تحتاج إلى إنشاء مجموعة. اختر اسمًا مناسبًا واضبط البُعد على 1536 لمطابقة البُعد المتجه الذي تم إنشاؤه بواسطة خدمة تضمينات OpenAI.

بعد الإنشاء، قم بتسجيل نقطة النهاية ومعلومات المصادقة.

إعداد الاتصال في Airbyte

قاعدة بياناتنا جاهزة، دعنا ننقل بعض البيانات! للقيام بذلك، نحتاج إلى تكوين اتصال في Airbyte. إما أن تقوم بالتسجيل للحصول على حساب Airbyte السحابي على cloud.airbyte.com أو قم بتشغيل مثيل محلي كما هو موضح في الوثائق.

إعداد المصدر

بمجرد تشغيل المثيل الخاص بك، نحتاج إلى إعداد الاتصال - انقر فوق "اتصال جديد" واختر موصل "دعم Zendesk" كمصدر. بعد النقر فوق الزر "اختبار وحفظ"، ستتحقق Airbyte من إمكانية إنشاء الاتصال.

على سحابة Airbyte، يمكنك المصادقة بسهولة عن طريق النقر على زر "مصادقة". عند استخدام مثيل Airbyte محلي، اتبع الإرشادات الموضحة في صفحة التوثيق.

إعداد الوجهة

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

يقوم موصل "ميلفوس" بثلاثة أشياء:

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

دعنا نختار حجم قطعة بحجم 1000 رمز وحقول نصية من النص الأساسي والعنوان والوصف والموضوع، حيث ستكون هذه موجودة في البيانات التي سنتلقاها من Zendesk.

• التضمين - يؤدي استخدام نماذج التعلّم الآلي إلى تحويل الأجزاء النصية التي ينتجها جزء المعالجة إلى تضمينات متجهة يمكنك بعد ذلك البحث عن التشابه الدلالي. لإنشاء التضمينات، يجب عليك توفير مفتاح OpenAI API. ستقوم Airbyte بإرسال كل جزء إلى OpenAI وإضافة المتجه الناتج إلى الكيانات التي تم تحميلها في مجموعة Milvus الخاصة بك.
• الفهرسة - بمجرد الانتهاء من فهرسة القطع، يمكنك تحميلها في قاعدة البيانات. للقيام بذلك، أدخل المعلومات التي حصلت عليها عند إعداد مجموعتك ومجموعتك في مجموعة Milvus العنقودية.
  
  سيؤدي النقر فوق "اختبار وحفظ" إلى التحقق مما إذا كان كل شيء مصطفًا بشكل صحيح (بيانات اعتماد صالحة، المجموعة موجودة ولها نفس الأبعاد المتجهة مثل التضمين الذي تم تكوينه، وما إلى ذلك).

إعداد تدفق مزامنة الدفق

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

يمكنك تحديد الحقول المراد استخراجها من المصدر بالنقر على اسم الدفق. يعني وضع المزامنة "تزايدي |إلحاق + مستخرج" أن عمليات تشغيل الاتصال اللاحقة تحافظ على مزامنة Zendesk و Milvus أثناء نقل الحد الأدنى من البيانات (فقط المقالات والتذاكر التي تغيرت منذ آخر تشغيل).

بمجرد إعداد الاتصال، سيبدأ Airbyte في مزامنة البيانات. قد يستغرق الأمر بضع دقائق لتظهر في مجموعة ميلفوس الخاصة بك.

إذا قمت بتحديد تكرار النسخ المتماثل، فسيتم تشغيل Airbyte بانتظام للحفاظ على تحديث مجموعة Milvus الخاصة بك بالتغييرات التي تطرأ على مقالات Zendesk والمشكلات التي تم إنشاؤها حديثًا.

التحقق من التدفق

يمكنك التحقق في واجهة مستخدم مجموعة Milvus من واجهة مستخدم Milvus من كيفية تنظيم البيانات في المجموعة من خلال الانتقال إلى الملعب وتنفيذ استعلام "استعلام البيانات" مع تعيين عامل تصفية إلى "_ab_stream == \"تذاكر\"".

كما ترى في طريقة عرض النتيجة، يتم تخزين كل سجل قادم من Zendesk ككيانات منفصلة في Milvus مع جميع البيانات الوصفية المحددة. يظهر الجزء النصي الذي يستند إليه التضمين كخاصية "نص" - هذا هو النص الذي تم تضمينه باستخدام OpenAI وسيكون ما سنبحث عنه.

إنشاء تطبيق Streamlit للاستعلام عن المجموعة

بياناتنا جاهزة - نحتاج الآن إلى بناء التطبيق لاستخدامها. في هذه الحالة، سيكون التطبيق عبارة عن نموذج دعم بسيط للمستخدمين لإرسال حالات الدعم. عندما يضغط المستخدم على إرسال سنقوم بأمرين:

• البحث عن التذاكر المماثلة التي أرسلها مستخدمون من نفس المؤسسة
• البحث عن المقالات القائمة على المعرفة التي قد تكون ذات صلة بالمستخدم

في كلتا الحالتين، سنستفيد من البحث الدلالي باستخدام تضمينات OpenAI. للقيام بذلك، يتم أيضًا تضمين وصف المشكلة التي أدخلها المستخدم واستخدامه لاسترداد الكيانات المماثلة من مجموعة Milvus. إذا كانت هناك نتائج ذات صلة، يتم عرضها أسفل النموذج.

إعداد بيئة واجهة المستخدم

ستحتاج إلى تثبيت بايثون محلي لأننا سنستخدم Streamlit لتنفيذ التطبيق.

أولًا، قم بتثبيت Streamlit ومكتبة عميل Milvus ومكتبة عميل OpenAI محليًا:

pip install streamlit pymilvus openai

لتقديم نموذج دعم أساسي، قم بإنشاء ملف بايثون basic_support_form.py:

import streamlit as st

with st.form("my_form"):
    st.write("Submit a support case")
    text_val = st.text_area("Describe your problem")

    submitted = st.form_submit_button("Submit")
    if submitted:
        # TODO check for related support cases and articles
        st.write("Submitted!")

لتشغيل تطبيقك، استخدم Streamlit run:

streamlit run basic_support_form.py

سيؤدي ذلك إلى تصيير نموذج أساسي:

يمكن أيضًا العثور على الكود الخاص بهذا المثال على GitHub.

إعداد خدمة استعلام الواجهة الخلفية

بعد ذلك، دعنا نتحقق من التذاكر المفتوحة الحالية التي قد تكون ذات صلة. للقيام بذلك، قمنا بتضمين النص الذي أدخله المستخدم باستخدام OpenAI، ثم أجرينا بحثًا عن التشابه في مجموعتنا، مع تصفية التذاكر التي لا تزال مفتوحة. إذا كانت هناك تذكرة ذات مسافة منخفضة جدًا بين التذكرة المقدمة والتذكرة الحالية، فأعلم المستخدم ولا ترسل:

import streamlit as st
import os
import pymilvus
import openai


with st.form("my_form"):
    st.write("Submit a support case")
    text_val = st.text_area("Describe your problem?")

    submitted = st.form_submit_button("Submit")
    if submitted:
        import os
        import pymilvus
        import openai

        org_id = 360033549136 # TODO Load from customer login data

        pymilvus.connections.connect(uri=os.environ["MILVUS_URL"], token=os.environ["MILVUS_TOKEN"])
        collection = pymilvus.Collection("zendesk")

        embedding = openai.Embedding.create(input=text_val, model="text-embedding-ada-002")['data'][0]['embedding']

        results = collection.search(data=[embedding], anns_field="vector", param={}, limit=2, output_fields=["_id", "subject", "description"], expr=f'status == "new" and organization_id == {org_id}')

        st.write(results[0])
        if len(results[0]) > 0 and results[0].distances[0] < 0.35:
            matching_ticket = results[0][0].entity
            st.write(f"This case seems very similar to {matching_ticket.get('subject')} (id #{matching_ticket.get('_id')}). Make sure it has not been submitted before")
        else:
            st.write("Submitted!")
            

تحدث عدة أشياء هنا

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

لتشغيل التطبيق الجديد، تحتاج إلى تعيين متغيرات البيئة لـ OpenAI و Milvus أولاً:

export MILVUS_TOKEN=...
export MILVUS_URL=https://...
export OPENAI_API_KEY=sk-...

streamlit run app.py

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

يمكن أيضًا العثور على الكود الخاص بهذا المثال على GitHub.

إظهار المزيد من المعلومات ذات الصلة

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

في الختام، دعنا نعرض معلومات مفيدة بعد إرسال التذكرة لإعطاء المستخدم أكبر قدر ممكن من المعلومات ذات الصلة مقدمًا.

للقيام بذلك، سنقوم بإجراء بحثٍ ثانٍ بعد إرسال التذكرة لجلب مقالات قاعدة المعرفة الأكثر مطابقة:

   ......
   
        else:
            # TODO Actually send out the ticket
            st.write("Submitted!")
            article_results = collection.search(data=[embedding], anns_field="vector", param={}, limit=5, output_fields=["title", "html_url"], expr=f'_ab_stream == "articles"')
            st.write(article_results[0])
            if len(article_results[0]) > 0:
                st.write("We also found some articles that might help you:")
                for hit in article_results[0]:
                    if hit.distance < 0.362:
                        st.write(f"* [{hit.entity.get('title')}]({hit.entity.get('html_url')})")

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

يمكن أيضًا العثور على الكود الخاص بهذا المثال على Github.

الخلاصة

على الرغم من أن واجهة المستخدم المعروضة هنا ليست نموذج دعم فعلي، بل هي مثال لتوضيح حالة الاستخدام، إلا أن الجمع بين Airbyte و Milvus قوي للغاية - فهو يجعل من السهل تحميل النص من مجموعة متنوعة من المصادر (من قواعد البيانات مثل Postgres عبر واجهات برمجة التطبيقات مثل Zendesk أو GitHub إلى مصادر مخصصة تمامًا تم إنشاؤها باستخدام مجموعة أدوات تطوير البرمجيات SDK الخاصة بـ Airbyte أو أداة إنشاء الموصلات المرئية) وفهرستها في شكل مضمن في Milvus، وهو محرك بحث متجه قوي قادر على التوسع إلى كميات هائلة من البيانات.

Airbyte و Milvus مفتوحا المصدر ومجانيان تمامًا للاستخدام على بنيتك التحتية، مع عروض سحابية لإلغاء تحميل العمليات إذا رغبت في ذلك.

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