Esta página foi traduzida pela API Cloud Translation.

Técnicas avançadas de RAG

1. Introdução

Visão geral

A Geração Aumentada de Recuperação (RAG) melhora as respostas dos modelos de linguagem grandes (LLMs) ao embasá-las em conhecimento externo. No entanto, criar um sistema RAG pronto para produção exige mais do que uma simples pesquisa vetorial. É necessário otimizar a forma como os dados são ingeridos, como os resultados relevantes são classificados e como as consultas dos usuários são processadas.

Neste laboratório abrangente, você vai criar um aplicativo RAG robusto usando o Cloud SQL para PostgreSQL (estendido com pgvector) e a Vertex AI. Você vai aprender três técnicas avançadas:

Estratégias de divisão:você vai observar como diferentes métodos de divisão de texto (caractere, recursivo, token) afetam a qualidade da recuperação.
Reclassificação:você vai implementar o Vertex AI Reranker para refinar os resultados da pesquisa e resolver o problema de "perda no meio".
Transformação de consultas:você vai usar o Gemini para otimizar as consultas dos usuários com técnicas como HyDE (incorporações de documentos hipotéticos) e comando de recuo.

Atividades deste laboratório

Configure uma instância do Cloud SQL para PostgreSQL com pgvector.
Crie um pipeline de ingestão de dados que divide o texto usando várias estratégias e armazena incorporações no Cloud SQL.
Faça pesquisas semânticas e compare a qualidade dos resultados de diferentes métodos de divisão.
Integre um reranker para reordenar os documentos recuperados com base na relevância.
Implemente transformações de consulta com tecnologia LLM para melhorar a recuperação de perguntas ambíguas ou complexas.

O que você vai aprender

Como usar o LangChain com a Vertex AI e o Cloud SQL.
O impacto dos divisores de texto Character, Recursive e Token.
Como implementar a Pesquisa vetorial no PostgreSQL.
Como usar o ContextualCompressionRetriever para reclassificação.
Como implementar o HyDE e o Step-back Prompting.

2. Configurar o projeto

Conta do Google

Se você ainda não tiver uma Conta do Google pessoal, crie uma.

Use uma conta pessoal em vez de uma conta escolar ou de trabalho.

Faça login no console do Google Cloud usando uma Conta do Google pessoal.

Ativar faturamento

Resgatar US $5 em créditos do Google Cloud (opcional)

Para fazer este workshop, você precisa de uma conta de faturamento com algum crédito. Se você planeja usar seu próprio faturamento, pule esta etapa.

Clique neste link e faça login com uma Conta do Google pessoal. Você vai encontrar algo assim:
Clique no botão CLIQUE AQUI PARA ACESSAR SEUS CRÉDITOS. Isso vai abrir uma página para configurar seu perfil de faturamento
Clique em Confirmar. Agora você está conectado a uma conta de faturamento de avaliação do Google Cloud Platform.

Configurar uma conta de faturamento pessoal

Se você configurou o faturamento usando créditos do Google Cloud, pule esta etapa.

Para configurar uma conta de faturamento pessoal, acesse este link para ativar o faturamento no console do Cloud.

Algumas observações:

A conclusão deste laboratório custa menos de US $1 em recursos do Cloud.
Siga as etapas no final deste laboratório para excluir recursos e evitar mais cobranças.
Novos usuários podem aproveitar a avaliação sem custos financeiros de US$300.

Criar um projeto (opcional)

Se você não tiver um projeto atual que gostaria de usar neste laboratório, crie um novo aqui.

3. Abrir editor do Cloud Shell

Clique neste link para navegar diretamente até o editor do Cloud Shell.
Se for preciso autorizar em algum momento hoje, clique em Autorizar para continuar.
Se o terminal não aparecer na parte de baixo da tela, abra-o:
- Clique em Visualizar.
- Clique em Terminal.

No terminal, defina o projeto com este comando:

gcloud config set project [PROJECT_ID]

Exemplo:

gcloud config set project lab-project-id-example

Se você não se lembrar do ID do projeto, liste todos os IDs com:
```
gcloud projects list
```

Você vai receber esta mensagem:
```
Updated property [core/project].
```
Se você vir um WARNING e for perguntado Do you want to continue (Y/n)?, provavelmente inseriu o ID do projeto incorretamente. Pressione n, Enter e tente executar o comando gcloud config set project novamente.

