تمت ترجمة هذه الصفحة بواسطة Cloud Translation API‏.

تقنيات التوليد المعزّز بالاسترجاع المتقدّمة

1- مقدمة

نظرة عامة

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

في هذا التمرين المعملي الشامل، ستنشئ تطبيقًا قويًا للتوليد المعزّز بالاسترجاع باستخدام Cloud SQL for PostgreSQL (الموسّع باستخدام pgvector) وVertex AI. ستنتقل إلى ثلاث تقنيات متقدّمة:

استراتيجيات التقسيم: ستلاحظ كيف تؤثر الطرق المختلفة لتقسيم النص (الحرفي، والتكراري، والرمزي) في جودة الاسترجاع.
إعادة الترتيب: ستستخدم Vertex AI Reranker لتحسين نتائج البحث ومعالجة مشكلة "الضياع في المنتصف".
تحويل طلب البحث: ستستخدم Gemini لتحسين طلبات بحث المستخدمين من خلال تقنيات مثل HyDE (تضمين المستندات الافتراضية) والتحليل التدريجي.

الإجراءات التي ستنفذّها

إعداد مثيل Cloud SQL for PostgreSQL باستخدام pgvector
إنشاء مسار نقل بيانات يقسّم النص إلى أجزاء باستخدام استراتيجيات متعددة ويخزّن عمليات التضمين في Cloud SQL
إجراء عمليات بحث دلالية ومقارنة جودة النتائج من طرق تقسيم مختلفة
دمج أداة إعادة ترتيب لترتيب المستندات التي تم استرجاعها استنادًا إلى مدى صلتها بموضوع البحث
تنفيذ عمليات تحويل طلبات البحث المستندة إلى نماذج لغوية كبيرة لتحسين عملية الاسترجاع للأسئلة الغامضة أو المعقّدة

أهداف الدورة التعليمية

كيفية استخدام LangChain مع Vertex AI وCloud SQL
تأثير أدوات تقسيم النصوص Character وRecursive وToken
كيفية تنفيذ البحث المتّجهي في PostgreSQL
كيفية استخدام ContextualCompressionRetriever لإعادة الترتيب
كيفية تنفيذ HyDE وStep-back Prompting

2. إعداد المشروع

حساب Google

إذا لم يكن لديك حساب شخصي على Google، عليك إنشاء حساب على Google.

استخدام حساب شخصي بدلاً من حساب تابع للعمل أو تديره مؤسسة تعليمية

سجِّل الدخول إلى Google Cloud Console باستخدام حساب Google شخصي.

تفعيل الفوترة

تحصيل قيمة رصيد Google Cloud بقيمة 5 دولار أمريكي (اختياري)

لإجراء ورشة العمل هذه، تحتاج إلى حساب فوترة يتضمّن بعض الرصيد. إذا كنت تخطّط لاستخدام نظام الفوترة الخاص بك، يمكنك تخطّي هذه الخطوة.

انقر على هذا الرابط وسجِّل الدخول باستخدام حساب Google شخصي. سيظهر لك محتوى مشابه لما يلي:
انقر على الزر انقر هنا للوصول إلى أرصدتك. سيتم توجيهك إلى صفحة لإعداد ملف الفوترة
انقر على تأكيد. لقد تم ربط حسابك الآن بحساب فوترة تجريبي على Google Cloud Platform.

إعداد حساب فوترة شخصي

إذا أعددت الفوترة باستخدام أرصدة Google Cloud، يمكنك تخطّي هذه الخطوة.

لإعداد حساب فوترة شخصي، يُرجى الانتقال إلى هنا لتفعيل الفوترة في Cloud Console.

ملاحظات:

يجب أن تكلّف إكمال هذا المختبر أقل من دولار أمريكي واحد من موارد السحابة الإلكترونية.
يمكنك اتّباع الخطوات في نهاية هذا المختبر لحذف الموارد وتجنُّب المزيد من الرسوم.
يكون المستخدمون الجدد مؤهّلين للاستفادة من الفترة التجريبية المجانية بقيمة 300 دولار أمريكي.

إنشاء مشروع (اختياري)

إذا لم يكن لديك مشروع حالي تريد استخدامه في هذا المختبر، يمكنك إنشاء مشروع جديد هنا.

3- فتح Cloud Shell Editor

