इस पेज का अनुवाद Cloud Translation API से किया गया है.

ऐडवांस आरएजी तकनीकें

1. परिचय

खास जानकारी

Retrieval Augmented Generation (RAG) की मदद से, लार्ज लैंग्वेज मॉडल (एलएलएम) के जवाबों को बेहतर बनाया जाता है. इसके लिए, बाहरी जानकारी का इस्तेमाल किया जाता है. हालांकि, प्रोडक्शन के लिए तैयार RAG सिस्टम बनाने के लिए, सिर्फ़ वेक्टर सर्च से ज़्यादा की ज़रूरत होती है. आपको यह ऑप्टिमाइज़ करना होगा कि डेटा कैसे लिया जाता है, काम के नतीजों को कैसे रैंक किया जाता है, और उपयोगकर्ता की क्वेरी को कैसे प्रोसेस किया जाता है.

इस बड़े लैब में, Cloud SQL for PostgreSQL (pgvector के साथ एक्सटेंड किया गया) और Vertex AI का इस्तेमाल करके, एक मज़बूत RAG ऐप्लिकेशन बनाया जाएगा. आपको तीन ऐडवांस तकनीकों के बारे में बताया जाएगा:

टेक्स्ट को छोटे-छोटे हिस्सों में बांटने की रणनीतियां: आपको यह पता चलेगा कि टेक्स्ट को छोटे-छोटे हिस्सों में बांटने के अलग-अलग तरीकों (वर्ण, रिकर्सिव, टोकन) से, जानकारी वापस पाने की क्वालिटी पर क्या असर पड़ता है.
फिर से रैंक करना: खोज के नतीजों को बेहतर बनाने और "बीच में मौजूद जानकारी को नज़रअंदाज़ करना" समस्या को हल करने के लिए, Vertex AI Reranker को लागू करें.
क्वेरी ट्रांसफ़ॉर्मेशन: Gemini का इस्तेमाल करके, उपयोगकर्ता की क्वेरी को ऑप्टिमाइज़ किया जाएगा. इसके लिए, HyDE (हाइपोथेटिकल दस्तावेज़ एम्बेडिंग) और स्टेप-बैक प्रॉम्प्टिंग जैसी तकनीकों का इस्तेमाल किया जाएगा.

आपको क्या करना होगा

pgvector की मदद से, PostgreSQL के लिए Cloud SQL इंस्टेंस सेट अप करें.
डेटा इंटेक पाइपलाइन बनाएं. यह पाइपलाइन, कई रणनीतियों का इस्तेमाल करके टेक्स्ट को छोटे-छोटे हिस्सों में बांटती है और Cloud SQL में एम्बेडिंग सेव करती है.
सिमैंटिक सर्च करें और अलग-अलग चंकिंग के तरीकों से मिले नतीजों की क्वालिटी की तुलना करें.
प्रासंगिकता के आधार पर, वापस लाए गए दस्तावेज़ों का क्रम बदलने के लिए, Reranker को इंटिग्रेट करें.
एलएलएम की मदद से क्वेरी को बदलकर, मुश्किल सवालों के जवाब बेहतर तरीके से पाएं.

आपको क्या सीखने को मिलेगा

Vertex AI और Cloud SQL के साथ LangChain का इस्तेमाल करने का तरीका.
Character, Recursive, और Token टेक्स्ट स्प्लिटर का असर.
PostgreSQL में वेक्टर सर्च को लागू करने का तरीका.
रीरैंक करने के लिए, ContextualCompressionRetriever का इस्तेमाल कैसे करें.
HyDE और स्टेप-बैक प्रॉम्प्टिंग को लागू करने का तरीका.

2. प्रोजेक्ट सेटअप करना

Google खाता

अगर आपके पास पहले से कोई निजी Google खाता नहीं है, तो आपको Google खाता बनाना होगा.

ऑफ़िस या स्कूल वाले खाते के बजाय, निजी खाते का इस्तेमाल करें.

किसी निजी Google खाते का इस्तेमाल करके, Google Cloud Console में साइन इन करें.

बिलिंग चालू करें

Google Cloud के 500 रुपये के क्रेडिट रिडीम करें (ज़रूरी नहीं)

इस वर्कशॉप को चलाने के लिए, आपके पास कुछ क्रेडिट वाला बिलिंग खाता होना चाहिए. अगर आपको अपने बिलिंग सिस्टम का इस्तेमाल करना है, तो इस चरण को छोड़ा जा सकता है.

