البحث باستخدام EmbeddingLists: ColBERT و ColPali

يوضح هذا البرنامج التعليمي كيفية إنشاء أنظمة استرجاع على غرار ColBERT وColPali باستخدام البحث عن قوائم التضمين (EmbeddingList) في الحقول الفرعية للمتجهات StructArray في Milvus. استخدمه عندما يتم تمثيل كل من استعلامك والبيانات المخزنة على شكل قوائم من المتجهات، وترغب في إجراء استرجاع متأخر على مستوى الكيان باستخدام مقاييس التفاعل المتأخر ( MAX_SIM* ).

للاطلاع على أساسيات StructArray التي يستند إليها هذا البرنامج التعليمي، راجع «إنشاء حقل StructArray» و«فهرسة حقول StructArray» و«البحث المتجهي الأساسي باستخدام StructArray». يركز هذا البرنامج التعليمي على سير عمل ColBERT وColPali بدلاً من بناء جملة StructArray العامة.

نظرة عامة

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

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

ColBERT (arXiv: 2004.12832) هو نظام استرجاع نص-نص يوفر بحثًا فعالاً وناجعًا عن المقاطع من خلال التفاعلات المتأخرة في السياق عبر BERT. وهو يتيح الترميز المستقل للاستعلامات والمستندات على مستوى الرموز ويحسب مدى تشابهها.

الترميز على مستوى الرموز

أثناء استيعاب البيانات في ColBERT، يتم تقسيم كل وثيقة إلى رموز، ثم يتم تحويلها إلى متجهات وتخزينها كقائمة تضمين، كما في $∈Rn×dd \rightarrow E_d = [e_{d1}, e_{d2}, \dots, e_{dn}] ∈ \R^{n×d}$ E d = [ $e$ d1 $,{}^{}e$ d2 $,\dots ,e$ dn $]$ $\in$ R ${}^{n\times d}$. عند وصول استعلام، يتم أيضًا تجزئته وتحويله إلى متجهات وتخزينه كقائمة تضمين، كما في $∈Rm×dq \rightarrow E_q = [e_{q1}, e_{q2}, \dots, e_{qm}] ∈ \R^{m×d}$ $\to$ E q $=$ [e q1 $,e$ q2 $,\dots ,e$ qm $]\in$ R ${}^{m\times d}$.

في الصيغ أعلاه،

• $dd$: وثيقة

• $qq$: الاستعلام

• $EdE_d$: قائمة التضمين التي تمثل الوثيقة.

• $EqE_q$: قائمة التضمين التي تمثل الاستعلام.

• $[e_{d1}, e_{d2}, \dots, e_{dn}] ∈ \R^{n×d}$$,e_{}$$,\dots$ ،${و}_{}$$]\in$ R${}^{n\times d}$: عدد التضمينات المتجهة في قائمة التضمينات التي تمثل المستند يقع ضمن نطاق $\R^{n×d}$.

• $[e_{q1}, e_{q2}, \dots, e_{qm}] ∈ \R^{m×d}$q1,eq2,…,eqm$]\in$ R${}^{m\times d}$: عدد التضمينات المتجهة في قائمة التضمينات التي تمثل الاستعلام يقع ضمن نطاق $\R^{m×d}$.

التفاعل المتأخر

بمجرد اكتمال التحويل إلى متجهات، تتم مقارنة قائمة تضمينات الاستعلام بقائمة تضمينات كل وثيقة، رمزًا برمز، لتحديد درجة التشابه النهائية.

كما هو موضح في الرسم البياني أعلاه، يحتوي الاستعلام على رمزين، وهما machine و learning ، بينما يحتوي المستند الموجود في النافذة على أربعة رموز: neural و network و python و tutorial. بمجرد تحويل هذه الرموز إلى متجهات، تتم مقارنة تضمينات المتجهات لكل رمز من رموز الاستعلام بتلك الموجودة في المستند للحصول على قائمة بدرجات التشابه. ثم يتم جمع أعلى الدرجات من كل قائمة درجات للحصول على الدرجة النهائية. تُعرف عملية تحديد الدرجة النهائية للوثيقة باسم «التشابه الأقصى» (MAX_SIM). للحصول على تفاصيل حول التشابه الأقصى، راجع «التشابه الأقصى».