انقر على هذا الرابط للانتقال مباشرةً إلى محرّر Cloud Shell
إذا طُلب منك منح الإذن في أي وقت اليوم، انقر على منح الإذن للمتابعة.
إذا لم تظهر المحطة الطرفية في أسفل الشاشة، افتحها باتّباع الخطوات التالية:
- انقر على عرض.
- انقر على Terminal
في الوحدة الطرفية، اضبط مشروعك باستخدام الأمر التالي:
```
gcloud config set project [PROJECT_ID]
```
- مثال:
```
gcloud config set project lab-project-id-example
```
- إذا تعذّر عليك تذكُّر معرّف مشروعك، يمكنك إدراج جميع معرّفات المشاريع باستخدام:
```
gcloud projects list
```
من المفترض أن تظهر لك هذه الرسالة:
```
Updated property [core/project].
```
إذا ظهر لك الرمز WARNING وطُلب منك Do you want to continue (Y/n)?، من المحتمل أنّك أدخلت رقم تعريف المشروع بشكل غير صحيح. اضغط على n، ثم على Enter، وحاوِل تنفيذ الأمر gcloud config set project مرة أخرى.

4. تفعيل واجهات برمجة التطبيقات

لإنشاء هذا الحلّ، عليك تفعيل العديد من واجهات برمجة التطبيقات في Google Cloud لكلّ من Vertex AI وCloud SQL وخدمة إعادة الترتيب.

في الوحدة الطرفية، فعِّل واجهات برمجة التطبيقات:

gcloud services enable \
  aiplatform.googleapis.com \
  sqladmin.googleapis.com \
  cloudresourcemanager.googleapis.com \
  serviceusage.googleapis.com \
  discoveryengine.googleapis.com

لمحة عن واجهات برمجة التطبيقات

واجهة برمجة التطبيقات Vertex AI API (aiplatform.googleapis.com): تتيح استخدام Gemini لإنشاء المحتوى وVertex AI Embeddings لتحويل النص إلى متجهات.
واجهة برمجة التطبيقات Cloud SQL Admin API (sqladmin.googleapis.com): تتيح لك إدارة مثيلات Cloud SQL آليًا.
Discovery Engine API (discoveryengine.googleapis.com): تتيح إمكانات Vertex AI Reranker.
واجهة برمجة التطبيقات Service Usage (serviceusage.googleapis.com): مطلوبة للتحقّق من حصص الخدمة وإدارتها.

5- إنشاء بيئة افتراضية وتثبيت التبعيات

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

أنشئ مجلدًا باسم rag-labs وانتقِل إليه. نفِّذ الرمز التالي في الوحدة الطرفية:
```
mkdir rag-labs && cd rag-labs
```

أنشئ بيئة افتراضية وفعِّلها:

uv venv --python 3.12
source .venv/bin/activate

أنشئ ملف requirements.txt يتضمّن التبعيات اللازمة. نفِّذ الرمز التالي في الوحدة الطرفية:
```
cloudshell edit requirements.txt
```

الصِق التبعيات المحسَّنة التالية في requirements.txt. يتم تثبيت هذه الإصدارات لتجنُّب التعارضات وتسريع عملية التثبيت.

# Core LangChain & AI
langchain-community==0.3.31
langchain-google-vertexai==2.1.2
langchain-google-community[vertexaisearch]==2.0.10

# Google Cloud
google-cloud-storage==2.19.0
google-cloud-aiplatform[langchain]==1.130.0

# Database
cloud-sql-python-connector[pg8000]==1.19.0
sqlalchemy==2.0.45
pgvector==0.4.2

# Utilities
tiktoken==0.12.0
python-dotenv==1.2.1
requests==2.32.5

ثبِّت التبعيات:
```
uv pip install -r requirements.txt
```

6. إعداد Cloud SQL for PostgreSQL

في هذه المهمة، ستوفّر مثيلاً من Cloud SQL for PostgreSQL، وتنشئ قاعدة بيانات، وتجهّزها للبحث المتّجه.

تحديد إعدادات Cloud SQL

أنشئ ملف .env لتخزين الإعدادات. نفِّذ الرمز التالي في الوحدة الطرفية:
```
cloudshell edit .env
```

الصِق الإعداد التالي في .env.

# Project Config
PROJECT_ID="[YOUR_PROJECT_ID]"
REGION="us-central1"

# Database Config
SQL_INSTANCE_NAME="rag-pg-instance-1"
SQL_DATABASE_NAME="rag_harry_potter_db"
SQL_USER="rag_user"
SQL_PASSWORD="StrongPassword123!" 

# RAG Config
PGVECTOR_COLLECTION_NAME="rag_harry_potter"
RANKING_LOCATION_ID="global"

# Connection Name (Auto-generated in scripts usually, but useful to have)
DB_INSTANCE_CONNECTION_NAME="${PROJECT_ID}:${REGION}:${SQL_INSTANCE_NAME}"

استبدِل [YOUR_PROJECT_ID] برقم تعريف مشروع Google Cloud الفعلي. (مثلاً PROJECT_ID = "google-cloud-labs")
إذا لم تتمكّن من تذكُّر رقم تعريف مشروعك، نفِّذ الأمر التالي في الوحدة الطرفية. ستظهر لك قائمة بجميع مشاريعك وأرقام تعريفها.
```
gcloud projects list
```
حمِّل المتغيرات في جلسة shell:
```
source .env
```

