Эта страница переведена с помощью Cloud Translation API.

Передовые методы RAG

1. Введение

Обзор

Генерация с расширенным поиском (Retrieval Augmented Generation, RAG) улучшает ответы больших языковых моделей (LLM), основываясь на внешних знаниях. Однако для создания готовой к использованию системы RAG требуется нечто большее, чем просто векторный поиск. Необходимо оптимизировать процесс ввода данных, ранжирование релевантных результатов и обработку пользовательских запросов.

В этой всесторонней лабораторной работе вы создадите надежное RAG-приложение, используя Cloud SQL для PostgreSQL (расширенное с помощью pgvector ) и Vertex AI . Вы освоите три продвинутые методики:

Стратегии сегментации текста: Вы увидите, как различные методы разделения текста (по символам, рекурсивно, по токенам) влияют на качество поиска.
Переранжирование: Вам предстоит внедрить инструмент Vertex AI Reranker для уточнения результатов поиска и решения проблемы "потерянных результатов".
Преобразование запросов: Вы будете использовать Gemini для оптимизации пользовательских запросов с помощью таких методов, как HyDE (гипотетические векторные представления документов) и Step-back Prompting .

Что вы будете делать

Настройте экземпляр Cloud SQL для PostgreSQL с помощью pgvector .
Создайте конвейер для загрузки данных, который разбивает текст на фрагменты с использованием различных стратегий и сохраняет векторные представления в Cloud SQL.
Проведите семантический поиск и сравните качество результатов, полученных с помощью различных методов сегментации текста.
Интегрируйте инструмент переранжирования для изменения порядка найденных документов в зависимости от их релевантности.
Внедрите преобразования запросов на основе LLM для улучшения поиска информации по неоднозначным или сложным вопросам.

Что вы узнаете

Как использовать LangChain с Vertex AI и Cloud SQL .
Влияние разделителей текста по символам , рекурсивных разделителей и разделителей по токенам .
Как реализовать векторный поиск в PostgreSQL.
Как использовать ContextualCompressionRetriever для переранжирования.
Как реализовать HyDE и подсказки с шагом назад .

2. Настройка проекта

Аккаунт Google

Если у вас еще нет личного аккаунта Google, вам необходимо его создать .

Используйте личный аккаунт вместо рабочего или учебного.

Войдите в консоль Google Cloud, используя личную учетную запись Google.

Включить выставление счетов

Обменяйте 5 долларов США на кредиты Google Cloud (по желанию)

Для проведения этого мастер-класса вам потребуется платежный аккаунт с достаточным балансом. Если вы планируете использовать собственную платежную систему, этот шаг можно пропустить.

Перейдите по этой ссылке и войдите в систему, используя свой личный аккаунт Google. Вы увидите примерно следующее:
Нажмите кнопку «НАЖМИТЕ ЗДЕСЬ ДЛЯ ДОСТУПА К ВАШИМ КРЕДИТАМ ». Это переведет вас на страницу настройки вашего платежного профиля.
Нажмите «Подтвердить». Теперь вы подключены к пробному платёжному аккаунту Google Cloud Platform.

Создайте личный платежный аккаунт.

Если вы настроили оплату с использованием кредитов Google Cloud, этот шаг можно пропустить.

Чтобы настроить личный платежный аккаунт, перейдите сюда, чтобы включить оплату в облачной консоли.

Несколько замечаний:

Выполнение этой лабораторной работы должно обойтись менее чем в 1 доллар США в виде облачных ресурсов.
В конце этой лабораторной работы вы можете выполнить действия по удалению ресурсов, чтобы избежать дальнейших списаний средств.
Новые пользователи могут воспользоваться бесплатной пробной версией стоимостью 300 долларов США .

Создать проект (необязательно)

Если у вас нет текущего проекта, который вы хотели бы использовать для этой лабораторной работы, создайте новый проект здесь .