امتداد ColPali

استنادًا إلى ColBERT، يقترح ColPali (arXiv: 2407.01449) نهجًا مبتكرًا لاسترجاع المستندات الغنية بصريًّا والذي يستفيد من نماذج الرؤية واللغة (VLMs). أثناء استيعاب البيانات، يتم تحويل كل صفحة من المستند إلى صورة عالية الدقة، ثم تقسيمها إلى أجزاء، بدلاً من تحويلها إلى رموز. على سبيل المثال، يمكن أن تنتج صورة صفحة مستند بحجم 448 × 448 بكسل 1,024 جزءًا، يبلغ حجم كل منها 14 × 14 بكسل.

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

يُسمى نموذج اللغة المرئي (VLM) المستخدم في ColPali بـ PaliGemma (arXiv: 2407.07726)، والذي يتألف من مُشفّر صور (SigLIP-400M)، ونموذج لغوي مخصص للفك فقط (Gemma2-2B)، وطبقة خطية تعمل على إسقاط ناتج مشفر الصور في الفضاء المتجهي لنموذج اللغة، كما هو موضح في الرسم البياني أعلاه.

أثناء استيعاب البيانات، تُقسَّم صفحة المستند، المُمثَّلة كصورة خام، إلى بقع بصرية متعددة، يتم تضمين كل منها لتوليد قائمة من التضمينات المتجهة. ثم يتم إسقاطها في الفضاء المتجهي لنموذج اللغة للحصول على قائمة التضمينات النهائية، كما في $∈Rn×dd \rightarrow E_d = [e_{d1}, e_{d2}, \dots, e_{dn}] ∈ \R^{n×d}$ E d = [ $e$ d1 $,e$ d2 $,\dots ,e$ dn $]$ $\in$ R ${}^{n\times d}$. عند وصول استعلام، يتم تجزئته إلى رموز، ويتم تضمين كل رمز لتوليد قائمة من التضمينات المتجهة، كما في $∈Rm×dq \rightarrow E_q = [e_{q1}, e_{q2}, \dots, e_{qm}] ∈ \R^{m×d}$ $\to$ E q $=$ [e q1 $,e$ q2 $,\dots ,e$ qm $]\in$ R ${}^{m\times d}$. ثم تم تطبيق MAX_SIM لمقارنة قائمتي التضمين والحصول على النتيجة النهائية بين الاستعلام وصفحة الوثيقة.

نظام استرجاع النصوص ColBERT

في هذا القسم، سنقوم بإعداد نظام استرجاع النصوص ColBERT باستخدام StructArray. قبل ذلك، قم بإعداد مثيل Milvus v2.6.x، واحصل على رمز وصول Cohere.

الخطوة 1: تثبيت التبعيات

قم بتشغيل الأمر التالي لتثبيت التبعيات.

pip install --upgrade huggingface-hub transformers datasets pymilvus cohere

الخطوة 2: تحميل مجموعة بيانات Cohere

في هذا المثال، سنستخدم مجموعة بيانات ويكيبيديا من Cohere ونسترد أول 10,000 سجل. يمكنك العثور على معلومات حول مجموعة البيانات هذه في هذه الصفحة.

from datasets import load_dataset

lang = "simple"
docs = load_dataset(
    "Cohere/wikipedia-2023-11-embed-multilingual-v3",
    lang,
    split="train[:10000]"
)

سيؤدي تشغيل البرامج النصية المذكورة أعلاه إلى تنزيل مجموعة البيانات إذا لم تكن متوفرة محليًّا. كل سجل في مجموعة البيانات يمثل فقرة من إحدى صفحات ويكيبيديا. يوضح الجدول التالي بنية مجموعة البيانات هذه.