إنشاء المثيل وقاعدة البيانات

أنشئ مثيلاً من Cloud SQL for PostgreSQL. ينشئ هذا الأمر مثيلاً صغيرًا مناسبًا لهذا التمرين العملي.
```
gcloud sql instances create ${SQL_INSTANCE_NAME} \
  --database-version=POSTGRES_15 \
  --tier=db-g1-small \
  --region=${REGION} \
  --project=${PROJECT_ID}
```
ملاحظة: قد يستغرق إنشاء مثيل Cloud SQL جديد حوالي 5 إلى 7 دقائق. نحتاج إلى بعض الوقت لإتاحة هذه الميزة للجميع.
بعد أن يصبح المثيل جاهزًا، أنشئ قاعدة البيانات باتّباع الخطوات التالية:
```
gcloud sql databases create ${SQL_DATABASE_NAME} \
  --instance=${SQL_INSTANCE_NAME} \
  --project=${PROJECT_ID}
```

أنشئ مستخدم قاعدة البيانات:

gcloud sql users create ${SQL_USER} \
  --instance=${SQL_INSTANCE_NAME} \
  --password=${SQL_PASSWORD} \
  --project=${PROJECT_ID}

تفعيل إضافة pgvector

تسمح إضافة pgvector لـ PostgreSQL بتخزين عمليات التضمين المتجهة والبحث عنها. يجب تفعيلها بشكل صريح في قاعدة البيانات.

أنشئ نصًا برمجيًا باسم enable_pgvector.py. نفِّذ الرمز التالي في الوحدة الطرفية:
```
cloudshell edit enable_pgvector.py
```

ألصِق الرمز التالي في enable_pgvector.py. يربط هذا النص البرمجي بقاعدة البيانات ويشغّل CREATE EXTENSION IF NOT EXISTS vector;.

import os
import sqlalchemy
from google.cloud.sql.connector import Connector, IPTypes
import logging
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO)

# Config
project_id = os.getenv("PROJECT_ID")
region = os.getenv("REGION")
instance_name = os.getenv("SQL_INSTANCE_NAME")
db_user = os.getenv("SQL_USER")
db_pass = os.getenv("SQL_PASSWORD")
db_name = os.getenv("SQL_DATABASE_NAME")
instance_connection_name = f"{project_id}:{region}:{instance_name}"

def getconn():
    with Connector() as connector:
        conn = connector.connect(
            instance_connection_name,
            "pg8000",
            user=db_user,
            password=db_pass,
            db=db_name,
            ip_type=IPTypes.PUBLIC,
        )
        return conn

def enable_pgvector():
    pool = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn,
    )
    with pool.connect() as db_conn:
        # Check if extension exists
        result = db_conn.execute(sqlalchemy.text("SELECT extname FROM pg_extension WHERE extname = 'vector';")).fetchone()
        if result:
            logging.info("pgvector extension is already enabled.")
        else:
            logging.info("Enabling pgvector extension...")
            db_conn.execute(sqlalchemy.text("CREATE EXTENSION IF NOT EXISTS vector;"))
            db_conn.commit()
            logging.info("pgvector extension enabled successfully.")

if __name__ == "__main__":
    enable_pgvector()

شغِّل النص البرمجي:
```
python enable_pgvector.py
```

7. الجزء 1: استراتيجيات التقسيم

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

تفرض النماذج اللغوية الكبيرة حدًا أقصى لقدرة الاستيعاب (مقدار النص الذي يمكنها معالجته في وقت واحد). بالإضافة إلى ذلك، يؤدي استرداد مستند من 50 صفحة للإجابة عن سؤال محدّد إلى تخفيف المعلومات. نقسم المستندات إلى "أجزاء" أصغر لعزل المعلومات ذات الصلة.

ومع ذلك، فإنّ طريقة تقسيم النص مهمة للغاية:

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

في هذا القسم، ستستوعب البيانات نفسها باستخدام الاستراتيجيات الثلاث كلها لمقارنتها.

إنشاء نص الاستيعاب البرمجي

ستستخدم نصًا برمجيًا ينزّل مجموعة بيانات Harry Potter، ويقسّمها باستخدام استراتيجيات الأحرف والتكرار والرموز المميزة، ويحمّل عمليات التضمين إلى ثلاثة جداول منفصلة في Cloud SQL.

أنشئ الملف ingest_data.py:
```
cloudshell edit ingest_data.py
```

ألصِق الرمز الثابت التالي في ingest_data.py. يحلّل هذا الإصدار بنية JSON لمجموعة البيانات بشكلٍ صحيح.

