Halaman ini diterjemahkan oleh Cloud Translation API.

Teknik RAG Lanjutan

1. Pengantar

Ringkasan

Retrieval Augmented Generation (RAG) meningkatkan respons Model Bahasa Besar (LLM) dengan menghubungkannya ke pengetahuan eksternal. Namun, membangun sistem RAG siap produksi memerlukan lebih dari sekadar penelusuran vektor sederhana. Anda harus mengoptimalkan cara data dimasukkan, cara hasil yang relevan diberi peringkat, dan cara kueri pengguna diproses.

Di lab komprehensif ini, Anda akan membangun aplikasi RAG yang andal menggunakan Cloud SQL untuk PostgreSQL (diperluas dengan pgvector) dan Vertex AI. Anda akan mempelajari tiga teknik lanjutan:

Strategi Chunking: Anda akan mengamati pengaruh berbagai metode pemisahan teks (Karakter, Rekursif, Token) terhadap kualitas pengambilan.
Pemeringkatan ulang: Anda akan menerapkan Vertex AI Reranker untuk menyempurnakan hasil penelusuran dan mengatasi masalah "hilang di tengah".
Transformasi Kueri: Anda akan menggunakan Gemini untuk mengoptimalkan kueri pengguna melalui teknik seperti HyDE (Hypothetical Document Embeddings) dan Step-back Prompting.

Yang akan Anda lakukan

Siapkan instance Cloud SQL untuk PostgreSQL dengan pgvector.
Bangun pipeline penyerapan data yang membagi teks menggunakan beberapa strategi dan menyimpan sematan di Cloud SQL.
Lakukan penelusuran semantik dan bandingkan kualitas hasil dari berbagai metode pengelompokan.
Mengintegrasikan Reranker untuk mengurutkan ulang dokumen yang diambil berdasarkan relevansi.
Menerapkan transformasi kueri yang didukung LLM untuk meningkatkan pengambilan informasi untuk pertanyaan yang ambigu atau kompleks.

Yang akan Anda pelajari

Cara menggunakan LangChain dengan Vertex AI dan Cloud SQL.
Dampak pemisah teks Karakter, Rekursif, dan Token.
Cara menerapkan Vector Search di PostgreSQL.
Cara menggunakan ContextualCompressionRetriever untuk mengubah peringkat.
Cara menerapkan HyDE dan Step-back Prompting.

2. Penyiapan project

Akun Google

Jika belum memiliki Akun Google pribadi, Anda harus membuat Akun Google.

Gunakan akun pribadi, bukan akun kantor atau sekolah.

Aktifkan Penagihan

Tukarkan kredit Google Cloud senilai $5 (opsional)

Untuk menjalankan workshop ini, Anda memerlukan Akun Penagihan dengan sejumlah kredit. Jika Anda berencana menggunakan penagihan sendiri, Anda dapat melewati langkah ini.

Klik link ini dan login dengan Akun Google pribadi. Anda akan melihat sesuatu seperti ini:
Klik tombol KLIK DI SINI UNTUK MENGAKSES KREDIT ANDA. Anda akan diarahkan ke halaman untuk menyiapkan profil penagihan
Klik Konfirmasi. Anda kini terhubung ke Akun Penagihan Uji Coba Google Cloud Platform.

Menyiapkan akun penagihan pribadi

Jika menyiapkan penagihan menggunakan kredit Google Cloud, Anda dapat melewati langkah ini.

Untuk menyiapkan akun penagihan pribadi, buka di sini untuk mengaktifkan penagihan di Cloud Console.

Beberapa Catatan:

Menyelesaikan lab ini akan dikenai biaya kurang dari $1 USD untuk resource Cloud.
Anda dapat mengikuti langkah-langkah di akhir lab ini untuk menghapus resource agar tidak dikenai biaya lebih lanjut.
Pengguna baru memenuhi syarat untuk mengikuti Uji Coba Gratis senilai$300 USD.

Membuat project (opsional)

