Padrões de Implementação RAG

O que Padrões de Implementação RAG faz?

Guia abrangente para implementar sistemas de Geração Aumentada por Recuperação (RAG), incluindo seleção de banco de dados vetorial, estratégias de chunking, modelos de embedding, otimização de recuperação e padrões de implantação em produção.

---

Referência Rápida

Quando usar esta habilidade:

• Construindo sistemas RAG/busca semântica
• Implementando pipelines de recuperação de documentos
• Otimizando o desempenho de banco de dados vetorial
• Depurando problemas de qualidade de recuperação
• Escolhendo entre opções de banco de dados vetorial
• Projetando estratégias de chunking
• Implementando busca híbrida

Tecnologias abordadas:

• Bancos de Dados Vetoriais: Qdrant, Pinecone, Chroma, Weaviate, Milvus
• Embeddings: OpenAI, Sentence Transformers, Cohere
• Frameworks: LangChain, LlamaIndex, Haystack

---

Parte 1: Seleção de Banco de Dados Vetorial

Matriz de Comparação de Bancos de Dados

Banco de Dados	Melhor Para	Implantação	Desempenho	Custo
Qdrant	Auto-hospedado, produção	Docker/K8s	Excelente (Rust)	Grátis (auto-hospedado)
Pinecone	Gerenciado, prototipagem rápida	Nuvem	Excelente	Pagamento por uso
Chroma	Desenvolvimento local, embarcado	Em processo	Bom (Python)	Grátis
Weaviate	Esquemas complexos, GraphQL	Docker/Nuvem	Excelente (Go)	Grátis + Nuvem
Milvus	Grande escala, distribuído	K8s	Excelente (C++)	Grátis (auto-hospedado)

Configuração do Qdrant (Recomendado para Produção)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

# Inicializar cliente (local ou nuvem)
client = QdrantClient(url="http://localhost:6333")  # ou URL da nuvem

# Criar coleção
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,  # Dimensão do OpenAI text-embedding-3-small
        distance=Distance.COSINE  # ou DOT, EUCLID
    )
)

# Inserir vetores com payload
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],  # 1536 dimensões
            payload={
                "text": "Conteúdo do documento",
                "source": "doc.pdf",
                "page": 1,
                "metadata": {...}
            }
        )
    ]
)

# Busca
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=5,
    score_threshold=0.7  # Similaridade mínima
)

Configuração do Pinecone (Serviço Gerenciado)

from pinecone import Pinecone, ServerlessSpec

# Inicializar
pc = Pinecone(api_key="sua-chave")

# Criar índice
pc.create_index(
    name="documents",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

# Obter índice
index = pc.Index("documents")

# Inserir vetores
index.upsert(vectors=[
    ("doc1", [0.1, 0.2, ...], {"text": "...", "source": "..."})
])

# Consultar
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True
)

---

Parte 2: Estratégias de Chunking

Estratégia 1: Chunking de Tamanho Fixo (Simples, Rápido)

def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """
    Dividir texto em chunks de tamanho fixo com sobreposição.

    Prós: Simples, tamanhos de chunk previsíveis
    Contras: Pode quebrar no meio da frase, limites semânticos pobres
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

# Uso
chunks = fixed_size_chunking(document, chunk_size=512, overlap=50)

Quando usar:

• Documentos simples (logs, transcrições)
• Prototipagem/MVP
• Orçamentos de tokens consistentes necessários

Estratégia 2: Chunking Semântico (Melhor Qualidade)

from langchain.text_splitter import RecursiveCharacterTextSplitter

def semantic_chunking(text: str, chunk_size: int = 1000, overlap: int = 200) -> list[str]:
    """
    Dividir em limites semânticos (parágrafos, frases).

    Prós: Preserva o significado, melhor qualidade de recuperação
    Contras: Tamanhos de chunk variáveis, processamento mais lento
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],  # Ordem de prioridade
        length_function=len
    )

    return splitter.split_text(text)

# Uso
chunks = semantic_chunking(document, chunk_size=1000, overlap=200)

Quando usar:

• Documentos longos (artigos, livros, relatórios)
• Qualidade > velocidade
• Conteúdo em linguagem natural

Estratégia 3: Chunking Hierárquico (Melhor para Documentos Estruturados)