import os
import json
import logging
import requests
from typing import List, Dict, Any
from dotenv import load_dotenv

from google.cloud.sql.connector import Connector, IPTypes
from langchain_google_vertexai import VertexAIEmbeddings
from langchain_community.vectorstores import PGVector
from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter, TokenTextSplitter
from langchain.docstore.document import Document

load_dotenv()
logging.basicConfig(level=logging.INFO)

# Configuration
PROJECT_ID = os.getenv("PROJECT_ID")
REGION = os.getenv("REGION")
DB_USER = os.getenv("SQL_USER")
DB_PASS = os.getenv("SQL_PASSWORD")
DB_NAME = os.getenv("SQL_DATABASE_NAME")
INSTANCE_CONNECTION_NAME = f"{PROJECT_ID}:{REGION}:{os.getenv('SQL_INSTANCE_NAME')}"
BASE_COLLECTION_NAME = os.getenv("PGVECTOR_COLLECTION_NAME")
BOOKS_JSON_URL = "https://storage.googleapis.com/github-repo/generative-ai/gemini/reasoning-engine/sample_data/harry_potter_books.json"

CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
MAX_DOCS_TO_PROCESS = 10 

# Database Connector
def getconn():
    with Connector() as connector:
        return connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pg8000",
            user=DB_USER,
            password=DB_PASS,
            db=DB_NAME,
            ip_type=IPTypes.PUBLIC,
        )

def download_data():
    logging.info(f"Downloading data from {BOOKS_JSON_URL}...")
    response = requests.get(BOOKS_JSON_URL)
    return response.json()

def prepare_chunks(json_data, strategy):
    documents = []

    # Iterate through the downloaded data
    for entry in json_data[:MAX_DOCS_TO_PROCESS]:

        # --- JSON PARSING LOGIC ---
        # The data structure nests content inside 'kwargs' -> 'page_content'
        if "kwargs" in entry and "page_content" in entry["kwargs"]:
            content = entry["kwargs"]["page_content"]

            # Extract metadata if available, ensuring it's a dict
            metadata = entry["kwargs"].get("metadata", {})
            if not isinstance(metadata, dict):
                metadata = {"source": "unknown"}

            # Add the strategy to metadata for tracking
            metadata["strategy"] = strategy
        else:
            continue

        if not content:
            continue

        # Choose the splitter based on the strategy
        if strategy == "character":
            splitter = CharacterTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separator="\n")
        elif strategy == "token":
            splitter = TokenTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)
        else: # default to recursive
            splitter = RecursiveCharacterTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)

        # Split the content into chunks
        chunks = splitter.split_text(content)

        # Create Document objects for each chunk
        for chunk in chunks:
            documents.append(Document(page_content=chunk, metadata=metadata))

    return documents

def main():
    logging.info("Initializing Embeddings...")
    embeddings = VertexAIEmbeddings(model_name="gemini-embedding-001", project=PROJECT_ID, location=REGION)

    data = download_data()
    strategies = ["character", "recursive", "token"]

    # Connection string for PGVector (uses the getconn helper)
    pg_conn_str = f"postgresql+pg8000://{DB_USER}:{DB_PASS}@placeholder/{DB_NAME}"

    for strategy in strategies:
        collection_name = f"{BASE_COLLECTION_NAME}_{strategy}"
        logging.info(f"--- Processing strategy: {strategy.upper()} ---")
        logging.info(f"Target Collection: {collection_name}")

        # Prepare documents with the specific strategy
        docs = prepare_chunks(data, strategy)

        if not docs:
            logging.warning(f"No documents generated for strategy {strategy}. Check data source.")
            continue

        logging.info(f"Generated {len(docs)} chunks. Uploading to Cloud SQL...")

        # Initialize the Vector Store
        store = PGVector(
            collection_name=collection_name,
            embedding_function=embeddings,
            connection_string=pg_conn_str,
            engine_args={"creator": getconn},
            pre_delete_collection=True # Clears old data for this collection before adding new
        )

        # Batch add documents
        store.add_documents(docs)
        logging.info(f"Successfully finished {strategy}.\n")

if __name__ == "__main__":
    main()

شغِّل النص البرمجي الخاص بنقل البيانات. سيؤدي ذلك إلى ملء قاعدة البيانات بثلاثة جداول (مجموعات) مختلفة.
```
python ingest_data.py
```

مقارنة نتائج تقسيم المحتوى

بعد تحميل البيانات، لننفّذ طلب بحث على جميع المجموعات الثلاث لمعرفة كيف تؤثّر استراتيجية التقسيم في النتائج.

إنشاء query_chunking.py:
```
cloudshell edit query_chunking.py
```

ألصِق الرمز التالي في query_chunking.py:

import os
import logging
from dotenv import load_dotenv
from google.cloud.sql.connector import Connector, IPTypes
from langchain_google_vertexai import VertexAIEmbeddings
from langchain_community.vectorstores import PGVector

load_dotenv()
logging.basicConfig(level=logging.ERROR) # Only show errors to keep output clean

# Config
PROJECT_ID = os.getenv("PROJECT_ID")
REGION = os.getenv("REGION")
DB_USER = os.getenv("SQL_USER")
DB_PASS = os.getenv("SQL_PASSWORD")
DB_NAME = os.getenv("SQL_DATABASE_NAME")
INSTANCE_CONNECTION_NAME = f"{PROJECT_ID}:{REGION}:{os.getenv('SQL_INSTANCE_NAME')}"
BASE_COLLECTION_NAME = os.getenv("PGVECTOR_COLLECTION_NAME")

def getconn():
    with Connector() as connector:
        return connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pg8000",
            user=DB_USER,
            password=DB_PASS,
            db=DB_NAME,
            ip_type=IPTypes.PUBLIC,
        )

def main():
    embeddings = VertexAIEmbeddings(model_name="gemini-embedding-001", project=PROJECT_ID, location=REGION)
    pg_conn_str = f"postgresql+pg8000://{DB_USER}:{DB_PASS}@placeholder/{DB_NAME}"

    query = "Tell me about the Dursleys and their relationship with Harry Potter"
    print(f"\nQUERY: {query}\n" + "="*50)

    strategies = ["character", "recursive", "token"]

    for strategy in strategies:
        collection = f"{BASE_COLLECTION_NAME}_{strategy}"
        print(f"\nSTRATEGY: {strategy.upper()}")

        store = PGVector(
            collection_name=collection,
            embedding_function=embeddings,
            connection_string=pg_conn_str,
            engine_args={"creator": getconn}
        )

        results = store.similarity_search_with_score(query, k=2)
        for i, (doc, score) in enumerate(results):
            print(f"  Result {i+1} (Score: {score:.4f}): {doc.page_content[:150].replace(chr(10), ' ')}...")

if __name__ == "__main__":
    main()

نفِّذ نص الاستعلام البرمجي:
```
python query_chunking.py
```

مراقبة الناتج

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

8. الجزء 2: إعادة الترتيب

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

في كثير من الأحيان، "تضيع" المستندات ذات الصلة في منتصف قائمة النتائج. قد لا يرصد النموذج اللغوي الكبير الإجابة المهمة التي تظهر في الموضع رقم 7 إذا كان يركّز على أهم 5 نتائج.

تتغلّب إعادة الترتيب على هذه المشكلة من خلال إضافة مرحلة ثانية.

أداة الاسترجاع: تسترجع مجموعة أكبر (مثل أفضل 25 نتيجة) باستخدام البحث السريع عن المتجهات.
أداة إعادة الترتيب: تستخدم نموذجًا متخصصًا (مثل Cross-Encoder) لفحص النص الكامل لطلبات البحث وأزواج المستندات. وهي أبطأ ولكنّها أكثر دقة. ويعيد تقييم أفضل 25 نتيجة ويعرض أفضل 3 نتائج على الإطلاق.

في هذه المهمة، ستبحث في مجموعة recursive التي تم إنشاؤها في الجزء 1، ولكن هذه المرة ستطبّق Vertex AI Reranker لتحسين النتائج.

إنشاء query_reranking.py:
```
cloudshell edit query_reranking.py
```

ألصِق الرمز التالي. لاحظ كيف يستهدف هذا المثال بشكل صريح عملية جمع _recursive واستخدام ContextualCompressionRetriever.

import os
import logging
from dotenv import load_dotenv
from google.cloud.sql.connector import Connector, IPTypes
from langchain_google_vertexai import VertexAIEmbeddings
from langchain_community.vectorstores import PGVector

# Reranking Imports
from langchain.retrievers import ContextualCompressionRetriever
from langchain_google_community.vertex_rank import VertexAIRank

load_dotenv()
logging.basicConfig(level=logging.ERROR)

PROJECT_ID = os.getenv("PROJECT_ID")
REGION = os.getenv("REGION")
DB_USER = os.getenv("SQL_USER")
DB_PASS = os.getenv("SQL_PASSWORD")
DB_NAME = os.getenv("SQL_DATABASE_NAME")
INSTANCE_CONNECTION_NAME = f"{PROJECT_ID}:{REGION}:{os.getenv('SQL_INSTANCE_NAME')}"

# IMPORTANT: Target the recursive collection created in ingest_data.py
COLLECTION_NAME = f"{os.getenv('PGVECTOR_COLLECTION_NAME')}_recursive"
RANKING_LOCATION = os.getenv("RANKING_LOCATION_ID")

