RAG باستخدام Milvus وواجهة برمجة التطبيقات غير المتزامنة لـ LlamaIndex

يوضح هذا البرنامج التعليمي كيفية استخدام LlamaIndex مع Milvus لإنشاء مسار معالجة مستندات غير متزامن لـ RAG. يوفر LlamaIndex طريقة لمعالجة المستندات وتخزينها في قاعدة بيانات متجهة مثل Milvus. من خلال الاستفادة من واجهة برمجة التطبيقات غير المتزامنة لـ LlamaIndex ومكتبة عميل Milvus لـ Python، يمكننا زيادة إنتاجية خط الأنابيب لمعالجة وفهرسة كميات كبيرة من البيانات بكفاءة.

في هذا البرنامج التعليمي، سنقدم أولاً استخدام الأساليب غير المتزامنة لإنشاء RAG باستخدام LlamaIndex وMilvus من منظور عام، ثم سنقدم استخدام الأساليب التفصيلية ومقارنة الأداء بين الأساليب المتزامنة وغير المتزامنة.

قبل البدء

تتطلب مقتطفات الكود في هذه الصفحة تبعيات pymilvus و llamaindex. يمكنك تثبيتها باستخدام الأوامر التالية:

$ pip install -U pymilvus llama-index-vector-stores-milvus llama-index nest-asyncio

سنستخدم النماذج من OpenAI. يجب عليك إعداد مفتاح واجهة برمجة التطبيقات (API key ) OPENAI_API_KEY كمتغير بيئة.

import os

os.environ["OPENAI_API_KEY"] = "sk-***********"

إذا كنت تستخدم Jupyter Notebook، فستحتاج إلى تشغيل سطر التعليمات البرمجية هذا قبل تشغيل التعليمات البرمجية غير المتزامنة.

import nest_asyncio

nest_asyncio.apply()

تحضير البيانات

يمكنك تنزيل بيانات نموذجية باستخدام الأوامر التالية:

$ mkdir -p 'data/'
$ wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt' -O 'data/paul_graham_essay.txt'
$ wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/10k/uber_2021.pdf' -O 'data/uber_2021.pdf'

بناء نظام RAG باستخدام المعالجة غير المتزامنة

يوضح هذا القسم كيفية إنشاء نظام RAG قادر على معالجة المستندات بطريقة غير متزامنة.

قم باستيراد المكتبات الضرورية وحدد عنوان URI الخاص بـ Milvus وأبعاد التضمين.

import asyncio
import random
import time

from llama_index.core.schema import TextNode, NodeRelationship, RelatedNodeInfo
from llama_index.core.vector_stores import VectorStoreQuery
from llama_index.vector_stores.milvus import MilvusVectorStore

URI = "http://localhost:19530"
DIM = 768

قم بتعريف دالة تهيئة يمكننا استخدامها مرة أخرى لإعادة بناء مجموعة Milvus.

def init_vector_store():
    return MilvusVectorStore(
        uri=URI,
        # token=TOKEN,
        dim=DIM,
        collection_name="test_collection",
        embedding_field="embedding",
        id_field="id",
        similarity_metric="COSINE",
        consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
        overwrite=True,  # To overwrite the collection if it already exists
    )


vector_store = init_vector_store()

2025-01-24 20:04:39,414 [DEBUG][_create_connection]: Created new connection using: faa8be8753f74288bffc7e6d38942f8a (async_milvus_client.py:600)

استخدم SimpleDirectoryReader لتغليف كائن مستند LlamaIndex من الملف paul_graham_essay.txt.

from llama_index.core import SimpleDirectoryReader

# load documents
documents = SimpleDirectoryReader(
    input_files=["./data/paul_graham_essay.txt"]
).load_data()

print("Document ID:", documents[0].doc_id)

Document ID: 41a6f99c-489f-49ff-9821-14e2561140eb

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

from llama_index.embeddings.huggingface import HuggingFaceEmbedding


embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-en-v1.5")

قم بإنشاء فهرس وإدراج المستند.

نقوم بتعيين use_async إلى True لتمكين وضع الإدراج غير المتزامن.

# Create an index over the documents
from llama_index.core import VectorStoreIndex, StorageContext

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
    embed_model=embed_model,
    use_async=True,
)

تهيئة نموذج اللغة الكبير (LLM).

from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo")

عند إنشاء محرك الاستعلام، يمكنك أيضًا تعيين المعلمة use_async إلى True لتمكين البحث غير المتزامن.

query_engine = index.as_query_engine(use_async=True, llm=llm)
response = await query_engine.aquery("What did the author learn?")

print(response)

The author learned that the field of artificial intelligence, as practiced at the time, was not as promising as initially believed. The approach of using explicit data structures to represent concepts in AI was not effective in achieving true understanding of natural language. This realization led the author to shift his focus towards Lisp and eventually towards exploring the field of art.

استكشاف واجهة برمجة التطبيقات غير المتزامنة

في هذا القسم، سنقدم استخدام واجهة برمجة التطبيقات (API) ذات المستوى الأدنى ونقارن أداء عمليات التشغيل المتزامنة وغير المتزامنة.

الإضافة غير المتزامنة

إعادة تهيئة مخزن المتجهات.

vector_store = init_vector_store()