Jika Anda tidak memiliki project saat ini yang ingin digunakan untuk lab ini, buat project baru di sini.

3. Buka Cloud Shell Editor

Klik link ini untuk langsung membuka Cloud Shell Editor
Jika diminta untuk memberikan otorisasi kapan saja hari ini, klik Authorize untuk melanjutkan.
Jika terminal tidak muncul di bagian bawah layar, buka terminal:
- Klik Lihat
- Klik Terminal
Di terminal, tetapkan project Anda dengan perintah ini:
```
gcloud config set project [PROJECT_ID]
```
- Contoh:
```
gcloud config set project lab-project-id-example
```
- Jika tidak ingat project ID, Anda dapat mencantumkan semua project ID dengan:
```
gcloud projects list
```
Anda akan melihat pesan ini:
```
Updated property [core/project].
```
Jika Anda melihat WARNING dan diminta Do you want to continue (Y/n)?, berarti Anda kemungkinan telah memasukkan ID project dengan salah. Tekan n, tekan Enter, lalu coba jalankan perintah gcloud config set project lagi.

4. Mengaktifkan API

Untuk membangun solusi ini, Anda perlu mengaktifkan beberapa Google Cloud API untuk Vertex AI, Cloud SQL, dan layanan Reranking.

Di terminal, aktifkan API:

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

Memperkenalkan API

Vertex AI API (aiplatform.googleapis.com): Memungkinkan penggunaan Gemini untuk pembuatan dan Vertex AI Embeddings untuk memvektorisasi teks.
Cloud SQL Admin API (sqladmin.googleapis.com): Memungkinkan Anda mengelola instance Cloud SQL secara terprogram.
Discovery Engine API (discoveryengine.googleapis.com): Mendukung kemampuan Pengubah Peringkat Vertex AI.
Service Usage API (serviceusage.googleapis.com): Diperlukan untuk memeriksa dan mengelola kuota layanan.

5. Buat lingkungan virtual & instal dependensi

Sebelum memulai project Python, sebaiknya buat lingkungan virtual. Hal ini mengisolasi dependensi project, sehingga mencegah konflik dengan project lain atau paket Python global sistem.

Buat folder bernama rag-labs, lalu pindah ke folder tersebut. Jalankan kode berikut di terminal:
```
mkdir rag-labs && cd rag-labs
```

Buat dan aktifkan lingkungan virtual:

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

Buat file requirements.txt dengan dependensi yang diperlukan. Jalankan kode berikut di terminal:
```
cloudshell edit requirements.txt
```

Tempel dependensi yang dioptimalkan berikut ke requirements.txt. Versi ini disematkan untuk menghindari konflik dan mempercepat penginstalan.

# 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

Instal dependensinya:
```
uv pip install -r requirements.txt
```

6. Menyiapkan Cloud SQL untuk PostgreSQL

Dalam tugas ini, Anda akan menyediakan instance Cloud SQL untuk PostgreSQL, membuat database, dan menyiapkannya untuk penelusuran vektor.

Tentukan Konfigurasi Cloud SQL

Buat file .env untuk menyimpan konfigurasi Anda. Jalankan kode berikut di terminal:
```
cloudshell edit .env
```

Tempel konfigurasi berikut ke .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}"

Ganti [YOUR_PROJECT_ID] dengan ID Project Google Cloud Anda yang sebenarnya. (misalnya, PROJECT_ID = "google-cloud-labs")
Jika Anda tidak dapat mengingat project ID, jalankan perintah berikut di terminal. Daftar semua project dan ID-nya akan ditampilkan.
```
gcloud projects list
```
Muat variabel ke dalam sesi shell Anda:
```
source .env
```

Membuat Instance dan Database

Buat instance Cloud SQL for PostgreSQL. Perintah ini membuat instance kecil yang cocok untuk lab ini.
```
gcloud sql instances create ${SQL_INSTANCE_NAME} \
  --database-version=POSTGRES_15 \
  --tier=db-g1-small \
  --region=${REGION} \
  --project=${PROJECT_ID}
```
Catatan: Pembuatan instance Cloud SQL baru dapat memerlukan waktu sekitar 5-7 menit. Harap bersabar.