def getconn():
    with Connector() as connector:
        return connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pg8000",
            user=DB_USER,
            password=DB_PASS,
            db=DB_NAME,
            ip_type=IPTypes.PUBLIC,
        )

def main():
    embeddings = VertexAIEmbeddings(model_name="gemini-embedding-001", project=PROJECT_ID, location=REGION)
    pg_conn_str = f"postgresql+pg8000://{DB_USER}:{DB_PASS}@placeholder/{DB_NAME}"

    print(f"Connecting to collection: {COLLECTION_NAME}")
    store = PGVector(
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings,
        connection_string=pg_conn_str,
        engine_args={"creator": getconn}
    )

    query = "What are the Horcruxes?"
    print(f"QUERY: {query}\n")

    # 1. Base Retriever (Vector Search) - Fetch top 10
    base_retriever = store.as_retriever(search_kwargs={"k": 10})

    # 2. Reranker - Select top 3 from the 10
    reranker = VertexAIRank(
        project_id=PROJECT_ID,
        location_id=RANKING_LOCATION,
        ranking_config="default_ranking_config",
        title_field="source",
        top_n=3
    )

    compression_retriever = ContextualCompressionRetriever(
        base_compressor=reranker,
        base_retriever=base_retriever
    )

    # Execute
    try:
        reranked_docs = compression_retriever.invoke(query)

        if not reranked_docs:
            print("No documents returned. Check if the collection exists and is populated.")

        print(f"--- Top 3 Reranked Results ---")
        for i, doc in enumerate(reranked_docs):
            print(f"Result {i+1} (Score: {doc.metadata.get('relevance_score', 'N/A')}):")
            print(f"  {doc.page_content[:200]}...\n")
    except Exception as e:
        print(f"Error during reranking: {e}")

if __name__ == "__main__":
    main()

نفِّذ طلب إعادة الترتيب:
```
python query_reranking.py
```

المراقبة

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

9- الجزء 3: تحويل طلب البحث

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

تستخدم ميزة تحويل الاستعلام نموذجًا لغويًا كبيرًا لإعادة كتابة الاستعلام أو توسيعه قبل أن يصل إلى قاعدة البيانات. ستنفّذ طريقتَين:

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

إنشاء query_transformation.py:

cloudshell edit query_transformation.py

ألصِق الرمز التالي:

import os
import logging
from dotenv import load_dotenv
from google.cloud.sql.connector import Connector, IPTypes
from langchain_google_vertexai import VertexAIEmbeddings, VertexAI
from langchain_community.vectorstores import PGVector
from langchain_core.prompts import PromptTemplate

load_dotenv()
logging.basicConfig(level=logging.ERROR)

PROJECT_ID = os.getenv("PROJECT_ID")
REGION = os.getenv("REGION")
DB_USER = os.getenv("SQL_USER")
DB_PASS = os.getenv("SQL_PASSWORD")
DB_NAME = os.getenv("SQL_DATABASE_NAME")
INSTANCE_CONNECTION_NAME = f"{PROJECT_ID}:{REGION}:{os.getenv('SQL_INSTANCE_NAME')}"
COLLECTION_NAME = f"{os.getenv('PGVECTOR_COLLECTION_NAME')}_recursive"

def getconn():
    with Connector() as connector:
        return connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pg8000",
            user=DB_USER,
            password=DB_PASS,
            db=DB_NAME,
            ip_type=IPTypes.PUBLIC,
        )

def generate_hyde_doc(query, llm):
    prompt = PromptTemplate(
        input_variables=["question"],
        template="Write a concise, hypothetical answer to the question. Question: {question} Answer:"
    )
    chain = prompt | llm
    return chain.invoke({"question": query})

def generate_step_back(query, llm):
    prompt = PromptTemplate(
        input_variables=["question"],
        template="Write a more general, abstract question that concepts in this question. Original: {question} Step-back:"
    )
    chain = prompt | llm
    return chain.invoke({"question": query})

def main():
    embeddings = VertexAIEmbeddings(model_name="gemini-embedding-001", project=PROJECT_ID, location=REGION)
    llm = VertexAI(model_name="gemini-2.5-flash", project=PROJECT_ID, location=REGION, temperature=0.5)

    pg_conn_str = f"postgresql+pg8000://{DB_USER}:{DB_PASS}@placeholder/{DB_NAME}"
    store = PGVector(
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings,
        connection_string=pg_conn_str,
        engine_args={"creator": getconn}
    )
    retriever = store.as_retriever(search_kwargs={"k": 2})

    original_query = "Tell me about the Dursleys."
    print(f"ORIGINAL QUERY: {original_query}\n" + "-"*30)

    # 1. HyDE
    hyde_doc = generate_hyde_doc(original_query, llm)
    print(f"HyDE Generated Doc: {hyde_doc.strip()[:100]}...")
    hyde_results = retriever.invoke(hyde_doc)
    print(f"HyDE Retrieval: {hyde_results[0].page_content[:100]}...\n")

    # 2. Step-back
    step_back_q = generate_step_back(original_query, llm)
    print(f"Step-back Query: {step_back_q.strip()}")
    step_results = retriever.invoke(step_back_q)
    print(f"Step-back Retrieval: {step_results[0].page_content[:100]}...")