4. Ativar APIs

Para criar essa solução, é necessário ativar várias APIs do Google Cloud para a Vertex AI, o Cloud SQL e o serviço de reclassificação.

No terminal, ative as APIs:

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

Apresentação das APIs

API Vertex AI (aiplatform.googleapis.com): permite o uso do Gemini para geração e do Vertex AI Embeddings para vetorizar texto.
API Cloud SQL Admin (sqladmin.googleapis.com): permite gerenciar instâncias do Cloud SQL de maneira programática.
API Discovery Engine (discoveryengine.googleapis.com): oferece recursos de reclassificação da Vertex AI.
API Service Usage (serviceusage.googleapis.com): necessária para verificar e gerenciar cotas de serviço.

5. Criar um ambiente virtual e instalar dependências

Antes de iniciar qualquer projeto em Python, é recomendável criar um ambiente virtual. Isso isola as dependências do projeto, evitando conflitos com outros projetos ou com os pacotes globais do Python no sistema.

Crie uma pasta chamada rag-labs e mude para ela. Execute o seguinte código no terminal:
```
mkdir rag-labs && cd rag-labs
```

Crie e ative um ambiente virtual:

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

Crie um arquivo requirements.txt com as dependências necessárias. Execute o seguinte código no terminal:
```
cloudshell edit requirements.txt
```

Cole as seguintes dependências otimizadas em requirements.txt. Essas versões são fixadas para evitar conflitos e acelerar a instalação.

# 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

Instale as dependências:
```
uv pip install -r requirements.txt
```

6. Configurar o Cloud SQL para PostgreSQL

Nesta tarefa, você vai provisionar uma instância do Cloud SQL para PostgreSQL, criar um banco de dados e prepará-lo para a pesquisa vetorial.

Definir a configuração do Cloud SQL

Crie um arquivo .env para armazenar sua configuração. Execute o seguinte código no terminal:
```
cloudshell edit .env
```

Cole a seguinte configuração em .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}"

Substitua [YOUR_PROJECT_ID] pelo ID do projeto do Google Cloud. (por exemplo, PROJECT_ID = "google-cloud-labs")
Se você não se lembrar do ID do projeto, execute o seguinte comando no terminal. Uma lista de todos os seus projetos e IDs vai aparecer.
```
gcloud projects list
```
Carregue as variáveis na sua sessão do shell:
```
source .env
```

Criar a instância e o banco de dados

Crie uma instância do Cloud SQL para PostgreSQL. Esse comando cria uma pequena instância adequada para este laboratório.
```
gcloud sql instances create ${SQL_INSTANCE_NAME} \
  --database-version=POSTGRES_15 \
  --tier=db-g1-small \
  --region=${REGION} \
  --project=${PROJECT_ID}
```
Observação:a criação de uma nova instância do Cloud SQL pode levar aproximadamente 5 a 7 minutos. Seja paciente.

Quando a instância estiver pronta, crie o banco de dados:

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

Crie o usuário do banco de dados:

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

Ativar a extensão pgvector

A extensão pgvector permite que o PostgreSQL armazene e pesquise embeddings de vetores. É necessário ativar explicitamente no banco de dados.

Crie um script chamado enable_pgvector.py. Execute o seguinte código no terminal:
```
cloudshell edit enable_pgvector.py
```

Cole o código a seguir em enable_pgvector.py. Esse script se conecta ao seu banco de dados e executa 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()

Execute o script:
```
python enable_pgvector.py
```

7. Parte 1: estratégias de divisão

A primeira etapa de qualquer pipeline de RAG é transformar documentos em um formato que o LLM possa entender: pedaços.

Os LLMs têm um limite de janela de contexto (a quantidade de texto que podem processar de uma só vez). Além disso, recuperar um documento de 50 páginas para responder a uma pergunta específica dilui as informações. Dividimos os documentos em "pedaços" menores para isolar as informações relevantes.

No entanto, como você divide o texto é muito importante:

Divisor de caracteres:divide estritamente por contagem de caracteres. Isso é rápido, mas arriscado, porque pode cortar palavras ou frases pela metade, destruindo o significado semântico.
Divisor recursivo:tenta dividir primeiro por parágrafo, depois por frase e, por fim, por palavra. Ele tenta manter as unidades semânticas juntas.
Divisor de tokens:divide com base no vocabulário do próprio LLM (tokens). Isso garante que os blocos se encaixem perfeitamente nas janelas de contexto, mas pode ser mais caro do ponto de vista computacional para gerar.