2025-01-24 20:07:38,727 [DEBUG][_create_connection]: Created new connection using: 5e0d130f3b644555ad7ea6b8df5f1fc2 (async_milvus_client.py:600)

دعونا نحدد دالة إنتاج العقد، والتي ستُستخدم لتوليد عدد كبير من العقد الاختبارية للفهرس.

def random_id():
    random_num_str = ""
    for _ in range(16):
        random_digit = str(random.randint(0, 9))
        random_num_str += random_digit
    return random_num_str


def produce_nodes(num_adding):
    node_list = []
    for i in range(num_adding):
        node = TextNode(
            id_=random_id(),
            text=f"n{i}_text",
            embedding=[0.5] * (DIM - 1) + [random.random()],
            relationships={NodeRelationship.SOURCE: RelatedNodeInfo(node_id=f"n{i+1}")},
        )
        node_list.append(node)
    return node_list

حدد دالة غير متزامنة لإضافة المستندات إلى مخزن المتجهات. نستخدم الدالة async_add() في مثيل مخزن المتجهات Milvus.

async def async_add(num_adding):
    node_list = produce_nodes(num_adding)
    start_time = time.time()
    tasks = []
    for i in range(num_adding):
        sub_nodes = node_list[i]
        task = vector_store.async_add([sub_nodes])  # use async_add()
        tasks.append(task)
    results = await asyncio.gather(*tasks)
    end_time = time.time()
    return end_time - start_time

add_counts = [10, 100, 1000]

الحصول على حلقة الأحداث.

loop = asyncio.get_event_loop()

أضف المستندات إلى مخزن المتجهات بشكل غير متزامن.

for count in add_counts:

    async def measure_async_add():
        async_time = await async_add(count)
        print(f"Async add for {count} took {async_time:.2f} seconds")
        return async_time

    loop.run_until_complete(measure_async_add())

Async add for 10 took 0.19 seconds
Async add for 100 took 0.48 seconds
Async add for 1000 took 3.22 seconds

vector_store = init_vector_store()

2025-01-24 20:07:45,554 [DEBUG][_create_connection]: Created new connection using: b14dde8d6d24489bba26a907593f692d (async_milvus_client.py:600)

قارن ذلك مع الإضافة المتزامنة

نحدد دالة إضافة متزامنة. ثم نقيس وقت التشغيل في نفس الظروف.

def sync_add(num_adding):
    node_list = produce_nodes(num_adding)
    start_time = time.time()
    for node in node_list:
        result = vector_store.add([node])
    end_time = time.time()
    return end_time - start_time

for count in add_counts:
    sync_time = sync_add(count)
    print(f"Sync add for {count} took {sync_time:.2f} seconds")

Sync add for 10 took 0.56 seconds
Sync add for 100 took 5.85 seconds
Sync add for 1000 took 62.91 seconds

تُظهر النتيجة أن عملية الإضافة المتزامنة أبطأ بكثير من العملية غير المتزامنة.

البحث غير المتزامن

أعد تهيئة مخزن المتجهات وأضف بعض المستندات قبل تشغيل عملية البحث.

vector_store = init_vector_store()
node_list = produce_nodes(num_adding=1000)
inserted_ids = vector_store.add(node_list)

2025-01-24 20:08:57,982 [DEBUG][_create_connection]: Created new connection using: 351dc7ea4fb14d4386cfab02621ab7d1 (async_milvus_client.py:600)

حدد دالة بحث غير متزامن. نستخدم دالة aquery() في مثيل مخزن المتجهات Milvus.

async def async_search(num_queries):
    start_time = time.time()
    tasks = []
    for _ in range(num_queries):
        query = VectorStoreQuery(
            query_embedding=[0.5] * (DIM - 1) + [0.6], similarity_top_k=3
        )
        task = vector_store.aquery(query=query)  # use aquery()
        tasks.append(task)
    results = await asyncio.gather(*tasks)
    end_time = time.time()
    return end_time - start_time

query_counts = [10, 100, 1000]

البحث غير المتزامن من مخزن Milvus.

for count in query_counts:

    async def measure_async_search():
        async_time = await async_search(count)
        print(f"Async search for {count} queries took {async_time:.2f} seconds")
        return async_time

    loop.run_until_complete(measure_async_search())

Async search for 10 queries took 0.55 seconds
Async search for 100 queries took 1.39 seconds
Async search for 1000 queries took 8.81 seconds

قارن ذلك بالبحث المتزامن

تحديد دالة بحث متزامن. ثم قياس وقت التشغيل في نفس الظروف.

def sync_search(num_queries):
    start_time = time.time()
    for _ in range(num_queries):
        query = VectorStoreQuery(
            query_embedding=[0.5] * (DIM - 1) + [0.6], similarity_top_k=3
        )
        result = vector_store.query(query=query)
    end_time = time.time()
    return end_time - start_time

for count in query_counts:
    sync_time = sync_search(count)
    print(f"Sync search for {count} queries took {sync_time:.2f} seconds")

Sync search for 10 queries took 3.29 seconds
Sync search for 100 queries took 30.80 seconds
Sync search for 1000 queries took 308.80 seconds

تُظهر النتيجة أن عملية البحث المتزامن أبطأ بكثير من عملية البحث غير المتزامن.