if __name__ == "__main__":
    main()

شغِّل نص التحويل البرمجي:
```
python query_transformation.py
```

مراقبة الناتج

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

10. الجزء 4: إنشاء المحتوى من البداية إلى النهاية

لقد قسّمنا البيانات وحسّنّا البحث وصقلنا طلب البحث الذي أدخله المستخدم. الآن، ننتقل إلى الخطوة الأخيرة في عملية RAG، وهي التوليد.

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

في مسار الإنتاج، يتضمّن ذلك سير عمل محدّدًا:

الاسترجاع: الحصول على مجموعة واسعة من المستندات المرشّحة (مثل أفضل 10) باستخدام البحث السريع عن المتجهات.
إعادة الترتيب: فلترة النتائج لتحديد الأفضل على الإطلاق (مثلاً: أفضل 3) باستخدام أداة Vertex AI Reranker.
إنشاء السياق: ادمج محتوى أهم 3 مستندات في سلسلة واحدة.
الطلبات المستندة إلى مصادر: أدخِل سلسلة السياق هذه في نموذج طلب صارم يفرض على النموذج اللغوي الكبير استخدام هذه المعلومات فقط.

إنشاء نص برمجي للإنشاء

سنستخدم gemini-2.5-flash في خطوة الإنشاء. هذا النموذج مثالي للتوليد المعزَّز بالاسترجاع لأنّه يتضمّن نافذة سياق طويلة ووقت استجابة منخفض، ما يتيح له معالجة مستندات متعدّدة تم استرجاعها بسرعة.

إنشاء end_to_end_rag.py:

cloudshell edit end_to_end_rag.py

ألصِق الرمز التالي. انتبه إلى المتغيّر template، فهو المكان الذي نطلب فيه من النموذج بشكل صارم تجنُّب "الهلوسات" (أي اختلاق معلومات) من خلال ربطه بالسياق المقدَّم.

import os
import logging
from dotenv import load_dotenv
from google.cloud.sql.connector import Connector, IPTypes
from langchain_google_vertexai import VertexAIEmbeddings, VertexAI
from langchain_community.vectorstores import PGVector
from langchain.retrievers import ContextualCompressionRetriever
from langchain_google_community.vertex_rank import VertexAIRank
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

load_dotenv()
logging.basicConfig(level=logging.ERROR)

PROJECT_ID = os.getenv("PROJECT_ID")
REGION = os.getenv("REGION")
# We use the recursive collection as it generally provides the best context boundaries
COLLECTION_NAME = f"{os.getenv('PGVECTOR_COLLECTION_NAME')}_recursive"

def getconn():
    instance_conn = f"{PROJECT_ID}:{REGION}:{os.getenv('SQL_INSTANCE_NAME')}"
    with Connector() as connector:
        return connector.connect(
            instance_conn, "pg8000",
            user=os.getenv("SQL_USER"), password=os.getenv("SQL_PASSWORD"),
            db=os.getenv("SQL_DATABASE_NAME"), ip_type=IPTypes.PUBLIC
        )