def hierarchical_chunking(document: dict) -> list[dict]:
    """
    Chunk baseado na estrutura do documento (seções, subseções).

    Prós: Preserva hierarquia, permite recuperação pai-filho
    Contras: Requer entrada estruturada, mais complexo
    """
    chunks = []

    for section in document['sections']:
        # Chunk pai (resumo da seção)
        chunks.append({
            'text': section['title'] + '\n' + section['summary'],
            'type': 'parent',
            'section_id': section['id']
        })

        # Chunks filhos (parágrafos)
        for para in section['paragraphs']:
            chunks.append({
                'text': para,
                'type': 'child',
                'parent_id': section['id']
            })

    return chunks

Quando usar:

• Documentação técnica
• Livros com sumário
• Documentos legais
• Necessidade de preservar hierarquia de contexto

Estratégia 4: Janela Deslizante (Preservação Máxima de Contexto)

def sliding_window_chunking(text: str, window_size: int = 512, stride: int = 256) -> list[str]:
    """
    Janelas sobrepostas para máximo contexto.

    Prós: Sem perda de informação nos limites
    Contras: Sobrecarga de armazenamento (conteúdo duplicado)
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words) - window_size + 1, stride):
        chunk = ' '.join(words[i:i + window_size])
        chunks.append(chunk)

    return chunks

Quando usar:

• Precisão crítica de recuperação necessária
• Consultas curtas precisam de contexto mais amplo
• Custo de armazenamento não é uma preocupação

---

Parte 3: Modelos de Embedding

Guia de Seleção de Modelo

Modelo	Dimensões	Velocidade	Qualidade	Custo	Caso de Uso
OpenAI text-embedding-3-small	1536	Rápido	Excelente	$0.02/1M tokens	Produção, uso geral
OpenAI text-embedding-3-large	3072	Médio	Melhor	$0.13/1M tokens	Recuperação de alta qualidade
all-MiniLM-L6-v2	384	Muito rápido	Bom	Grátis	Auto-hospedado, prototipagem
all-mpnet-base-v2	768	Rápido	Muito bom	Grátis	Auto-hospedado, qualidade
Cohere embed-english-v3.0	1024	Rápido	Excelente	$0.10/1M tokens	Foco em busca semântica

Embeddings OpenAI (Recomendado)

from openai import OpenAI

client = OpenAI(api_key="sua-chave")

def get_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """
    Gerar embeddings usando OpenAI.

    Tamanho do lote: Até 2048 entradas por requisição
    Limites de taxa: Verifique limites do plano
    """
    response = client.embeddings.create(
        model=model,
        input=texts
    )

    return [item.embedding for item in response.data]

# Uso
chunks = ["chunk 1", "chunk 2", ...]
embeddings = get_embeddings(chunks)

Sentence Transformers (Auto-Hospedado)

from sentence_transformers import SentenceTransformer

# Carregar modelo (armazenado em cache após o primeiro download)
model = SentenceTransformer('all-MiniLM-L6-v2')

def get_embeddings_local(texts: list[str]) -> list[list[float]]:
    """
    Gerar embeddings localmente (sem custos de API).

    GPU recomendada para lotes > 100
    CPU aceitável para lotes pequenos
    """
    return model.encode(texts, show_progress_bar=True).tolist()

# Uso
embeddings = get_embeddings_local(chunks)

---

Parte 4: Otimização de Recuperação

Técnica 1: Busca Híbrida (Denso + Esparso)

from qdrant_client.models import Filter, FieldCondition, MatchValue

def hybrid_search(query: str, query_vector: list[float], top_k: int = 10):
    """
    Combinar busca densa (vetor) e esparsa (palavra-chave).

    Denso: Similaridade semântica
    Esparso: Correspondências exatas de palavras-chave
    """
    # Busca densa
    dense_results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        limit=top_k * 2  # Obter mais candidatos
    )

    # Busca esparsa (BM25 via metadados)
    sparse_results = client.search(
        collection_name="documents",
        query_filter=Filter(
            must=[
                FieldCondition(
                    key="text",
                    match=MatchValue(value=query)
                )
            ]
        ),
        limit=top_k * 2
    )

    # Mesclar e reordenar
    combined = merge_results(dense_results, sparse_results, weights=(0.7, 0.3))
    return combined[:top_k]

Técnica 2: Expansão de Consulta

def expand_query(query: str) -> list[str]:
    """
    Gerar variações de consulta para melhor recall.

    Técnicas:
    - Expansão de sinônimos
    - Reformulação de perguntas
    - Extração de entidades
    """
    from openai import OpenAI
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "Gere 3 paráfrases alternativas da consulta do usuário."
        }, {
            "role": "user",
            "content": query
        }]
    )

    expanded = [query] + response.choices[0].message.content.split('\n')
    return expanded

# Uso
queries = expand_query("Como treinar redes neurais?")
# → ["Como treinar redes neurais?",
#    "Quais são as técnicas de treinamento de redes neurais?",
#    "Métodos de otimização de redes neurais",
#    "Treinamento de modelos de aprendizado profundo"]

Técnica 3: Reordenamento

from sentence_transformers import CrossEncoder

# Carregar cross-encoder (melhor que bi-encoder para reordenamento)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_results(query: str, results: list[dict], top_k: int = 5) -> list[dict]:
    """
    Reordenar resultados iniciais usando cross-encoder.

    Mais preciso, porém mais lento que a recuperação inicial
    Use apenas nos 20-50 candidatos principais
    """
    # Pontuar cada par consulta-documento
    pairs = [(query, result['text']) for result in results]
    scores = reranker.predict(pairs)

    # Combinar pontuações com resultados
    for result, score in zip(results, scores):
        result['rerank_score'] = float(score)

    # Ordenar e retornar top_k
    reranked = sorted(results, key=lambda x: x['rerank_score'], reverse=True)
    return reranked[:top_k]

Técnica 4: Filtragem por Metadados

def filtered_search(
    query_vector: list[float],
    filters: dict,
    top_k: int = 5
):
    """
    Filtrar busca por metadados (data, categoria, autor, etc.)

    Pré-filtro: Mais rápido, mas pode perder resultados
    Pós-filtro: Mais resultados, mas mais lento
    """
    from qdrant_client.models import Filter, FieldCondition, Range

    # Construir condições de filtro
    conditions = []

    if 'date_range' in filters:
        conditions.append(
            FieldCondition(
                key="date",
                range=Range(
                    gte=filters['date_range']['start'],
                    lte=filters['date_range']['end']
                )
            )
        )

    if 'category' in filters:
        conditions.append(
            FieldCondition(
                key="category",
                match=MatchValue(value=filters['category'])
            )
        )

    # Buscar com filtros
    results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        query_filter=Filter(must=conditions) if conditions else None,
        limit=top_k
    )

    return results

---

Parte 5: Gerenciamento de Contexto

Padrão 1: Otimização de Contexto Recuperado

def optimize_context(query: str, retrieved_docs: list[str], max_tokens: int = 4000) -> str:
    """
    Otimizar contexto recuperado para caber na janela de contexto do LLM.

    Estratégias:
    1. Truncamento baseado em relevância
    2. Sumarização extrativa
    3. Remoção de sobreposição
    """
    # Ordenar por relevância
    sorted_docs = sorted(retrieved_docs, key=lambda d: d['score'], reverse=True)

    # Construir contexto dentro do orçamento de tokens
    context_parts = []
    total_tokens = 0

    for doc in sorted_docs:
        doc_tokens = estimate_tokens(doc['text'])

        if total_tokens + doc_tokens <= max_tokens:
            context_parts.append(f"[Fonte: {doc['source']}]\n{doc['text']}")
            total_tokens += doc_tokens
        else:
            # Truncar último documento para caber
            remaining = max_tokens - total_tokens
            truncated = truncate_to_tokens(doc['text'], remaining)
            context_parts.append(f"[Fonte: {doc['source']}]\n{truncated}")
            break

    return "\n\n---\n\n".join(context_parts)

Padrão 2: Rastreamento de Citações

def generate_with_citations(query: str, context: str, sources: list[dict]) -> dict:
    """
    Gerar resposta com rastreamento de citações.

    Retorna:
    - answer: Texto gerado
    - citations: Lista de documentos fonte usados
    """
    from openai import OpenAI
    client = OpenAI()

    # Criar mapa de fontes
    source_map = {i+1: source for i, source in enumerate(sources)}
    numbered_context = "\n\n".join([
        f"[{i+1}] {source['text']}"
        for i, source in enumerate(sources)
    ])

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "Responda usando as fontes fornecidas. Cite as fontes como [1], [2], etc."
        }, {
            "role": "user",
            "content": f"Contexto:\n{numbered_context}\n\nPergunta: {query}"
        }]
    )

    answer = response.choices[0].message.content
    citations = extract_citations(answer, source_map)

    return {
        "answer": answer,
        "citations": citations
    }

---

Parte 6: Padrões de Produção

Pipeline de Indexação

class IndexingPipeline:
    """
    Pipeline completo de indexação para documentos RAG.
    """
    def __init__(self, vector_db_client, embedding_model: str = "text-embedding-3-small"):
        self.client = vector_db_client
        self.embedding_model = embedding_model
        self.openai_client = OpenAI()

    def ingest_document(self, document: dict):
        """Inserir documento no banco de dados vetorial."""
        # Etapa 1: Extrair texto
        raw_text = self.extract_text(document)

        # Etapa 2: Chunking
        chunks = semantic_chunking(raw_text)

        # Etapa 3: Gerar embeddings
        embeddings = self.get_embeddings(chunks)

        # Etapa 4: Inserir em lote
        points = [
            PointStruct(
                id=hash(chunk),
                vector=embedding,
                payload={
                    "text": chunk,
                    "source": document['filename'],
                    "page": None
                }
            )
            for chunk, embedding in zip(chunks, embeddings)
        ]
        self.client.upsert(
            collection_name="documents",
            points=points
        )

Pipeline de Consulta

class QueryPipeline:
    """
    Pipeline completo de consulta para sistemas RAG.
    """
    def __init__(self, vector_db_client, llm_client):
        self.vector_client = vector_db_client
        self.llm_client = llm_client

    def query(self, user_query: str, filters: dict = None) -> dict:
        """Executar consulta RAG completa."""
        # Etapa 1: Expandir consulta
        expanded_queries = expand_query(user_query)

        # Etapa 2: Buscar documentos
        results = []
        for query in expanded_queries:
            query_vector = get_embeddings([query])[0]
            batch = filtered_search(query_vector, filters)
            results.extend(batch)

        # Etapa 3: Reordenar resultados
        reranked = rerank_results(user_query, results)

        # Etapa 4: Otimizar contexto
        context = optimize_context(user_query, reranked)

        # Etapa 5: Gerar resposta com citações
        response = generate_with_citations(user_query, context, reranked)

        return response

Considerações para Implantação

1. Monitoramento: Acompanhe latência de recuperação, recall e taxas de acerto
2. Estratégia de Cache: Implemente cache de embeddings e resultados frequentes
3. Escalabilidade: Use sharding e replicação para grandes conjuntos de dados
4. Atualizações: Implemente indexação incremental para documentos novos
5. Fallbacks: Tenha estratégias de fallback para falhas de LLM ou banco de dados vetorial
6. Segurança: Sanitize entradas, implemente autenticação e gerencie chaves de API

---

Boas Práticas Resumidas

Seleção de Banco de Dados

• Use Qdrant para produção auto-hospedada
• Use Pinecone para prototipagem rápida e gerenciada
• Use Chroma para desenvolvimento local

Chunking

• Tamanho fixo para prototipagem
• Semântico para documentos em linguagem natural
• Hierárquico para documentos estruturados
• Janela deslizante quando precisão de contexto é crítica

Embeddings

• OpenAI text-embedding-3-small para produção, qualidade geral
• Sentence Transformers para auto-hospedado, sem custos de API
• Dimensões maiores para melhor qualidade, dimensões menores para velocidade

Otimização de Recuperação

• Use busca híbrida para melhor recall
• Expanda consultas para capturar variações
• Reordene candidatos principais para melhor precisão
• Filtre por metadados para recuperação de subconjunto

Gerenciamento de Contexto

• Otimize o contexto para caber na janela LLM
• Acompanhe citações para rastreabilidade
• Classifique por relevância antes de truncar
• Remova documentos que não agregam valor

---

Leitura Adicional

• Documentação do Qdrant
• Documentação do Pinecone
• Guia de Embeddings da OpenAI
• Documentação do Sentence Transformers
• Documentação do LangChain
• Guia de RAG do LlamaIndex