इस लिंक पर क्लिक करें और किसी निजी Google खाते से साइन इन करें. आपको इस तरह की विंडो दिखेगी:
अपने क्रेडिट ऐक्सेस करने के लिए यहां क्लिक करें बटन पर क्लिक करें. इससे आपको एक ऐसे पेज पर ले जाया जाएगा जहां आपको अपनी बिलिंग प्रोफ़ाइल सेट अप करनी होगी
पुष्टि करें पर क्लिक करें. अब आप Google Cloud Platform के ट्रायल बिलिंग खाते से कनेक्ट हो गए हैं.

निजी बिलिंग खाता सेट अप करना

अगर आपने Google Cloud क्रेडिट का इस्तेमाल करके बिलिंग सेट अप की है, तो इस चरण को छोड़ा जा सकता है.

निजी बिलिंग खाता सेट अप करने के लिए, Cloud Console में बिलिंग की सुविधा चालू करने के लिए यहां जाएं.

ध्यान दें:

इस लैब को पूरा करने में, Cloud संसाधनों पर 1 डॉलर से कम का खर्च आना चाहिए.
ज़्यादा शुल्क से बचने के लिए, इस लैब के आखिर में दिए गए निर्देशों का पालन करके संसाधनों को मिटाया जा सकता है.
नए उपयोगकर्ता, 300 डॉलर के मुफ़्त में आज़माने की सुविधा का फ़ायदा पा सकते हैं.

प्रोजेक्ट बनाना (ज़रूरी नहीं)

अगर आपके पास कोई ऐसा मौजूदा प्रोजेक्ट नहीं है जिसका इस्तेमाल आपको इस लैब के लिए करना है, तो यहां नया प्रोजेक्ट बनाएं.

3. Cloud Shell Editor खोलें