Setelah instance siap, buat database:

gcloud sql databases create ${SQL_DATABASE_NAME} \
  --instance=${SQL_INSTANCE_NAME} \
  --project=${PROJECT_ID}

Buat pengguna database:

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

Mengaktifkan ekstensi pgvector

Ekstensi pgvector memungkinkan PostgreSQL menyimpan dan menelusuri embedding vektor. Anda harus mengaktifkannya secara eksplisit di database Anda.

Buat skrip bernama enable_pgvector.py. Jalankan kode berikut di terminal:
```
cloudshell edit enable_pgvector.py
```

Tempelkan kode berikut ke enable_pgvector.py. Skrip ini terhubung ke database Anda dan menjalankan 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()

Jalankan skrip:
```
python enable_pgvector.py
```

7. Bagian 1: Strategi Chunking

Langkah pertama dalam pipeline RAG adalah mengubah dokumen menjadi format yang dapat dipahami LLM: potongan.

LLM memiliki batas jendela konteks (jumlah teks yang dapat diproses sekaligus). Selain itu, mengambil dokumen 50 halaman untuk menjawab pertanyaan tertentu akan mengurangi kualitas informasi. Kami membagi dokumen menjadi "potongan" yang lebih kecil untuk mengisolasi informasi yang relevan.

Namun, cara Anda memisahkan teks sangatlah penting:

Pemecah Karakter: Memecah secara ketat berdasarkan jumlah karakter. Cara ini cepat, tetapi berisiko; dapat memotong kata atau kalimat menjadi dua, sehingga merusak makna semantik.
Recursive Splitter: Berupaya memisahkan berdasarkan paragraf terlebih dahulu, lalu kalimat, lalu kata. Fitur ini mencoba menyatukan unit semantik.
Pemecah Token: Memisahkan berdasarkan kosakata (token) LLM itu sendiri. Hal ini memastikan potongan cocok dengan sempurna ke dalam jendela konteks, tetapi dapat lebih mahal secara komputasi untuk dihasilkan.

Di bagian ini, Anda akan menyerap data yang sama menggunakan ketiga strategi untuk membandingkannya.

Membuat Skrip Penyerapan

Anda akan menggunakan skrip yang mendownload set data Harry Potter, membaginya menggunakan strategi Karakter, Rekursif, dan Token, serta mengupload sematan ke tiga tabel terpisah di Cloud SQL.

Buat file ingest_data.py:
```
cloudshell edit ingest_data.py
```

Tempelkan kode fixed berikut ke dalam ingest_data.py. Versi ini mengurai struktur JSON set data dengan benar.

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()

Jalankan skrip penyerapan. Tindakan ini akan mengisi database Anda dengan tiga tabel (koleksi) yang berbeda.
```
python ingest_data.py
```

Membandingkan Hasil Pemecahan

Setelah data dimuat, mari jalankan kueri terhadap ketiga koleksi untuk melihat pengaruh strategi chunking terhadap hasil.

Buat query_chunking.py:
```
cloudshell edit query_chunking.py
```

Tempelkan kode berikut ke 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()

Jalankan skrip kueri:
```
python query_chunking.py
```

Amati outputnya.

Perhatikan bagaimana pemisahan Karakter dapat memotong kalimat di tengah-tengah, sementara Rekursif mencoba mematuhi batas paragraf. Pemisahan token memastikan potongan cocok dengan sempurna ke jendela konteks LLM, tetapi mungkin mengabaikan struktur semantik.

8. Bagian 2: Peringkatan ulang

Penelusuran vektor (pengambilan) sangat cepat karena mengandalkan representasi matematika (embedding) yang dikompresi. Metode ini menjaring luas untuk memastikan Recall (menemukan semua item yang berpotensi relevan), tetapi sering kali menderita Presisi rendah (peringkat item tersebut mungkin tidak sempurna).