def main():
    print("--- Initializing Production RAG Pipeline ---")

    # 1. Setup Embeddings (Gemini Embedding 001)
    # We use this to vectorize the user's query to match our database.
    embeddings = VertexAIEmbeddings(model_name="gemini-embedding-001", project=PROJECT_ID, location=REGION)

    # 2. Connect to Vector Store
    pg_conn_str = f"postgresql+pg8000://{os.getenv('SQL_USER')}:{os.getenv('SQL_PASSWORD')}@placeholder/{os.getenv('SQL_DATABASE_NAME')}"
    store = PGVector(
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings,
        connection_string=pg_conn_str,
        engine_args={"creator": getconn}
    )

    # 3. Setup The 'Filter Funnel' (Retriever + Reranker)
    # Step A: Fast retrieval of top 10 similar documents
    base_retriever = store.as_retriever(search_kwargs={"k": 10})

    # Step B: Precise reranking to find the top 3 most relevant
    reranker = VertexAIRank(
        project_id=PROJECT_ID,
        location_id="global", 
        ranking_config="default_ranking_config",
        title_field="source",
        top_n=3
    )

    # Combine A and B into a single retrieval object
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=reranker,
        base_retriever=base_retriever
    )

    # 4. Setup LLM (Gemini 2.5 Flash)
    # We use a low temperature (0.1) to reduce creativity and increase factual adherence.
    llm = VertexAI(model_name="gemini-2.5-flash", project=PROJECT_ID, location=REGION, temperature=0.1)

    # --- Execution Loop ---
    user_query = "Who is Harry Potter?"
    print(f"\nUser Query: {user_query}")
    print("Retrieving and Reranking documents...")

    # Retrieve the most relevant documents
    top_docs = compression_retriever.invoke(user_query)

    if not top_docs:
        print("No relevant documents found.")
        return

    # Build the Context String
    # We stitch the documents together, labeling them as Source 1, Source 2, etc.
    context_str = "\n\n".join([f"Source {i+1}: {d.page_content}" for i, d in enumerate(top_docs)])

    print(f"Found {len(top_docs)} relevant context chunks.")

    # 5. The Grounded Prompt
    template = """You are a helpful assistant. Answer the question strictly based on the provided context.
    If the answer is not in the context, say "I don't know."

    Context:
    {context}

    Question:
    {question}

    Answer:
    """

    prompt = PromptTemplate(template=template, input_variables=["context", "question"])

    # Create the chain: Prompt -> LLM
    chain = prompt | llm

    print("Generating Answer via Gemini 2.5 Flash...")
    final_answer = chain.invoke({"context": context_str, "question": user_query})

    print(f"\nFINAL ANSWER:\n{final_answer}")

if __name__ == "__main__":
    main()

تشغيل التطبيق النهائي:

python end_to_end_rag.py

فهم الناتج

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

ومن خلال ربط هذه المكوّنات، يمكنك الانتقال من "تخمين" احتمالي إلى سير عمل حتمي ومستند إلى بيانات. يُلقي Retriever الشبكة، ويختار Reranker أفضل ما تم صيده، ويُعدّ Generator الوجبة.

11. الخاتمة

تهانينا! لقد أنشأت بنجاح مسار RAG متقدّمًا يتجاوز بكثير البحث الأساسي عن المتجهات.

ملخّص

لقد أعددت Cloud SQL باستخدام pgvector لتخزين المتجهات القابل للتوسيع.
قارنت استراتيجيات التقسيم لفهم كيفية تأثير إعداد البيانات في عملية الاسترجاع.
لقد نفّذت إعادة الترتيب باستخدام Vertex AI لتحسين دقة نتائجك.
استخدمت عمليات تحويل طلبات البحث (HyDE وStep-back) لمواءمة نية المستخدم مع بياناتك.

مزيد من المعلومات

البُنى المرجعية للتوليد المعزّز بالاسترجاع: يمكنك الاطّلاع على قائمة بأدلّة البُنى المرجعية المتعلّقة بالتوليد المعزّز بالاسترجاع.

من النموذج الأوّلي إلى الإنتاج

يشكّل هذا المختبر جزءًا من المسار التعليمي "الذكاء الاصطناعي الجاهز للإنتاج مع Google Cloud".

استكشاف المنهج الدراسي الكامل لسدّ الفجوة بين النموذج الأولي والإنتاج
شارِك مستوى تقدّمك باستخدام الهاشتاغ ‎ #ProductionReadyAI.

تقنيات التوليد المعزّز بالاسترجاع المتقدّمة

1- مقدمة

نظرة عامة

الإجراءات التي ستنفذّها

أهداف الدورة التعليمية

2. إعداد المشروع

حساب Google

تسجيل الدخول إلى Google Cloud Console

تفعيل الفوترة

تحصيل قيمة رصيد Google Cloud بقيمة 5 دولار أمريكي (اختياري)

إعداد حساب فوترة شخصي

إنشاء مشروع (اختياري)

3- فتح Cloud Shell Editor

4. تفعيل واجهات برمجة التطبيقات

لمحة عن واجهات برمجة التطبيقات

5- إنشاء بيئة افتراضية وتثبيت التبعيات

6. إعداد Cloud SQL for PostgreSQL

تحديد إعدادات Cloud SQL

إنشاء المثيل وقاعدة البيانات

تفعيل إضافة pgvector

7. الجزء 1: استراتيجيات التقسيم

إنشاء نص الاستيعاب البرمجي

مقارنة نتائج تقسيم المحتوى

مراقبة الناتج

8. الجزء 2: إعادة الترتيب

المراقبة

9- الجزء 3: تحويل طلب البحث

مراقبة الناتج

10. الجزء 4: إنشاء المحتوى من البداية إلى النهاية

إنشاء نص برمجي للإنشاء

فهم الناتج

11. الخاتمة

ملخّص

مزيد من المعلومات

من النموذج الأوّلي إلى الإنتاج