اسم العمود	الوصف
_id	معرف السجل
url	عنوان URL للسجل الحالي.
title	عنوان المستند المصدر.
text	فقرة من المستند المصدر.
emb	عمليات التضمين للنص المأخوذ من المستند المصدر.

الخطوة 3: تجميع الفقرات حسب العنوان

للبحث عن المستندات بدلاً من الفقرات، يجب تجميع الفقرات حسب العنوان.

df = docs.to_pandas()
groups = df.groupby('title')

data = []

for title, group in groups:
  data.append({
      "title": title,
      "paragraphs": [{
          "text": row['text'],
          'emb': row['emb']
      } for _, row in group.iterrows()]
  })

في هذا الكود، نقوم بتخزين الفقرات المجمعة كوثائق وإدراجها في قائمة data. تحتوي كل وثيقة على مفتاح paragraphs ، وهو عبارة عن قائمة بالفقرات؛ ويحتوي كل كائن فقرة على مفتاحي text و emb.

الخطوة 4: إنشاء مجموعة لمجموعة بيانات Cohere

بمجرد أن تصبح البيانات جاهزة، سنقوم بإنشاء مجموعة. في المجموعة، يُعد « paragraphs » حقل StructArray. للحصول على شرح عام لمخططات StructArray، راجع «إنشاء حقل StructArray».

from pymilvus import MilvusClient, DataType

client = MilvusClient(
    uri="http://localhost:19530",
    token="root:Milvus"
)

# Create collection schema
schema = client.create_schema()

schema.add_field('id', DataType.INT64, is_primary=True, auto_id=True)
schema.add_field('title', DataType.VARCHAR, max_length=512)

# Create struct schema
struct_schema = client.create_struct_field_schema()
struct_schema.add_field('text', DataType.VARCHAR, max_length=65535)
struct_schema.add_field('emb', DataType.FLOAT_VECTOR, dim=512)

schema.add_field('paragraphs', DataType.ARRAY,
                 element_type=DataType.STRUCT,
                 struct_schema=struct_schema, max_capacity=200)

# Create index parameters
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="paragraphs[emb]",
    index_type="AUTOINDEX",
    metric_type="MAX_SIM_COSINE"
)

# Create a collection
client.create_collection(
    collection_name='wiki_documents',
    schema=schema,
    index_params=index_params
)

الخطوة 5: إدراج مجموعة بيانات Cohere في المجموعة

يمكننا الآن إدراج البيانات المعدة في المجموعة التي أنشأناها أعلاه.

client.insert(
    collection_name='wiki_documents',
    data=data
)

الخطوة 6: البحث داخل مجموعة بيانات Cohere

وفقًا لتصميم ColBERT، يجب تقطيع نص الاستعلام إلى رموز (tokenized) ثم تضمينه في قائمة EmbeddingList. في هذه الخطوة، سنستخدم نفس النموذج الذي استخدمه Cohere لتوليد التضمينات (embeddings) للفقرات في مجموعة بيانات ويكيبيديا.

import cohere

co = cohere.ClientV2("COHERE_API_KEY")

query_inputs = [
    {
        'content': [
            {'type': 'text', 'text': 'Adobe'},
        ]
    },
    {
        'content': [
            {'type': 'text', 'text': 'software'}
        ]
    }
]

embeddings = co.embed(
    inputs=query_inputs,
    model='embed-multilingual-v3.0',
    input_type="classification",
    embedding_types=["float"],
)

في الكود، يتم تنظيم نصوص الاستعلام إلى رموز في ` query_inputs ` وتضمينها في قائمة من المتجهات العائمة. بعد ذلك، يمكنك استخدام `EmbeddingList` من Milvus لإجراء بحث عن التشابه على النحو التالي.

from pymilvus.client.embedding_list import EmbeddingList