Nesta seção, você vai ingerir os mesmos dados usando as três estratégias para compará-las.

Criar o script de ingestão

Você vai usar um script que baixa um conjunto de dados de Harry Potter, divide-o usando as estratégias Character, Recursive e Token e faz upload dos embeddings para três tabelas separadas no Cloud SQL.

Crie o arquivo ingest_data.py:
```
cloudshell edit ingest_data.py
```

Cole o seguinte código fixo em ingest_data.py. Essa versão analisa corretamente a estrutura JSON do conjunto de dados.

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

Execute o script de ingestão. Isso vai preencher seu banco de dados com três tabelas (coleções) diferentes.
```
python ingest_data.py
```

Comparar resultados de divisão em partes

Agora que os dados foram carregados, vamos executar uma consulta em todas as três coleções para ver como a estratégia de divisão afeta os resultados.

Crie query_chunking.py:
```
cloudshell edit query_chunking.py
```

Cole o seguinte código em 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()

Execute o script de consulta:
```
python query_chunking.py
```

Observe a saída.

Observe como a divisão por caracteres pode cortar frases no meio do pensamento, enquanto a recursiva tenta respeitar os limites dos parágrafos. A divisão por tokens garante que os blocos se encaixem perfeitamente nas janelas de contexto do LLM, mas pode ignorar a estrutura semântica.

8. Parte 2: reclassificação

A pesquisa vetorial (recuperação) é incrivelmente rápida porque depende de representações matemáticas compactadas (embeddings). Ele lança uma rede ampla para garantir o recall (encontrar todos os itens potencialmente relevantes), mas geralmente sofre de baixa precisão (o ranking desses itens pode ser imperfeito).

Muitas vezes, os documentos relevantes se perdem no meio da lista de resultados. Um LLM que presta atenção aos cinco principais resultados pode perder a resposta crucial na posição 7.

O reordenamento resolve isso adicionando uma segunda etapa.

Recuperador:busca um conjunto maior (por exemplo, os 25 principais) usando a pesquisa rápida de vetores.
Refinador:usa um modelo especializado (como um Cross-Encoder) para examinar o texto completo da consulta e os pares de documentos. Ele é mais lento, mas muito mais preciso. Ele reavalia os 25 melhores e retorna os três melhores absolutos.

Nesta tarefa, você vai pesquisar a coleção recursive criada na Parte 1, mas desta vez vai aplicar o Vertex AI Reranker para refinar os resultados.

Crie query_reranking.py:
```
cloudshell edit query_reranking.py
```

Cole o código a seguir. Observe como ele segmenta explicitamente a coleção _recursive e usa 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()

Execute a consulta de reclassificação:
```
python query_reranking.py
```

Observe

Você pode notar pontuações de relevância mais altas ou uma ordenação diferente em comparação com uma pesquisa de vetor bruta. Isso garante que o LLM receba o contexto mais preciso possível.

9. Parte 3: transformação de consulta

Muitas vezes, o maior gargalo na RAG é o usuário. As consultas dos usuários costumam ser ambíguas, incompletas ou mal formuladas. Se o encadeamento da consulta não se alinhar matematicamente com o encadeamento do documento, a recuperação vai falhar.

A transformação de consultas usa um LLM para reescrever ou expandir a consulta antes de ela chegar ao banco de dados. Você vai implementar duas técnicas:

HyDE (Hypothetical Document Embeddings): a similaridade vetorial entre uma pergunta e uma resposta geralmente é menor do que entre uma resposta e uma resposta hipotética. O HyDE pede ao LLM para alucinar uma resposta perfeita, incorpora essa resposta e pesquisa documentos que se parecem com a alucinação.
Comando de recuo:se um usuário fizer uma pergunta detalhada específica, o sistema poderá perder o contexto mais amplo. O comando de recuo pede ao LLM para gerar uma pergunta mais abstrata e de nível superior ("Qual é a história dessa família?") para recuperar informações básicas junto com os detalhes específicos.

Crie query_transformation.py:

cloudshell edit query_transformation.py

