۱. مقدمه
نمای کلی
بازیابی نسل افزوده (RAG) با پایهگذاری پاسخهای مدل زبان بزرگ (LLM) بر اساس دانش خارجی، آنها را بهبود میبخشد. با این حال، ساخت یک سیستم RAG آماده برای تولید به چیزی بیش از یک جستجوی برداری ساده نیاز دارد. شما باید نحوه دریافت دادهها، رتبهبندی نتایج مرتبط و نحوه پردازش پرسوجوهای کاربر را بهینه کنید.
در این آزمایشگاه جامع، شما یک برنامه RAG قوی با استفاده از Cloud SQL برای PostgreSQL (که با pgvector توسعه داده شده است) و Vertex AI خواهید ساخت. شما با سه تکنیک پیشرفته پیشرفت خواهید کرد:
- استراتژیهای قطعهبندی: شما مشاهده خواهید کرد که چگونه روشهای مختلف تقسیم متن (کاراکتر، بازگشتی، توکن) بر کیفیت بازیابی تأثیر میگذارند.
- رتبهبندی مجدد: شما Vertex AI Reranker را برای اصلاح نتایج جستجو و رفع مشکل «گم شدن در میانه» پیادهسازی خواهید کرد.
- تبدیل پرسوجو: شما از Gemini برای بهینهسازی پرسوجوهای کاربر از طریق تکنیکهایی مانند HyDE (جاسازیهای فرضی سند) و Step-back Prompting استفاده خواهید کرد.
کاری که انجام خواهید داد
- با استفاده از
pgvectorیک نمونه Cloud SQL برای PostgreSQL راهاندازی کنید. - یک خط لوله دریافت داده بسازید که متن را با استفاده از چندین استراتژی تکهتکه کرده و جاسازیها را در Cloud SQL ذخیره کند.
- جستجوهای معنایی انجام دهید و کیفیت نتایج حاصل از روشهای مختلف قطعهبندی را مقایسه کنید.
- یک Reranker را برای مرتبسازی مجدد اسناد بازیابی شده بر اساس میزان مرتبط بودن، ادغام کنید.
- پیادهسازی تبدیلهای پرسوجو مبتنی بر LLM برای بهبود بازیابی سوالات مبهم یا پیچیده.
آنچه یاد خواهید گرفت
- نحوه استفاده از LangChain با Vertex AI و Cloud SQL .
- تأثیر جداکنندههای متن کاراکتری ، بازگشتی و توکنی .
- نحوه پیادهسازی جستجوی برداری در PostgreSQL
- نحوه استفاده از ContextualCompressionRetriever برای رتبهبندی مجدد
- نحوه پیادهسازی HyDE و راهنمای گام به گام .
۲. راهاندازی پروژه
حساب گوگل
اگر از قبل حساب گوگل شخصی ندارید، باید یک حساب گوگل ایجاد کنید .
به جای حساب کاری یا تحصیلی از حساب شخصی استفاده کنید .
ورود به کنسول ابری گوگل
با استفاده از یک حساب کاربری شخصی گوگل، وارد کنسول ابری گوگل شوید.
فعال کردن صورتحساب
استفاده از اعتبار ۵ دلاری گوگل کلود (اختیاری)
برای اجرای این کارگاه، به یک حساب صورتحساب با مقداری اعتبار نیاز دارید. اگر قصد دارید از صورتحساب خودتان استفاده کنید، میتوانید از این مرحله صرف نظر کنید.
- روی این لینک کلیک کنید و با یک حساب گوگل شخصی وارد شوید. چیزی شبیه به این خواهید دید:
- روی دکمه «برای دسترسی به اعتبارات خود اینجا کلیک کنید» کلیک کنید. این شما را به صفحهای میبرد که میتوانید نمایه صورتحساب خود را تنظیم کنید.
- روی تأیید کلیک کنید. اکنون به حساب پرداخت آزمایشی پلتفرم گوگل کلود متصل شدهاید.
یک حساب پرداخت شخصی تنظیم کنید
اگر صورتحساب را با استفاده از اعتبارهای Google Cloud تنظیم کردهاید، میتوانید از این مرحله صرف نظر کنید.
برای تنظیم یک حساب پرداخت شخصی، به اینجا بروید تا پرداخت را در کنسول ابری فعال کنید .
برخی نکات:
- تکمیل این آزمایشگاه باید کمتر از ۱ دلار آمریکا از طریق منابع ابری هزینه داشته باشد.
- شما میتوانید مراحل انتهای این آزمایش را برای حذف منابع دنبال کنید تا از هزینههای بیشتر جلوگیری شود.
- کاربران جدید واجد شرایط استفاده از دوره آزمایشی رایگان ۳۰۰ دلاری هستند.
ایجاد پروژه (اختیاری)
اگر پروژه فعلی ندارید که بخواهید برای این آزمایشگاه استفاده کنید، اینجا یک پروژه جدید ایجاد کنید .
۳. ویرایشگر Cloud Shell را باز کنید
- برای دسترسی مستقیم به ویرایشگر Cloud Shell ، روی این لینک کلیک کنید.
- اگر امروز در هر مرحلهای از شما خواسته شد که مجوز دهید، برای ادامه روی تأیید کلیک کنید.
- اگر ترمینال در پایین صفحه نمایش داده نشد، آن را باز کنید:
- روی مشاهده کلیک کنید
- روی ترمینال کلیک کنید
- در ترمینال، پروژه خود را با این دستور تنظیم کنید:
gcloud config set project [PROJECT_ID]- مثال:
gcloud config set project lab-project-id-example - اگر نمیتوانید شناسه پروژه خود را به خاطر بیاورید، میتوانید تمام شناسههای پروژه خود را با استفاده از دستور زیر فهرست کنید:
gcloud projects list
- مثال:
- شما باید این پیام را ببینید:
Updated property [core/project].
۴. فعال کردن APIها
برای ساخت این راهکار، باید چندین API گوگل کلود را برای Vertex AI، Cloud SQL و سرویس Reranking فعال کنید.
- در ترمینال، APIها را فعال کنید:
gcloud services enable \ aiplatform.googleapis.com \ sqladmin.googleapis.com \ cloudresourcemanager.googleapis.com \ serviceusage.googleapis.com \ discoveryengine.googleapis.com
معرفی API ها
- رابط برنامهنویسی کاربردی هوش مصنوعی ورتکس (
aiplatform.googleapis.com): استفاده از Gemini برای تولید و Vertex AI Embeddings را برای برداریسازی متن فعال میکند. - رابط برنامهنویسی کاربردی مدیریت SQL ابری (
sqladmin.googleapis.com): به شما امکان میدهد نمونههای SQL ابری را به صورت برنامهنویسی مدیریت کنید. - رابط برنامهنویسی کاربردی موتور اکتشاف (
discoveryengine.googleapis.com): قابلیتهای Vertex AI Reranker را تقویت میکند. - API استفاده از سرویس (
serviceusage.googleapis.com): برای بررسی و مدیریت سهمیههای سرویس مورد نیاز است.
۵. یک محیط مجازی ایجاد کنید و وابستگیها را نصب کنید
قبل از شروع هر پروژه پایتون، ایجاد یک محیط مجازی تمرین خوبی است. این کار وابستگیهای پروژه را ایزوله میکند و از تداخل با سایر پروژهها یا بستههای پایتون سراسری سیستم جلوگیری میکند.
- یک پوشه به نام
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
۶. تنظیم Cloud SQL برای PostgreSQL
در این کار، شما یک نمونه Cloud SQL برای PostgreSQL تهیه خواهید کرد، یک پایگاه داده ایجاد خواهید کرد و آن را برای جستجوی برداری آماده خواهید کرد.
تعریف پیکربندی 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]، شناسه واقعی پروژه گوگل کلود خود را وارد کنید. (مثلاًPROJECT_ID = "google-cloud-labs")
اگر شناسه پروژه خود را به خاطر نمیآورید، دستور زیر را در ترمینال خود اجرا کنید. این دستور لیستی از تمام پروژههای شما و شناسههای آنها را به شما نشان میدهد.gcloud projects list - متغیرها را در جلسه پوسته خود بارگذاری کنید:
source .env
ایجاد نمونه و پایگاه داده
- یک نمونه Cloud SQL برای PostgreSQL ایجاد کنید. این دستور یک نمونه کوچک مناسب برای این آزمایش ایجاد میکند.
gcloud sql instances create ${SQL_INSTANCE_NAME} \ --database-version=POSTGRES_15 \ --tier=db-g1-small \ --region=${REGION} \ --project=${PROJECT_ID} - پس از آماده شدن نمونه، پایگاه داده را ایجاد کنید:
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
۷. بخش ۱: استراتژیهای قطعهبندی
اولین قدم در هر خط لوله RAG، تبدیل اسناد به فرمتی است که LLM بتواند آن را درک کند: تکهها .
LLMها محدودیت پنجره زمینه (میزان متنی که میتوانند همزمان پردازش کنند) دارند. علاوه بر این، بازیابی یک سند ۵۰ صفحهای برای پاسخ به یک سوال خاص، اطلاعات را رقیق میکند. ما اسناد را به "قطعههای" کوچکتر تقسیم میکنیم تا اطلاعات مرتبط را جدا کنیم.
با این حال، نحوه تقسیم متن بسیار مهم است:
- تقسیمکنندهی کاراکتر: متن را صرفاً بر اساس تعداد کاراکترها تقسیم میکند. این روش سریع اما پرخطر است؛ میتواند کلمات یا جملات را به نصف تقسیم کند و معنای معنایی را از بین ببرد.
- تقسیمکننده بازگشتی: تلاش میکند ابتدا بر اساس پاراگراف، سپس جمله و در نهایت کلمه تقسیمبندی کند. این روش سعی میکند واحدهای معنایی را در کنار هم نگه دارد.
- تقسیمکننده توکن: تقسیمها را بر اساس واژگان خود LLM (توکنها) انجام میدهد. این تضمین میکند که تکهها کاملاً در پنجرههای زمینه قرار میگیرند، اما تولید آنها میتواند از نظر محاسباتی گرانتر باشد.
در این بخش، شما دادههای یکسانی را با استفاده از هر سه استراتژی برای مقایسه آنها دریافت خواهید کرد.
اسکریپت مصرف را ایجاد کنید
شما از اسکریپتی استفاده خواهید کرد که یک مجموعه داده هری پاتر را دانلود میکند، آن را با استفاده از استراتژیهای 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() - اسکریپت ingestion را اجرا کنید. این کار پایگاه داده شما را با سه جدول (مجموعه) مختلف پر میکند.
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
خروجی را مشاهده کنید.
توجه کنید که چگونه تقسیمبندی کاراکتری ممکن است جملات را در میانه تفکر قطع کند، در حالی که بازگشتی سعی میکند مرزهای پاراگراف را رعایت کند. تقسیمبندی توکن تضمین میکند که قطعات به طور کامل در پنجرههای متن LLM قرار میگیرند، اما ممکن است ساختار معنایی را نادیده بگیرد.
۸. بخش ۲: رتبهبندی مجدد
جستجوی برداری (بازیابی) فوقالعاده سریع است زیرا به نمایشهای ریاضی فشرده (جاسازیها) متکی است. این جستجو شبکهی گستردهای را برای اطمینان از بازیابی (یافتن تمام موارد بالقوه مرتبط) ایجاد میکند، اما اغلب از دقت پایین رنج میبرد (رتبهبندی آن موارد ممکن است ناقص باشد).
اغلب، اسناد مرتبط در میانهی فهرست نتایج گم میشوند. یک LLM که به ۵ نتیجهی برتر توجه میکند، ممکن است پاسخ حیاتی را که در جایگاه شماره ۷ قرار دارد، از دست بدهد.
رتبهبندی مجدد با اضافه کردن یک مرحله دوم، این مشکل را حل میکند.
- بازیابیکننده: با استفاده از جستجوی برداری سریع، مجموعهای بزرگتر (مثلاً ۲۵ مورد برتر) را واکشی میکند.
- Reranker: از یک مدل تخصصی (مانند Cross-Encoder) برای بررسی متن کامل پرسوجو و جفت اسناد استفاده میکند. این روش کندتر اما بسیار دقیقتر است. این روش ۲۵ مورد برتر را دوباره امتیازدهی میکند و ۳ مورد برتر مطلق را برمیگرداند.
در این کار، شما مجموعه 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
مشاهده کردن
ممکن است در مقایسه با جستجوی بردار خام، متوجه امتیازهای مرتبط بالاتر یا ترتیب متفاوتی شوید. این تضمین میکند که LLM دقیقترین زمینه ممکن را دریافت میکند.
۹. بخش ۳: تبدیل پرسوجو
اغلب، بزرگترین گلوگاه در RAG، کاربر است. پرسوجوهای کاربر اغلب مبهم، ناقص یا با عبارات ضعیف هستند. اگر جاسازی پرسوجو از نظر ریاضی با جاسازی سند همسو نباشد، بازیابی با شکست مواجه میشود.
تبدیل پرسوجو از یک LLM برای بازنویسی یا گسترش پرسوجو قبل از رسیدن به پایگاه داده استفاده میکند. شما دو تکنیک را پیادهسازی خواهید کرد:
- HyDE (جاسازیهای فرضی سند): شباهت برداری بین یک سوال و یک پاسخ اغلب کمتر از شباهت بین یک پاسخ و یک پاسخ فرضی است. HyDE از LLM میخواهد که یک پاسخ کامل را تصور کند، آن را جاسازی میکند و اسنادی را که شبیه به آن توهم هستند، جستجو میکند.
- پرسش گام به گام: اگر کاربر یک سوال جزئی و خاص بپرسد، سیستم ممکن است زمینه وسیعتر را از دست بدهد. پرسش گام به گام از LLM میخواهد که یک سوال انتزاعی سطح بالاتر ("تاریخچه این خانواده چیست؟") ایجاد کند تا اطلاعات بنیادی را در کنار جزئیات خاص بازیابی کند.
-
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
خروجی را مشاهده کنید.
توجه کنید که چگونه پرسوجوی Step-back ممکن است زمینه وسیعتری را در مورد تاریخچه خانواده دورسلی بازیابی کند، در حالی که HyDE بر جزئیات خاص تولید شده در پاسخ فرضی تمرکز میکند.
۱۰. بخش ۴: تولید سرتاسری
ما دادههایمان را خرد کردهایم، جستجویمان را اصلاح کردهایم و عبارت جستجوی کاربر را اصلاح کردهایم. حالا، بالاخره "G" را در RAG: Generation قرار دادهایم.
تا این لحظه، ما فقط اطلاعات را پیدا میکردیم . برای ساخت یک دستیار هوش مصنوعی واقعی، باید آن اسناد با کیفیت بالا و رتبهبندی مجدد شده را به یک LLM (Gemini) بدهیم تا یک پاسخ به زبان طبیعی را ترکیب کند.
در یک خط تولید، این شامل یک جریان خاص است:
- بازیابی: با استفاده از جستجوی برداری سریع، مجموعهای گسترده از کاندیداها (مثلاً 10 مورد برتر) را دریافت کنید.
- رتبهبندی مجدد: با استفاده از Vertex AI Reranker، بهترینها (مثلاً ۳ مورد برتر) را فیلتر کنید.
- ساخت زمینه: محتوای آن سه سند برتر را در یک رشته واحد به هم وصل کنید.
- اعلان مبتنی بر زمینه: آن رشته زمینه را در یک الگوی اعلان دقیق قرار دهید که LLM را مجبور میکند فقط از آن اطلاعات استفاده کند.
ایجاد اسکریپت تولید
ما برای مرحله تولید 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
درک خروجی
وقتی این اسکریپت را اجرا میکنید، تفاوت بین تکههای خام بازیابی شده (که در مراحل قبل دیدید) و پاسخ نهایی را مشاهده کنید. LLM به عنوان یک ترکیبکننده عمل میکند - «تکههای» تکهتکه شده متن ارائه شده توسط Reranker را میخواند و آنها را به یک جمله منسجم و قابل خواندن توسط انسان تبدیل میکند.
با زنجیرهسازی این اجزا، شما از یک «حدس» تصادفی به یک گردش کار قطعی و مبتنی بر پایه حرکت میکنید. رتریور تور را میاندازد، رِینکر بهترین صید را انتخاب میکند و مولد غذا را میپزد.
۱۱. نتیجهگیری
تبریک! شما با موفقیت یک خط لوله RAG پیشرفته ساختید که بسیار فراتر از جستجوی برداری ساده است.
خلاصه
- شما Cloud SQL را با pgvector برای ذخیرهسازی برداری مقیاسپذیر پیکربندی کردید.
- شما استراتژیهای قطعهبندی را مقایسه کردید تا بفهمید آمادهسازی دادهها چگونه بر بازیابی تأثیر میگذارد.
- شما Reranking را با Vertex AI پیادهسازی کردید تا دقت نتایج خود را بهبود بخشید.
- شما از تبدیلات پرسوجو (HyDE، Step-back) برای همسو کردن قصد کاربر با دادههایتان استفاده کردید.
اطلاعات بیشتر
- معماریهای مرجع RAG : فهرستی از راهنماهای معماری مرجع مرتبط با RAG را بررسی کنید.
از نمونه اولیه تا تولید
این آزمایشگاه بخشی از پروژه «هوش مصنوعی آماده تولید با مسیر یادگیری ابری گوگل» است.
- برای پر کردن شکاف بین نمونه اولیه و تولید، برنامه درسی کامل را بررسی کنید .
- پیشرفت خود را با هشتگ #ProductionReadyAI به اشتراک بگذارید .