Sering kali, dokumen yang relevan "Hilang di Tengah" daftar hasil. LLM yang hanya memperhatikan 5 hasil teratas mungkin melewatkan jawaban penting yang berada di posisi #7.

Peringkatan ulang mengatasi hal ini dengan menambahkan tahap kedua.

Pengambil: Mengambil kumpulan yang lebih besar (misalnya, 25 teratas) menggunakan penelusuran vektor cepat.
Pengubah peringkat: Menggunakan model khusus (seperti Cross-Encoder) untuk memeriksa teks lengkap dari pasangan kueri dan dokumen. Cara ini lebih lambat, tetapi jauh lebih akurat. Fitur ini memberi skor ulang 25 teratas dan menampilkan 3 yang terbaik.

Dalam tugas ini, Anda akan menelusuri koleksi recursive yang dibuat di Bagian 1, tetapi kali ini Anda akan menerapkan Vertex AI Reranker untuk menyaring hasil.

Buat query_reranking.py:
```
cloudshell edit query_reranking.py
```

Tempelkan kode berikut. Perhatikan cara penargetan koleksi _recursive secara eksplisit dan penggunaan 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()

Jalankan kueri peringkat ulang:
```
python query_reranking.py
```

Mengamati

Anda mungkin melihat skor relevansi yang lebih tinggi atau pengurutan yang berbeda dibandingkan dengan penelusuran vektor mentah. Hal ini memastikan LLM menerima konteks yang paling tepat.

9. Bagian 3: Transformasi Kueri

Sering kali, hambatan terbesar dalam RAG adalah pengguna. Kueri pengguna sering kali ambigu, tidak lengkap, atau tidak tepat. Jika sematan kueri tidak selaras secara matematis dengan sematan dokumen, pengambilan akan gagal.

Transformasi Kueri menggunakan LLM untuk menulis ulang atau memperluas kueri sebelum kueri tersebut masuk ke database. Anda akan menerapkan dua teknik:

HyDE (Hypothetical Document Embeddings): Kemiripan vektor antara pertanyaan dan jawaban sering kali lebih rendah daripada kemiripan antara jawaban dan jawaban hipotetis. HyDE meminta LLM untuk berhalusinasi jawaban yang sempurna, menyematkannya, dan menelusuri dokumen yang terlihat seperti halusinasi tersebut.
Perintah Mundur: Jika pengguna mengajukan pertanyaan mendetail yang spesifik, sistem mungkin melewatkan konteks yang lebih luas. Perintah mundur meminta LLM untuk membuat pertanyaan abstrak tingkat yang lebih tinggi ("Apa sejarah keluarga ini?") untuk mengambil informasi dasar bersama dengan detail spesifik.

Buat query_transformation.py:

cloudshell edit query_transformation.py

Tempelkan kode berikut:

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()

Jalankan skrip transformasi:
```
python query_transformation.py
```

Amati outputnya.

Perhatikan bagaimana kueri Step-back dapat mengambil konteks yang lebih luas tentang sejarah keluarga Dursley, sementara HyDE berfokus pada detail spesifik yang dihasilkan dalam jawaban hipotetis.

10. Bagian 4: Pembuatan End-to-End

Kami telah membagi data, menyempurnakan penelusuran, dan memoles kueri pengguna. Sekarang, kita akhirnya memasukkan "G" dalam RAG: Generation (Generasi).

Sampai saat ini, kita hanya menemukan informasi. Untuk membangun asisten AI yang sebenarnya, kita perlu memasukkan dokumen berkualitas tinggi yang telah diurutkan ulang tersebut ke dalam LLM (Gemini) untuk menyintesis jawaban dalam bahasa alami.

Dalam pipeline produksi, hal ini melibatkan alur tertentu:

Mengambil: Mendapatkan kumpulan kandidat yang luas (misalnya, 10 Teratas) menggunakan penelusuran vektor cepat.
Peringkat ulang: Memfilter hingga yang terbaik (misalnya, 3 Teratas) menggunakan Vertex AI Reranker.
Konstruksi Konteks: Gabungkan konten 3 dokumen teratas tersebut menjadi satu string.
Grounded Prompting (Perintah Berbasis): Sisipkan string konteks tersebut ke dalam template perintah ketat yang memaksa LLM untuk hanya menggunakan informasi tersebut.