सीधे Cloud Shell Editor पर जाने के लिए, इस लिंक पर क्लिक करें
अगर आज किसी भी समय अनुमति देने के लिए कहा जाता है, तो जारी रखने के लिए अनुमति दें पर क्लिक करें.
अगर टर्मिनल स्क्रीन पर सबसे नीचे नहीं दिखता है, तो इसे खोलें:
- देखें पर क्लिक करें
- टर्मिनल पर क्लिक करें
टर्मिनल में, इस कमांड का इस्तेमाल करके अपना प्रोजेक्ट सेट करें:
```
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. एपीआई चालू करें

इस समाधान को बनाने के लिए, आपको Vertex AI, Cloud SQL, और Reranking सेवा के लिए कई Google Cloud API चालू करने होंगे.

टर्मिनल में, इन एपीआई को चालू करें:

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 API (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. PostgreSQL के लिए Cloud SQL सेट अप करना

इस टास्क में, 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
```
अपने शेल सेशन में वैरिएबल लोड करें:
```
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 इंस्टेंस बनाने में करीब पांच से सात मिनट लग सकते हैं. कृपया थोड़ा इंतज़ार करें.

इंस्टेंस तैयार होने के बाद, डेटाबेस बनाएं:

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. पहला हिस्सा: जानकारी को छोटे-छोटे हिस्सों में बांटने की रणनीतियां

किसी भी RAG पाइपलाइन का पहला चरण, दस्तावेज़ों को ऐसे फ़ॉर्मैट में बदलना होता है जिसे एलएलएम समझ सके: चंक.

एलएलएम में कॉन्टेक्स्ट विंडो की सीमा होती है. इसका मतलब है कि वे एक बार में सिर्फ़ तय मात्रा में टेक्स्ट प्रोसेस कर सकते हैं. इसके अलावा, किसी सवाल का जवाब देने के लिए 50 पेज का दस्तावेज़ पाने से, जानकारी कम हो जाती है. हम दस्तावेज़ों को छोटे-छोटे "टुकड़ों" में बांटते हैं, ताकि काम की जानकारी को अलग किया जा सके.

हालांकि, टेक्स्ट को कैसे बांटा जाता है, यह बहुत मायने रखता है:

वर्णों के हिसाब से बांटने वाला स्प्लिटर: यह स्प्लिटर, वर्णों की संख्या के हिसाब से टेक्स्ट को बांटता है. यह तरीका तेज़ है, लेकिन जोखिम भरा है. इससे शब्दों या वाक्यों को आधा काटा जा सकता है, जिससे उनका मतलब बदल जाता है.
रिकर्सिव स्प्लिटर: यह सबसे पहले पैराग्राफ़ के हिसाब से, फिर वाक्य के हिसाब से, और आखिर में शब्द के हिसाब से टेक्स्ट को बांटने की कोशिश करता है. यह सिमैंटिक यूनिट को एक साथ रखने की कोशिश करता है.
टोकन स्प्लिटर: यह एलएलएम के अपने शब्दकोश (टोकन) के आधार पर टेक्स्ट को बांटता है. इससे यह पक्का किया जाता है कि कॉन्टेक्स्ट विंडो में चंक पूरी तरह से फ़िट हो जाएं. हालांकि, इन्हें जनरेट करने में ज़्यादा समय लग सकता है.

इस सेक्शन में, तीनों रणनीतियों का इस्तेमाल करके एक ही डेटा को शामिल किया जाएगा, ताकि उनकी तुलना की जा सके.

डेटा ट्रांसफ़र करने वाली स्क्रिप्ट बनाना

आपको एक ऐसी स्क्रिप्ट का इस्तेमाल करना होगा जो हैरी पॉटर के डेटासेट को डाउनलोड करती है. साथ ही, Character, Recursive, और Token रणनीतियों का इस्तेमाल करके उसे अलग-अलग हिस्सों में बांटती है. इसके बाद, एम्बेडिंग को 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. दूसरा भाग: फिर से रैंक करना

वेक्टर सर्च (जानकारी वापस पाना) बहुत तेज़ी से काम करती है, क्योंकि यह कंप्रेस किए गए गणितीय फ़ॉर्मैट (एम्बेडिंग) पर निर्भर करती है. यह ज़्यादा से ज़्यादा आइटम को शामिल करता है, ताकि रिकॉल (सभी संभावित तौर पर काम के आइटम ढूंढना) किया जा सके. हालांकि, इसमें अक्सर सटीकता कम होती है (उन आइटम की रैंकिंग सही नहीं होती).

अक्सर, काम के दस्तावेज़ नतीजों की सूची में "बीच में कहीं खो जाते हैं". एलएलएम, खोज के नतीजों में सबसे ऊपर दिखने वाले पांच नतीजों पर ध्यान देता है. इसलिए, हो सकता है कि वह सातवें नंबर पर मौजूद ज़रूरी जवाब को न देख पाए.

फिर से रैंक करना इस समस्या को हल करता है. इसके लिए, यह दूसरा चरण जोड़ता है.

रीट्रिवर: यह फ़ास्ट वेक्टर सर्च का इस्तेमाल करके, ज़्यादा बड़ा सेट (जैसे, टॉप 25) फ़ेच करता है.
रीरैंकर: यह क्वेरी और दस्तावेज़ों के जोड़े के पूरे टेक्स्ट की जांच करने के लिए, खास मॉडल (जैसे कि क्रॉस-एनकोडर) का इस्तेमाल करता है. यह ज़्यादा समय लेता है, लेकिन ज़्यादा सटीक होता है. यह सबसे ऊपर के 25 नतीजों को फिर से स्कोर करता है और सबसे बेहतर तीन नतीजे दिखाता है.

इस टास्क में, आपको पहले पार्ट में बनाए गए recursive कलेक्शन में खोजना होगा. हालांकि, इस बार आपको नतीजों को बेहतर बनाने के लिए, 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. तीसरा भाग: क्वेरी में बदलाव करना

अक्सर, 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. चौथा हिस्सा: एंड-टू-एंड जनरेशन

हमने अपने डेटा को छोटा किया है, खोज को बेहतर बनाया है, और उपयोगकर्ता की क्वेरी को और बेहतर बनाया है. अब हम RAG में "G" को शामिल करते हैं: जनरेशन.

अब तक, हमने सिर्फ़ जानकारी खोजने का काम किया है. एक बेहतरीन एआई असिस्टेंट बनाने के लिए, हमें उन अच्छी क्वालिटी वाले और फिर से रैंक किए गए दस्तावेज़ों को एलएलएम (Gemini) में फ़ीड करना होगा, ताकि वह आम बोलचाल की भाषा में जवाब दे सके.

प्रोडक्शन पाइपलाइन में, इसके लिए एक खास फ़्लो होता है:

वापस पाना: उम्मीदवारों का एक बड़ा सेट पाना (जैसे, टॉप 10) फ़ास्ट वेक्टर सर्च का इस्तेमाल करके.
फिर से रैंक करना: सबसे अच्छे नतीजे पाने के लिए फ़िल्टर करना (जैसे, टॉप 3) Vertex AI Reranker का इस्तेमाल करके.
संदर्भ तैयार करना: सबसे ज़्यादा काम के तीन दस्तावेज़ों के कॉन्टेंट को एक स्ट्रिंग में जोड़ें.
भरोसेमंद स्रोतों से जानकारी लेकर प्रॉम्प्ट तैयार करना: उस कॉन्टेक्स्ट स्ट्रिंग को प्रॉम्प्ट के ऐसे टेम्प्लेट में डालें जो एलएलएम को सिर्फ़ उस जानकारी का इस्तेमाल करने के लिए मजबूर करे.

जनरेशन स्क्रिप्ट बनाना

हम जनरेशन के चरण के लिए gemini-2.5-flash का इस्तेमाल करेंगे. यह मॉडल, RAG के लिए सबसे सही है. इसकी वजह यह है कि इसमें कॉन्टेक्स्ट विंडो बड़ी होती है और इंतज़ार का समय कम होता है. इससे यह मॉडल, खोजे गए कई दस्तावेज़ों को तेज़ी से प्रोसेस कर पाता है.

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

आउटपुट को समझना

इस स्क्रिप्ट को चलाने पर, आपको पिछले चरणों में मिले रॉ डेटा वाले चंक और फ़ाइनल जवाब के बीच का अंतर दिखेगा. एलएलएम, सिंथेसाइज़र के तौर पर काम करता है. यह रीरैंकर से मिले टेक्स्ट के "चंक" को पढ़ता है और उन्हें एक ऐसे वाक्य में बदलता है जिसे कोई भी व्यक्ति आसानी से पढ़ सकता है.

इन कॉम्पोनेंट को एक साथ इस्तेमाल करने से, आपको अनुमान लगाने के बजाय, सटीक और भरोसेमंद जानकारी मिलती है. Retriever, जानकारी इकट्ठा करता है. Reranker, सबसे सही जानकारी चुनता है. Generator, जवाब तैयार करता है.

11. नतीजा

बधाई हो! आपने एक बेहतर RAG पाइपलाइन बना ली है. यह बुनियादी वेक्टर सर्च से कहीं ज़्यादा बेहतर है.

रीकैप

आपने बड़े पैमाने पर वेक्टर स्टोरेज के लिए, pgvector के साथ Cloud SQL को कॉन्फ़िगर किया हो.
आपने डेटा को छोटे-छोटे हिस्सों में बांटने की रणनीतियों की तुलना की, ताकि यह समझा जा सके कि डेटा तैयार करने से डेटा को वापस पाने पर क्या असर पड़ता है.
आपने नतीजों को ज़्यादा सटीक बनाने के लिए, Vertex AI की मदद से रीरैंकिंग की सुविधा लागू की है.
आपने उपयोगकर्ता के इंटेंट को अपने डेटा के साथ अलाइन करने के लिए, क्वेरी ट्रांसफ़ॉर्मेशन (HyDE, स्टेप-बैक) का इस्तेमाल किया हो.

ज़्यादा जानें

आरएजी के रेफ़रंस के लिए डायग्राम: आरएजी से जुड़े रेफ़रंस के लिए डायग्राम की गाइड की सूची देखें.

प्रोटोटाइप से लेकर प्रोडक्शन तक

यह लैब, Google Cloud की मदद से प्रोडक्शन-रेडी एआई बनाने के बारे में जानकारी देने वाले लर्निंग पाथ का हिस्सा है.

प्रोटोटाइप से लेकर प्रोडक्शन तक के सभी चरणों के बारे में जानने के लिए, पूरा पाठ्यक्रम देखें.
#ProductionReadyAI हैशटैग का इस्तेमाल करके, अपनी प्रोग्रेस शेयर करें.

ऐडवांस आरएजी तकनीकें

1. परिचय

खास जानकारी

आपको क्या करना होगा

आपको क्या सीखने को मिलेगा

2. प्रोजेक्ट सेटअप करना

Google खाता

Google Cloud Console में साइन इन करना

बिलिंग चालू करें

Google Cloud के 500 रुपये के क्रेडिट रिडीम करें (ज़रूरी नहीं)

निजी बिलिंग खाता सेट अप करना

प्रोजेक्ट बनाना (ज़रूरी नहीं)

3. Cloud Shell Editor खोलें

4. एपीआई चालू करें

एपीआई के बारे में जानकारी

5. वर्चुअल एनवायरमेंट बनाना और डिपेंडेंसी इंस्टॉल करना

6. PostgreSQL के लिए Cloud SQL सेट अप करना

Cloud SQL कॉन्फ़िगरेशन तय करना

इंस्टेंस और डेटाबेस बनाना

pgvector एक्सटेंशन चालू करना

7. पहला हिस्सा: जानकारी को छोटे-छोटे हिस्सों में बांटने की रणनीतियां

डेटा ट्रांसफ़र करने वाली स्क्रिप्ट बनाना

चंकिंग के नतीजों की तुलना करना

आउटपुट देखें.

8. दूसरा भाग: फिर से रैंक करना

निगरानी करें

9. तीसरा भाग: क्वेरी में बदलाव करना

आउटपुट देखें.

10. चौथा हिस्सा: एंड-टू-एंड जनरेशन

जनरेशन स्क्रिप्ट बनाना

आउटपुट को समझना

11. नतीजा

रीकैप

ज़्यादा जानें

प्रोटोटाइप से लेकर प्रोडक्शन तक