3. Откройте редактор 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].
```
Если вы видите WARNING и вас спрашивают Do you want to continue (Y/n)? , то, скорее всего, вы неправильно ввели идентификатор проекта. Нажмите n , затем Enter и попробуйте снова выполнить команду gcloud config set project .

4. Включите API.

Для создания этого решения необходимо активировать несколько API Google Cloud для 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.

API Vertex AI ( aiplatform.googleapis.com ): позволяет использовать Gemini для генерации и Vertex AI Embeddings для векторизации текста.
API администратора Cloud SQL ( sqladmin.googleapis.com ): позволяет программно управлять экземплярами Cloud SQL.
API Discovery Engine ( discoveryengine.googleapis.com ): Обеспечивает работу функции переранжирования Vertex AI.
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. Настройка Cloud SQL для PostgreSQL

В этом задании вам нужно будет развернуть экземпляр Cloud SQL for PostgreSQL, создать базу данных и подготовить её для векторного поиска.

Определение конфигурации Cloud SQL

Создайте файл .env для хранения вашей конфигурации. Выполните следующий код в терминале :
```
cloudshell edit .env
```

Вставьте следующую конфигурацию в файл .env .

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

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

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

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

Замените [YOUR_PROJECT_ID] на фактический идентификатор вашего проекта в Google Cloud. (например, PROJECT_ID = "google-cloud-labs" )
Если вы не помните идентификатор своего проекта, выполните следующую команду в терминале . Она покажет вам список всех ваших проектов и их идентификаторов.
```
gcloud projects list
```
Загрузите переменные в свою сессию командной оболочки:
```
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}
```
Примечание: Создание нового экземпляра Cloud SQL может занять приблизительно 5-7 минут . Пожалуйста, наберитесь терпения.

После того, как экземпляр будет готов, создайте базу данных:

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

Создайте пользователя базы данных:

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

Включите расширение pgvector

Расширение pgvector позволяет PostgreSQL хранить и искать векторные представления данных. Необходимо явно включить его в вашей базе данных.

Создайте скрипт с именем enable_pgvector.py . Запустите следующий код в терминале :
```
cloudshell edit enable_pgvector.py
```

Вставьте следующий код в файл enable_pgvector.py . Этот скрипт подключается к вашей базе данных и выполняет CREATE EXTENSION IF NOT EXISTS vector; .

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

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

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

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

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

if __name__ == "__main__":
    enable_pgvector()

Запустите скрипт:
```
python enable_pgvector.py
```

7. Часть 1: Стратегии сегментации информации

Первым шагом в любом конвейере RAG является преобразование документов в формат, понятный LLM: в виде фрагментов .

У программ обработки длинных текстов (LLM) есть ограничение на контекстное окно (объем текста, который они могут обрабатывать за один раз). Кроме того, поиск 50-страничного документа для ответа на конкретный вопрос приводит к размыванию информации. Мы разбиваем документы на более мелкие «фрагменты», чтобы выделить релевантную информацию.

Однако то, как вы разделите текст, имеет огромное значение:

Разделитель символов: Разделяет строго по количеству символов. Это быстро, но рискованно; может разделить слова или предложения пополам, разрушая смысловое содержание.
Рекурсивный разделитель: пытается разделить текст сначала по абзацам, затем по предложениям, а затем по словам. Он старается сохранить семантические единицы вместе.
Разделитель токенов: Разделяет текст на основе собственного словаря (токенов) 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()

Запустите скрипт загрузки данных. Это заполнит вашу базу данных тремя различными таблицами (коллекциями).
```
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, но может игнорировать семантическую структуру.

8. Часть 2: Переранжирование

Векторный поиск (извлечение) невероятно быстр, потому что он основан на сжатых математических представлениях (встраиваниях). Он охватывает широкий круг объектов, чтобы обеспечить полноту (нахождение всех потенциально релевантных элементов), но часто страдает от низкой точности (ранжирование этих элементов может быть несовершенным).

Часто важные документы "теряются в середине" списка результатов. Магистр права, обращающий внимание на первые 5 результатов, может упустить из виду важный ответ, находящийся на 7-й позиции.

Переранжирование решает эту проблему за счет добавления второго этапа.

Retriever: Извлекает больший набор данных (например, топ-25) с помощью быстрого векторного поиска.
Reranker: Использует специализированную модель (например, кросс-кодировщик) для анализа полного текста запроса и пар документов. Работает медленнее, но гораздо точнее. Переоценивает 25 лучших результатов и возвращает 3 абсолютных лучших.