Membuat Skrip Pembuatan

Kita akan menggunakan gemini-2.5-flash untuk langkah pembuatan. Model ini ideal untuk RAG karena memiliki jendela konteks yang panjang dan latensi yang rendah, sehingga dapat memproses beberapa dokumen yang diambil dengan cepat.

Buat end_to_end_rag.py:

cloudshell edit end_to_end_rag.py

Tempelkan kode berikut. Perhatikan variabel template—di sinilah kita secara ketat menginstruksikan model untuk menghindari "halusinasi" (mengarang-ngarang) dengan mengikatnya ke konteks yang diberikan.

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()

Jalankan aplikasi akhir:

python end_to_end_rag.py

Memahami Output

Saat Anda menjalankan skrip ini, amati perbedaan antara potongan yang diambil secara mentah (yang Anda lihat di langkah-langkah sebelumnya) dan jawaban akhir. LLM bertindak sebagai alat sintesis—LLM membaca "potongan" teks yang terfragmentasi yang disediakan oleh Reranker dan menyatukannya menjadi kalimat yang koheren dan dapat dibaca manusia.

Dengan merangkai komponen ini, Anda beralih dari "tebakan" stokastik ke alur kerja deterministik yang mendasar. Pengambil menebar jala, Pengurut Ulang memilih hasil tangkapan terbaik, dan Generator memasak makanan.

11. Kesimpulan

Selamat! Anda telah berhasil membuat pipeline RAG tingkat lanjut yang jauh melampaui penelusuran vektor dasar.

Rekap

Anda telah mengonfigurasi Cloud SQL dengan pgvector untuk penyimpanan vektor yang dapat diskalakan.
Anda membandingkan Strategi Chunking untuk memahami pengaruh penyiapan data terhadap pengambilan.
Anda menerapkan Peringkatan Ulang dengan Vertex AI untuk meningkatkan presisi hasil Anda.
Anda menggunakan Transformasi Kueri (HyDE, Step-back) untuk menyelaraskan niat pengguna dengan data Anda.

Pelajari Lebih Lanjut

Arsitektur Referensi RAG: Jelajahi daftar panduan arsitektur referensi terkait RAG.

Dari Prototipe hingga Produksi

Lab ini merupakan bagian dari Alur Pembelajaran AI Siap Produksi dengan Google Cloud.

Jelajahi kurikulum lengkap untuk menjembatani kesenjangan dari prototipe hingga produksi.
Bagikan progres Anda dengan hashtag #ProductionReadyAI.

Teknik RAG Lanjutan

1. Pengantar

Ringkasan

Yang akan Anda lakukan

Yang akan Anda pelajari

2. Penyiapan project

Akun Google

Login ke Konsol Google Cloud

Aktifkan Penagihan

Tukarkan kredit Google Cloud senilai $5 (opsional)

Menyiapkan akun penagihan pribadi

Membuat project (opsional)

3. Buka Cloud Shell Editor

4. Mengaktifkan API

Memperkenalkan API

5. Buat lingkungan virtual & instal dependensi

6. Menyiapkan Cloud SQL untuk PostgreSQL

Tentukan Konfigurasi Cloud SQL

Membuat Instance dan Database

Mengaktifkan ekstensi pgvector

7. Bagian 1: Strategi Chunking

Membuat Skrip Penyerapan

Membandingkan Hasil Pemecahan

Amati outputnya.

8. Bagian 2: Peringkatan ulang

Mengamati

9. Bagian 3: Transformasi Kueri

Amati outputnya.

10. Bagian 4: Pembuatan End-to-End

Membuat Skrip Pembuatan

Memahami Output

11. Kesimpulan

Rekap

Pelajari Lebih Lanjut

Dari Prototipe hingga Produksi