query_emb_list = EmbeddingList()

if (embeddings.embeddings.float):
  query_emb_list.add_batch(embeddings.embeddings.float)

results = client.search(
    collection_name="wiki_documents",
    data=[query_emb_list],
    anns_field="paragraphs[emb]",
    limit=10,
    output_fields=["title"]
)

for hit in results[0]:
  print(f"Document {hit['entity']['title']}: {hit['distance']:.4f}")

تكون ناتج الكود أعلاه مشابهًا لما يلي:

# Document Software: 2.3035
# Document Application: 2.1875
# Document Adobe Illustrator: 2.1167
# Document Open source: 2.0542
# Document Computer: 1.9811
# Document Microsoft: 1.9784
# Document Web browser: 1.9655
# Document Program: 1.9627
# Document Website: 1.9594
# Document Computer science: 1.9460

تتراوح كل درجة تشابه جيب التمام بين -1 و 1. ويمكن أن تكون درجة MAX_SIM_COSINE النهائية أكبر من 1 لأنها تجمع بين عدة درجات تشابه قصوى على مستوى الرموز.

نظام استرجاع المستندات ColPali

في هذا القسم، سنقوم بإعداد نظام استرجاع المستندات القائم على ColPali باستخدام StructArray. قبل ذلك، قم بإعداد مثيل Milvus v2.6.x.

الخطوة 1: تثبيت التبعيات

pip install --upgrade huggingface-hub transformers datasets pymilvus 'colpali-engine>=0.3.0,<0.4.0'

الخطوة 2: تحميل مجموعة بيانات Vidore

في هذا القسم، سنستخدم مجموعة بيانات Vidore المسماة vidore_v2_finance_en. هذه المجموعة عبارة عن مجموعة من التقارير السنوية من القطاع المصرفي، مخصصة لمهام فهم المستندات الطويلة. وهي واحدة من 10 مجموعات تشكل معيار ViDoRe v3. يمكنك العثور على تفاصيل حول مجموعة البيانات هذه على هذه الصفحة.

from datasets import load_dataset

ds = load_dataset("vidore/vidore_v3_finance_en", "corpus")
df = ds['test'].to_pandas()

سيؤدي تشغيل البرامج النصية المذكورة أعلاه إلى تنزيل مجموعة البيانات إذا لم تكن متوفرة محليًّا. كل سجل في مجموعة البيانات يمثل صفحة من تقرير مالي. يوضح الجدول التالي بنية مجموعة البيانات هذه.

اسم العمود	الوصف
corpus_id	سجل في المجموعة
image	صورة الصفحة بالبايت.
doc_id	معرف المستند الوصفي.
page_number_in_doc	رقم الصفحة الحالية في المستند.

الخطوة 3: إنشاء تضمينات لصور الصفحات

كما هو موضح في قسم "نظرة عامة" ، فإن نموذج ColPali هو نموذج VLM الذي يعرض الصور في الفضاء المتجهي لنموذج نصي. في هذه الخطوة، سنستخدم أحدث إصدار من نموذج ColPali وهو vidore/colpali-v1.3. يمكنك العثور على تفاصيل حول هذا النموذج في هذه الصفحة.

import torch
from typing import cast
from colpali_engine.models import ColPali, ColPaliProcessor

model_name = "vidore/colpali-v1.3"

model = ColPali.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="cuda:0",  # or "mps" if on Apple Silicon
).eval()

processor = ColPaliProcessor.from_pretrained(model_name)

بمجرد أن يصبح النموذج جاهزًا، يمكنك محاولة إنشاء رقع لصورة معينة على النحو التالي.

from PIL import Image
from io import BytesIO

# Use the iterrows() generator to get the first row.
row = next(df.iterrows())[1]

# Decode the image bytes and generate patch embeddings.
images = [Image.open(BytesIO(row["image"]["bytes"]))]
batch_images = processor.process_images(images).to(model.device)