В этом задании вам предстоит выполнить поиск в recursive коллекции, созданной в Части 1, но на этот раз вы примените алгоритм Vertex AI Reranker для уточнения результатов.

Создайте query_reranking.py :
```
cloudshell edit query_reranking.py
```

Вставьте следующий код. Обратите внимание, как он явно обращается к коллекции _recursive и использует ContextualCompressionRetriever .

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

if __name__ == "__main__":
    main()

Выполните запрос на переранжирование:
```
python query_reranking.py
```

Наблюдать

Вы можете заметить более высокие показатели релевантности или иной порядок результатов по сравнению с обычным векторным поиском. Это гарантирует, что LLM получит максимально точный контекст.

9. Часть 3: Преобразование запроса

Зачастую самым большим узким местом в 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 фокусируется на конкретных деталях, полученных в гипотетическом ответе.

10. Часть 4: Сквозное построение

Мы обработали данные, уточнили поиск и доработали пользовательский запрос. Теперь мы наконец-то добавили букву «G» в аббревиатуру RAG: Generation (Поколение) .

До сих пор мы лишь находили информацию. Для создания настоящего ИИ-помощника нам необходимо загружать эти высококачественные, переранжированные документы в LLM (Gemini), чтобы синтезировать ответ на естественном языке.

В производственной цепочке это подразумевает определенный поток:

Получение: Получение широкого набора кандидатов (например, топ-10) с помощью быстрого векторного поиска.
Переранжирование: Отфильтруйте результаты, оставив только лучшие (например, топ-3), используя инструмент переранжирования Vertex AI.
Создание контекста: Объедините содержимое этих трех документов в одну строку.
Обоснованная подсказка: Вставьте эту контекстную строку в строгий шаблон подсказки, который заставляет 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, и сглаживает их, превращая в связное, удобочитаемое предложение.

Соединяя эти компоненты, вы переходите от случайного «угадывания» к детерминированному, обоснованному рабочему процессу. « Извлекатель » забрасывает сеть, «Переранжировщик» выбирает лучший улов, а «Генератор» готовит блюдо.

11. Заключение

Поздравляем! Вы успешно создали продвинутый конвейер RAG, который выходит далеко за рамки базового векторного поиска.

Краткий обзор

Вы настроили Cloud SQL с pgvector для масштабируемого векторного хранилища.
Вы сравнили стратегии сегментации данных , чтобы понять, как подготовка данных влияет на их поиск.
Вы внедрили функцию переранжирования с помощью Vertex AI для повышения точности результатов.
Вы использовали преобразования запросов (HyDE, Step-back), чтобы согласовать намерения пользователя с вашими данными.

Узнать больше

Референтные архитектуры RAG : Ознакомьтесь со списком руководств по референтным архитектурам, связанных с RAG.

От прототипа к серийному производству

Данная лабораторная работа является частью учебного курса "Готовый к внедрению ИИ в производство с помощью Google Cloud" .

Изучите полный учебный план , чтобы преодолеть разрыв между прототипом и серийным производством.
Делитесь своими успехами, используя хэштег #ProductionReadyAI.

Передовые методы RAG

1. Введение

Обзор

Что вы будете делать

Что вы узнаете

2. Настройка проекта

Аккаунт Google

Войдите в консоль Google Cloud.

Включить выставление счетов

Обменяйте 5 долларов США на кредиты Google Cloud (по желанию)

Создайте личный платежный аккаунт.

Создать проект (необязательно)

3. Откройте редактор Cloud Shell.

4. Включите API.

Представляем API.

5. Создайте виртуальное окружение и установите зависимости.

6. Настройка Cloud SQL для PostgreSQL

Определение конфигурации Cloud SQL

Создайте экземпляр и базу данных.

Включите расширение pgvector

7. Часть 1: Стратегии сегментации информации

Создайте скрипт загрузки данных.

Сравните результаты сегментации.

Просмотрите результат.

8. Часть 2: Переранжирование

Наблюдать

9. Часть 3: Преобразование запроса

Просмотрите результат.

10. Часть 4: Сквозное построение

Создайте скрипт генерации

Понимание результата

11. Заключение

Краткий обзор

Узнать больше

От прототипа к серийному производству