Cole o seguinte código:

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

Execute o script de transformação:
```
python query_transformation.py
```

Observe a saída.

Perceba como a consulta Step-back pode recuperar um contexto mais amplo sobre a história da família Dursley, enquanto a HyDE se concentra nos detalhes específicos gerados na resposta hipotética.

10. Parte 4: geração de ponta a ponta

Cortamos nossos dados, refinamos nossa pesquisa e aprimoramos a consulta do usuário. Agora, vamos colocar o "G" em RAG: Geração.

Até agora, apenas encontramos informações. Para criar um verdadeiro assistente de IA, precisamos inserir esses documentos de alta qualidade e reclassificados em um LLM (Gemini) para sintetizar uma resposta em linguagem natural.

Em um pipeline de produção, isso envolve um fluxo específico:

Recuperar:receba um amplo conjunto de candidatos (por exemplo, Top 10) usando a pesquisa vetorial rápida.
Reclassificar:filtre para encontrar o melhor resultado (por exemplo, Top 3) usando o Vertex AI Reranker.
Construção de contexto:combine o conteúdo dos três principais documentos em uma única string.
Comandos fundamentados:insira essa string de contexto em um modelo de comando estrito que force o LLM a usar apenas essas informações.

Criar o script de geração

Vamos usar gemini-2.5-flash para a etapa de geração. Esse modelo é ideal para RAG porque tem uma janela de contexto longa e baixa latência, o que permite processar vários documentos recuperados rapidamente.

Crie end_to_end_rag.py:

cloudshell edit end_to_end_rag.py

Cole o código a seguir. Preste atenção à variável template. É aqui que instruímos o modelo a evitar "alucinações" (inventar coisas) vinculando-o ao contexto fornecido.

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

Execute o aplicativo final:

python end_to_end_rag.py

Como entender a saída

Ao executar esse script, observe a diferença entre os trechos brutos recuperados (que você viu nas etapas anteriores) e a resposta final. O LLM age como um sintetizador: ele lê os "pedaços" fragmentados de texto fornecidos pelo Reranker e os transforma em uma frase coerente e legível.

Ao encadear esses componentes, você passa de um "palpite" estocástico para um fluxo de trabalho determinista e fundamentado. O Retriever lança a rede, o Reranker seleciona a melhor captura e o Generator prepara a refeição.

11. Conclusão

Parabéns! Você criou um pipeline de RAG avançado que vai muito além da pesquisa vetorial básica.

Recapitulação

Você configurou o Cloud SQL com pgvector para armazenamento de vetores escalonável.
Você comparou estratégias de divisão em partes para entender como a preparação de dados afeta a recuperação.
Você implementou o reclassificação com a Vertex AI para melhorar a precisão dos resultados.
Você usou transformações de consulta (HyDE, Step-back) para alinhar a intenção do usuário aos seus dados.

Saiba mais

Arquiteturas de referência de RAG: confira uma lista de guias de arquitetura de referência relacionados a RAG.

Do protótipo à produção

Este laboratório faz parte do programa de aprendizado "IA pronta para produção com o Google Cloud".

Confira o currículo completo para diminuir a distância entre o protótipo e a produção.
Compartilhe seu progresso com a hashtag #ProductionReadyAI.

Técnicas avançadas de RAG

1. Introdução

Visão geral

Atividades deste laboratório

O que você vai aprender

2. Configurar o projeto

Conta do Google

Fazer login no console do Google Cloud

Ativar faturamento

Resgatar US $5 em créditos do Google Cloud (opcional)

Configurar uma conta de faturamento pessoal

Criar um projeto (opcional)

3. Abrir editor do Cloud Shell

4. Ativar APIs

Apresentação das APIs

5. Criar um ambiente virtual e instalar dependências

6. Configurar o Cloud SQL para PostgreSQL

Definir a configuração do Cloud SQL

Criar a instância e o banco de dados

Ativar a extensão pgvector

7. Parte 1: estratégias de divisão

Criar o script de ingestão

Comparar resultados de divisão em partes

Observe a saída.

8. Parte 2: reclassificação

Observe

9. Parte 3: transformação de consulta

Observe a saída.

10. Parte 4: geração de ponta a ponta

Criar o script de geração

Como entender a saída

11. Conclusão

Recapitulação

Saiba mais

Do protótipo à produção