with torch.no_grad():
    patches_embeddings = model(**batch_images)[0]

# Check the shape of the embeddings generated for the patches.
print(patches_embeddings.shape)

# [1031, 128]

في الكود أعلاه، يقوم نموذج ColPali بتغيير حجم الصورة إلى 448 × 448 بكسل، ثم يقسمها إلى أجزاء، يبلغ حجم كل منها 14 × 14 بكسل. وأخيرًا، يتم تضمين هذه الأجزاء في 1,031 تمثيلًا، لكل منها 128 بُعدًا.

يمكنك إنشاء التضمينات لجميع الصور باستخدام حلقة كما يلي:

data = []

for _, row in df.iterrows():
    corpus_id = row["corpus_id"]
    images = [Image.open(BytesIO(row["image"]["bytes"]))]
    batch_images = processor.process_images(images).to(model.device)

    with torch.no_grad():
        patches = model(**batch_images)[0]

    doc_id = row["doc_id"]
    page_number_in_doc = row["page_number_in_doc"]

    data.append({
        "corpus_id": corpus_id,
        "patches": [
            {"emb": emb.float().cpu().tolist()}
            for emb in patches
        ],
        "doc_id": doc_id,
        "page_number_in_doc": page_number_in_doc,
    })

الخطوة 4: إنشاء مجموعة لمجموعة بيانات التقارير المالية

بمجرد أن تصبح البيانات جاهزة، سنقوم بإنشاء مجموعة. في المجموعة، يُعد « patches » حقلًا من نوع StructArray. يخزن كل عنصر من عناصر Struct تضمينًا واحدًا للرقعة. للاطلاع على متطلبات الفهرسة الخاصة بالحقول الفرعية للمتجه StructArray، راجع فهرسة حقول StructArray.

from pymilvus import MilvusClient, DataType

client = MilvusClient(
    uri=YOUR_CLUSTER_ENDPOINT,
    token=YOUR_API_KEY
)

schema = client.create_schema()

schema.add_field(
    field_name="corpus_id",
    datatype=DataType.INT64,
    is_primary=True
)

patch_schema = client.create_struct_field_schema()

patch_schema.add_field(
    field_name="emb",
    datatype=DataType.FLOAT_VECTOR,
    dim=128
)

schema.add_field(
    field_name="patches",
    datatype=DataType.ARRAY,
    element_type=DataType.STRUCT,
    struct_schema=patch_schema,
    max_capacity=1031
)

schema.add_field(
    field_name="doc_id",
    datatype=DataType.VARCHAR,
    max_length=512
)

schema.add_field(
    field_name="page_number_in_doc",
    datatype=DataType.INT64
)

index_params = client.prepare_index_params()

index_params.add_index(
    field_name="patches[emb]",
    index_type="AUTOINDEX",
    metric_type="MAX_SIM_COSINE"
)

client.create_collection(
    collection_name="financial_reports",
    schema=schema,
    index_params=index_params
)

الخطوة 5: إدراج التقارير المالية في المجموعة

يمكننا الآن إدراج التقارير المالية المعدة في المجموعة.

client.insert(
    collection_name="financial_reports",
    data=data
)

من الناتج، يمكنك ملاحظة أن جميع الصفحات من مجموعة بيانات Vidore قد تم إدراجها.

الخطوة 6: البحث داخل التقارير المالية

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

from pymilvus.client.embedding_list import EmbeddingList

queries = [
    "quarterly revenue growth chart"
]

batch_queries = processor.process_queries(queries).to(model.device)

with torch.no_grad():
    query_embeddings = model(**batch_queries)

query_emb_list = EmbeddingList()
query_emb_list.add_batch(query_embeddings[0].float().cpu().tolist())

results = client.search(
    collection_name="financial_reports",
    data=[query_emb_list],
    anns_field="patches[emb]",
    limit=10,
    output_fields=["doc_id", "page_number_in_doc"]
)