Infraestrutura RAG

Infraestrutura de produção para Geração Aumentada por Recuperação (Retrieval-Augmented Generation): ingere documentos, gera embeddings, armazena em bancos de dados vetoriais e serve respostas de LLM fundamentadas.

Quando Usar Esta Habilidade

Use esta habilidade quando:

• Estiver construindo um sistema de Perguntas e Respostas (Q&A) sobre documentos internos
• Estiver implementando busca semântica em grandes coleções de documentos
• Estiver reduzindo alucinações de LLM com contexto recuperado
• Estiver configurando pipelines de embeddings e infraestrutura de armazenamento vetorial
• Estiver implantando busca híbrida (densa + esparsa/BM25)

Pré-requisitos

• Python 3.10+ com pip
• Um banco de dados vetorial (Qdrant, Weaviate, Pinecone ou pgvector)
• Um modelo de embedding (OpenAI, Cohere ou local via sentence-transformers)
• Um endpoint de LLM (API OpenAI ou vLLM auto-hospedado)
• Docker para implantação local de banco de dados vetorial

Visão Geral da Arquitetura

Documents → Chunker → Embedder → Vector Store
                                      ↓
User Query → Embedder → Vector Store (search) → Reranker → LLM → Answer

Pipeline de Embeddings

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid

# Local embedding model (no API cost)
model = SentenceTransformer("BAAI/bge-large-en-v1.5")

# Connect to Qdrant
client = QdrantClient("http://localhost:6333")

# Create collection
client.create_collection(
    collection_name="knowledge-base",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
)

def ingest_documents(docs: list[dict]):
    """Chunk, embed, and upsert documents."""
    points = []
    for doc in docs:
        chunks = chunk_text(doc["text"], chunk_size=512, overlap=50)
        embeddings = model.encode(chunks, batch_size=32, show_progress_bar=True)
        for chunk, embedding in zip(chunks, embeddings):
            points.append(PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding.tolist(),
                payload={"text": chunk, "source": doc["source"], "title": doc["title"]},
            ))
    client.upsert(collection_name="knowledge-base", points=points)
    print(f"Ingested {len(points)} chunks")

Estratégias de Chunking

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_text(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """Recursive character splitter — best general-purpose strategy."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    return splitter.split_text(text)

# For code/markdown — use language-aware splitter
from langchain.text_splitter import MarkdownHeaderTextSplitter

headers = [("#", "H1"), ("##", "H2"), ("###", "H3")]
md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers)

Busca Híbrida (Densa + Esparsa)

from qdrant_client.models import SparseVector, SparseVectorParams, NamedSparseVector
from fastembed import SparseTextEmbedding

# Qdrant hybrid collection (dense + BM25 sparse)
client.create_collection(
    collection_name="hybrid-kb",
    vectors_config={"dense": VectorParams(size=1024, distance=Distance.COSINE)},
    sparse_vectors_config={"sparse": SparseVectorParams()},
)

sparse_model = SparseTextEmbedding("prithivida/Splade_PP_en_v1")

def hybrid_search(query: str, top_k: int = 10) -> list[dict]:
    dense_vec = model.encode(query).tolist()
    sparse_vec = list(sparse_model.embed(query))[0]

    results = client.query_points(
        collection_name="hybrid-kb",
        prefetch=[
            {"query": dense_vec, "using": "dense", "limit": 20},
            {"query": SparseVector(indices=sparse_vec.indices.tolist(),
                                   values=sparse_vec.values.tolist()),
             "using": "sparse", "limit": 20},
        ],
        query={"fusion": "rrf"},   # Reciprocal Rank Fusion
        limit=top_k,
    )
    return [{"text": p.payload["text"], "score": p.score} for p in results.points]

Reclassificação (Reranking)

import cohere

co = cohere.Client("sua-chave-api")

def rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    """Reclassifica os trechos recuperados por relevância (melhora a qualidade do RAG em ~20-30%)."""
    response = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=candidates,
        top_n=top_n,
    )

…