📚 Tutoriais•17 min de leitura
RAG do Zero: Retrieval Augmented Generation com Python
Aprenda a construir um sistema RAG completo do zero usando ChromaDB, embeddings da OpenAI, LangChain e Python — indexação de documentos, busca semântica e geração de respostas contextualizadas
#rag#langchain#chromadb#openai#python#embeddings
RAG do Zero: Retrieval Augmented Generation com Python
Retrieval Augmented Generation (RAG) é uma das técnicas mais poderosas e práticas da IA moderna. Em vez de depender exclusivamente do conhecimento interno de um modelo de linguagem, o RAG permite que o modelo consulte uma base de documentos externa antes de gerar uma resposta — combinando o melhor da busca por informação com a fluência da geração de texto.
Neste tutorial, vamos construir um sistema RAG completo do zero. Você vai aprender a indexar documentos, criar embeddings, armazená-los em um banco vetorial e usar tudo isso para gerar respostas precisas e contextualizadas com LLMs.
O Que É RAG e Por Que Usar
O Problema que RAG Resolve
Modelos de linguagem como GPT-4 e Claude são incrivelmente capazes, mas têm limitações fundamentais:
- Conhecimento estático: o treinamento tem um ponto de corte. O modelo não sabe sobre eventos ou documentos publicados após esse ponto.
- Conhecimento genérico: o modelo conhece muitas coisas, mas não conhece seus documentos — manuais internos, políticas da empresa, documentação técnica proprietária.
- Alucinações: quando o modelo não sabe a resposta, ele pode inventar uma que parece convincente mas é incorreta.
Como RAG Funciona
O fluxo de um sistema RAG é:
Pergunta do Usuário
│
▼
┌─────────────────┐
│ 1. Embedding │ Converte a pergunta em vetor numérico
└────────┬────────┘
│
▼
┌─────────────────┐
│ 2. Retrieval │ Busca documentos similares no banco vetorial
└────────┬────────┘
│
▼
┌─────────────────┐
│ 3. Augment │ Combina documentos encontrados com a pergunta
└────────┬────────┘
│
▼
┌─────────────────┐
│ 4. Generate │ LLM gera resposta baseada no contexto
└─────────────────┘
Pré-requisitos
Instalação das Bibliotecas
pip install chromadb langchain langchain-openai langchain-community \
openai tiktoken unstructured pypdf python-docx
Configuração da API Key
export OPENAI_API_KEY="sk-sua-chave-aqui"
Estrutura do Projeto
rag-project/
├── app/
│ ├── __init__.py
│ ├── indexer.py # Indexação de documentos
│ ├── retriever.py # Busca semântica
│ ├── generator.py # Geração de respostas
│ ├── rag_chain.py # Pipeline RAG completo
│ └── chunking.py # Estratégias de chunking
├── documents/ # Documentos para indexar
├── chroma_db/ # Banco vetorial persistente
├── main.py # Ponto de entrada
├── requirements.txt
└── README.md
Parte 1: Entendendo Embeddings
O Que São Embeddings
Embeddings são representações numéricas (vetores) de texto. Textos com significados semelhantes terão vetores próximos no espaço vetorial. Isso permite buscar documentos por significado, não apenas por palavras-chave.
"""Demonstração básica de embeddings e similaridade semântica."""
from openai import OpenAI
import numpy as np
client = OpenAI()
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
"""Gera embedding para um texto usando a API da OpenAI."""
response = client.embeddings.create(input=text, model=model)
return response.data[0].embedding
def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
"""Calcula a similaridade do cosseno entre dois vetores."""
a = np.array(vec_a)
b = np.array(vec_b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
# Demonstração
texts = [
"Python é uma linguagem de programação popular",
"JavaScript é muito usado no desenvolvimento web",
"O gato dormiu no sofá durante toda a tarde",
"A linguagem Python é amplamente utilizada por programadores",
]
embeddings = [get_embedding(text) for text in texts]
print("Similaridades semânticas:")
print(f" Dimensão dos embeddings: {len(embeddings[0])}")
print()
for i in range(len(texts)):
for j in range(i + 1, len(texts)):
sim = cosine_similarity(embeddings[i], embeddings[j])
print(f' "{texts[i][:50]}..."')
print(f' vs "{texts[j][:50]}..."')
print(f" Similaridade: {sim:.4f}")
print()
Ao executar, você verá que os textos sobre programação terão alta similaridade entre si, enquanto o texto sobre o gato terá baixa similaridade com todos os outros.
Parte 2: Carregamento e Chunking de Documentos
Por Que Chunking é Importante
Documentos longos precisam ser divididos em pedaços (chunks) menores porque:
- Embeddings funcionam melhor com textos curtos e focados.
- A janela de contexto do LLM é limitada.
- Chunks menores permitem busca mais precisa.
Estratégias de Chunking
Crie o arquivo app/chunking.py:
"""Estratégias de chunking para documentos."""
from langchain.text_splitter import (
CharacterTextSplitter,
RecursiveCharacterTextSplitter,
MarkdownHeaderTextSplitter,
)
from langchain.schema import Document
def recursive_chunk(
text: str,
chunk_size: int = 1000,
chunk_overlap: int = 200,
metadata: dict | None = None,
) -> list[Document]:
"""
Chunking recursivo — a estratégia mais versátil.
Tenta dividir por parágrafos, depois por sentenças, depois por palavras.
"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ". ", ", ", " ", ""],
length_function=len,
)
chunks = splitter.create_documents(
texts=[text],
metadatas=[metadata or {}],
)
# Adicionar índice do chunk aos metadados
for i, chunk in enumerate(chunks):
chunk.metadata["chunk_index"] = i
chunk.metadata["total_chunks"] = len(chunks)
return chunks
def markdown_chunk(text: str, metadata: dict | None = None) -> list[Document]:
"""
Chunking baseado em headers Markdown.
Ideal para documentação técnica estruturada.
"""
headers_to_split = [
("#", "header_1"),
("##", "header_2"),
("###", "header_3"),
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split)
chunks = splitter.split_text(text)
# Adicionar metadados extras
for i, chunk in enumerate(chunks):
if metadata:
chunk.metadata.update(metadata)
chunk.metadata["chunk_index"] = i
return chunks
def semantic_chunk(
text: str,
chunk_size: int = 1000,
metadata: dict | None = None,
) -> list[Document]:
"""
Chunking com contexto semântico — adiciona contexto dos headers
anteriores a cada chunk.
"""
lines = text.split("\n")
chunks = []
current_chunk = []
current_headers = []
current_size = 0
for line in lines:
# Detectar headers
if line.startswith("#"):
level = len(line) - len(line.lstrip("#"))
# Remover headers de nível igual ou inferior
current_headers = [
h for h in current_headers if h[0] < level
]
current_headers.append((level, line.strip("# ").strip()))
current_chunk.append(line)
current_size += len(line) + 1 # +1 para o \n
if current_size >= chunk_size:
# Criar contexto dos headers
header_context = " > ".join(h[1] for h in current_headers)
chunk_text = "\n".join(current_chunk)
if header_context:
chunk_text = f"Contexto: {header_context}\n\n{chunk_text}"
doc = Document(
page_content=chunk_text,
metadata={**(metadata or {}), "chunk_index": len(chunks)},
)
chunks.append(doc)
# Reset com overlap (manter últimas linhas)
overlap_lines = current_chunk[-3:]
current_chunk = overlap_lines
current_size = sum(len(l) + 1 for l in overlap_lines)
# Último chunk
if current_chunk:
header_context = " > ".join(h[1] for h in current_headers)
chunk_text = "\n".join(current_chunk)
if header_context:
chunk_text = f"Contexto: {header_context}\n\n{chunk_text}"
chunks.append(
Document(
page_content=chunk_text,
metadata={**(metadata or {}), "chunk_index": len(chunks)},
)
)
return chunks
Carregamento de Documentos
Crie o arquivo app/indexer.py:
"""Indexação de documentos no banco vetorial ChromaDB."""
import logging
from pathlib import Path
import chromadb
from chromadb.config import Settings
from langchain_community.document_loaders import (
DirectoryLoader,
PyPDFLoader,
TextLoader,
UnstructuredMarkdownLoader,
)
from langchain_openai import OpenAIEmbeddings
from app.chunking import recursive_chunk
logger = logging.getLogger(__name__)
CHROMA_DIR = Path("chroma_db")
COLLECTION_NAME = "documents"
class DocumentIndexer:
"""Gerencia a indexação de documentos no ChromaDB."""
def __init__(self, persist_dir: str | Path = CHROMA_DIR):
self.persist_dir = Path(persist_dir)
self.persist_dir.mkdir(exist_ok=True)
self.client = chromadb.PersistentClient(
path=str(self.persist_dir),
settings=Settings(anonymized_telemetry=False),
)
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.collection = self.client.get_or_create_collection(
name=COLLECTION_NAME,
metadata={"hnsw:space": "cosine"},
)
def load_documents(self, directory: str | Path) -> list:
"""Carrega documentos de um diretório."""
directory = Path(directory)
documents = []
# Carregar diferentes tipos de arquivo
loaders = {
"*.txt": TextLoader,
"*.md": UnstructuredMarkdownLoader,
"*.pdf": PyPDFLoader,
}
for pattern, loader_cls in loaders.items():
files = list(directory.glob(pattern))
for file_path in files:
try:
loader = loader_cls(str(file_path))
docs = loader.load()
for doc in docs:
doc.metadata["source"] = str(file_path)
doc.metadata["filename"] = file_path.name
documents.extend(docs)
logger.info(f"Carregado: {file_path.name} ({len(docs)} páginas)")
except Exception as e:
logger.error(f"Erro ao carregar {file_path}: {e}")
logger.info(f"Total de documentos carregados: {len(documents)}")
return documents
def index_documents(
self,
directory: str | Path,
chunk_size: int = 1000,
chunk_overlap: int = 200,
) -> int:
"""Carrega, divide e indexa documentos no ChromaDB."""
# Carregar documentos
documents = self.load_documents(directory)
if not documents:
logger.warning("Nenhum documento encontrado para indexar!")
return 0
# Dividir em chunks
all_chunks = []
for doc in documents:
chunks = recursive_chunk(
text=doc.page_content,
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
metadata=doc.metadata,
)
all_chunks.extend(chunks)
logger.info(f"Total de chunks criados: {len(all_chunks)}")
# Gerar embeddings e indexar
batch_size = 100
total_indexed = 0
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i + batch_size]
texts = [chunk.page_content for chunk in batch]
metadatas = [chunk.metadata for chunk in batch]
ids = [f"doc_{i + j}" for j in range(len(batch))]
# Gerar embeddings
embedding_vectors = self.embeddings.embed_documents(texts)
# Adicionar ao ChromaDB
self.collection.add(
documents=texts,
embeddings=embedding_vectors,
metadatas=metadatas,
ids=ids,
)
total_indexed += len(batch)
logger.info(f"Indexados: {total_indexed}/{len(all_chunks)}")
logger.info(f"Indexação concluída! {total_indexed} chunks indexados.")
return total_indexed
def get_stats(self) -> dict:
"""Retorna estatísticas da coleção."""
return {
"collection_name": COLLECTION_NAME,
"total_documents": self.collection.count(),
"persist_dir": str(self.persist_dir),
}
Parte 3: Busca Semântica
Crie o arquivo app/retriever.py:
"""Busca semântica no banco vetorial."""
import logging
from dataclasses import dataclass
from pathlib import Path
import chromadb
from chromadb.config import Settings
from langchain_openai import OpenAIEmbeddings
logger = logging.getLogger(__name__)
CHROMA_DIR = Path("chroma_db")
COLLECTION_NAME = "documents"
@dataclass
class SearchResult:
"""Resultado de uma busca semântica."""
content: str
metadata: dict
score: float # Similaridade (0.0 a 1.0, maior = mais similar)
class SemanticRetriever:
"""Realiza busca semântica no banco vetorial ChromaDB."""
def __init__(self, persist_dir: str | Path = CHROMA_DIR):
self.client = chromadb.PersistentClient(
path=str(persist_dir),
settings=Settings(anonymized_telemetry=False),
)
self.collection = self.client.get_collection(name=COLLECTION_NAME)
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
def search(
self,
query: str,
top_k: int = 5,
min_score: float = 0.0,
filter_metadata: dict | None = None,
) -> list[SearchResult]:
"""
Busca documentos semanticamente similares à query.
Args:
query: Texto de busca
top_k: Número máximo de resultados
min_score: Score mínimo de similaridade (0.0 a 1.0)
filter_metadata: Filtros opcionais de metadados
Returns:
Lista de SearchResult ordenados por similaridade
"""
# Gerar embedding da query
query_embedding = self.embeddings.embed_query(query)
# Buscar no ChromaDB
where = filter_metadata if filter_metadata else None
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
where=where,
include=["documents", "metadatas", "distances"],
)
# Converter distância cosseno para score de similaridade
search_results = []
for doc, metadata, distance in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0],
):
# ChromaDB retorna distância, convertemos para similaridade
score = 1.0 - distance
if score >= min_score:
search_results.append(
SearchResult(content=doc, metadata=metadata, score=score)
)
return search_results
def search_with_reranking(
self,
query: str,
top_k: int = 5,
initial_k: int = 20,
) -> list[SearchResult]:
"""
Busca com reranking — busca mais resultados inicialmente
e usa um segundo critério para refinar.
"""
# Busca inicial mais ampla
initial_results = self.search(query, top_k=initial_k)
if not initial_results:
return []
# Reranking simples: combinar score de similaridade com diversidade
reranked = []
selected_sources = set()
for result in initial_results:
source = result.metadata.get("source", "")
# Bônus de diversidade: preferir resultados de fontes diferentes
diversity_bonus = 0.05 if source not in selected_sources else 0.0
adjusted_score = result.score + diversity_bonus
reranked.append((adjusted_score, result))
selected_sources.add(source)
# Ordenar por score ajustado
reranked.sort(key=lambda x: x[0], reverse=True)
return [result for _, result in reranked[:top_k]]
def format_context(self, results: list[SearchResult]) -> str:
"""Formata os resultados de busca como contexto para o LLM."""
if not results:
return "Nenhum documento relevante encontrado."
context_parts = []
for i, result in enumerate(results, 1):
source = result.metadata.get("filename", "desconhecido")
context_parts.append(
f"[Documento {i}] (Fonte: {source}, Relevância: {result.score:.2f})\n"
f"{result.content}"
)
return "\n\n---\n\n".join(context_parts)
Parte 4: Geração de Respostas
Crie o arquivo app/generator.py:
"""Geração de respostas usando LLM com contexto RAG."""
import logging
from openai import OpenAI
logger = logging.getLogger(__name__)
class ResponseGenerator:
"""Gera respostas contextualizadas usando LLM."""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def generate(
self,
query: str,
context: str,
system_prompt: str | None = None,
temperature: float = 0.1,
max_tokens: int = 2000,
) -> dict:
"""
Gera uma resposta baseada na query e no contexto recuperado.
Args:
query: Pergunta do usuário
context: Contexto recuperado do banco vetorial
system_prompt: Prompt de sistema customizado
temperature: Criatividade da resposta (0.0 = determinístico)
max_tokens: Máximo de tokens na resposta
Returns:
Dicionário com resposta e metadados
"""
if system_prompt is None:
system_prompt = self._default_system_prompt()
user_message = f"""Contexto dos documentos:
{context}
---
Pergunta: {query}
Instruções: Responda à pergunta baseando-se EXCLUSIVAMENTE no contexto fornecido acima.
Se a informação não estiver no contexto, diga claramente que não encontrou a informação
nos documentos disponíveis. Cite as fontes quando possível."""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
temperature=temperature,
max_tokens=max_tokens,
)
answer = response.choices[0].message.content
usage = response.usage
return {
"answer": answer,
"model": self.model,
"usage": {
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
},
}
def generate_with_citations(
self,
query: str,
context: str,
) -> dict:
"""Gera resposta com citações estruturadas."""
system_prompt = """Você é um assistente especializado que responde perguntas
baseado em documentos fornecidos.
REGRAS:
1. Use APENAS informações presentes no contexto fornecido.
2. Cite as fontes usando o formato [Documento N].
3. Se a informação não estiver no contexto, diga explicitamente.
4. Seja preciso e conciso.
5. Responda em português brasileiro.
FORMATO DA RESPOSTA:
Resposta principal com [Documento 1] citações inline [Documento 2].
Fontes utilizadas:
- [Documento 1]: breve descrição
- [Documento 2]: breve descrição"""
return self.generate(query, context, system_prompt=system_prompt)
def _default_system_prompt(self) -> str:
return """Você é um assistente inteligente que responde perguntas baseado em
documentos fornecidos. Responda em português brasileiro de forma clara e precisa.
Se a informação não estiver disponível nos documentos, informe ao usuário."""
Parte 5: Pipeline RAG Completo
Crie o arquivo app/rag_chain.py:
"""Pipeline RAG completo: retrieval + generation."""
import logging
import time
from dataclasses import dataclass
from app.generator import ResponseGenerator
from app.retriever import SemanticRetriever
logger = logging.getLogger(__name__)
@dataclass
class RAGResponse:
"""Resposta completa do pipeline RAG."""
answer: str
sources: list[dict]
query: str
retrieval_time_ms: float
generation_time_ms: float
total_time_ms: float
tokens_used: dict
model: str
class RAGPipeline:
"""Pipeline RAG completo: busca + geração."""
def __init__(
self,
retriever: SemanticRetriever | None = None,
generator: ResponseGenerator | None = None,
top_k: int = 5,
min_score: float = 0.3,
):
self.retriever = retriever or SemanticRetriever()
self.generator = generator or ResponseGenerator()
self.top_k = top_k
self.min_score = min_score
def query(
self,
question: str,
top_k: int | None = None,
with_citations: bool = True,
) -> RAGResponse:
"""
Executa o pipeline RAG completo.
Args:
question: Pergunta do usuário
top_k: Número de documentos a recuperar (override)
with_citations: Se True, gera resposta com citações
Returns:
RAGResponse com resposta, fontes e métricas
"""
k = top_k or self.top_k
total_start = time.time()
# 1. Retrieval
retrieval_start = time.time()
results = self.retriever.search_with_reranking(
query=question, top_k=k
)
retrieval_time = (time.time() - retrieval_start) * 1000
# 2. Formatar contexto
context = self.retriever.format_context(results)
# 3. Geração
generation_start = time.time()
if with_citations:
response = self.generator.generate_with_citations(question, context)
else:
response = self.generator.generate(question, context)
generation_time = (time.time() - generation_start) * 1000
total_time = (time.time() - total_start) * 1000
# 4. Compilar fontes
sources = [
{
"content": r.content[:200] + "..." if len(r.content) > 200 else r.content,
"source": r.metadata.get("filename", "desconhecido"),
"score": round(r.score, 4),
}
for r in results
]
logger.info(
f"RAG query concluída: {len(results)} docs recuperados, "
f"retrieval={retrieval_time:.0f}ms, generation={generation_time:.0f}ms"
)
return RAGResponse(
answer=response["answer"],
sources=sources,
query=question,
retrieval_time_ms=round(retrieval_time, 2),
generation_time_ms=round(generation_time, 2),
total_time_ms=round(total_time, 2),
tokens_used=response["usage"],
model=response["model"],
)
def interactive_session(self):
"""Inicia uma sessão interativa de perguntas e respostas."""
print("=" * 60)
print("Sistema RAG Interativo")
print("Digite suas perguntas (ou 'sair' para encerrar)")
print("=" * 60)
while True:
question = input("\n📝 Pergunta: ").strip()
if question.lower() in ("sair", "exit", "quit"):
print("Até logo!")
break
if not question:
continue
print("\n🔍 Buscando documentos relevantes...")
response = self.query(question)
print(f"\n💡 Resposta:\n{response.answer}")
print(f"\n📊 Métricas:")
print(f" Documentos recuperados: {len(response.sources)}")
print(f" Tempo de busca: {response.retrieval_time_ms:.0f}ms")
print(f" Tempo de geração: {response.generation_time_ms:.0f}ms")
print(f" Tokens usados: {response.tokens_used['total_tokens']}")
if response.sources:
print(f"\n📚 Fontes:")
for s in response.sources:
print(f" - {s['source']} (relevância: {s['score']:.2f})")
Parte 6: Ponto de Entrada
Crie o arquivo main.py:
"""
RAG System - Ponto de entrada principal.
Uso: python main.py [index|query|interactive]
"""
import argparse
import logging
import sys
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
def cmd_index(args):
"""Indexa documentos de um diretório."""
from app.indexer import DocumentIndexer
indexer = DocumentIndexer()
count = indexer.index_documents(
directory=args.directory,
chunk_size=args.chunk_size,
chunk_overlap=args.overlap,
)
print(f"\n✅ {count} chunks indexados com sucesso!")
print(f"📊 Estatísticas: {indexer.get_stats()}")
def cmd_query(args):
"""Faz uma pergunta ao sistema RAG."""
from app.rag_chain import RAGPipeline
rag = RAGPipeline(top_k=args.top_k)
response = rag.query(args.question)
print(f"\n💡 Resposta:\n{response.answer}")
print(f"\n📊 Tempo total: {response.total_time_ms:.0f}ms")
def cmd_interactive(args):
"""Inicia sessão interativa."""
from app.rag_chain import RAGPipeline
rag = RAGPipeline(top_k=args.top_k)
rag.interactive_session()
def main():
parser = argparse.ArgumentParser(description="Sistema RAG com ChromaDB e LangChain")
subparsers = parser.add_subparsers(dest="command", help="Comandos disponíveis")
# Comando: index
index_parser = subparsers.add_parser("index", help="Indexar documentos")
index_parser.add_argument("directory", help="Diretório com documentos")
index_parser.add_argument("--chunk-size", type=int, default=1000)
index_parser.add_argument("--overlap", type=int, default=200)
index_parser.set_defaults(func=cmd_index)
# Comando: query
query_parser = subparsers.add_parser("query", help="Fazer uma pergunta")
query_parser.add_argument("question", help="Pergunta")
query_parser.add_argument("--top-k", type=int, default=5)
query_parser.set_defaults(func=cmd_query)
# Comando: interactive
interactive_parser = subparsers.add_parser("interactive", help="Sessão interativa")
interactive_parser.add_argument("--top-k", type=int, default=5)
interactive_parser.set_defaults(func=cmd_interactive)
args = parser.parse_args()
if args.command is None:
parser.print_help()
sys.exit(1)
args.func(args)
if __name__ == "__main__":
main()
Parte 7: Uso Prático
Passo 1: Preparar Documentos
Coloque seus documentos na pasta documents/:
mkdir -p documents
# Exemplo: criar um documento de teste
cat > documents/manual-python.md << 'EOF'
# Manual de Python para Iniciantes
## Variáveis e Tipos de Dados
Python é uma linguagem dinamicamente tipada. Os principais tipos são:
- int: números inteiros (42, -7)
- float: números decimais (3.14, -0.5)
- str: strings de texto ("hello", 'world')
- bool: valores booleanos (True, False)
- list: listas ordenadas ([1, 2, 3])
- dict: dicionários chave-valor ({"nome": "Ana"})
## Funções
Funções são definidas com a palavra-chave def:
```python
def saudacao(nome: str) -> str:
return f"Olá, {nome}!"
Classes
Python suporta programação orientada a objetos:
class Animal:
def __init__(self, nome: str, especie: str):
self.nome = nome
self.especie = especie
def falar(self) -> str:
return f"{self.nome} faz um som"
EOF
### Passo 2: Indexar
```bash
python main.py index documents/
Passo 3: Fazer Perguntas
# Pergunta única
python main.py query "Quais são os tipos de dados em Python?"
# Sessão interativa
python main.py interactive
Parte 8: Otimizações Avançadas
Hybrid Search: Combinando Busca Vetorial e Keyword
def hybrid_search(
retriever: SemanticRetriever,
query: str,
top_k: int = 5,
alpha: float = 0.7,
) -> list[SearchResult]:
"""
Combina busca semântica (vetorial) com busca por keyword (BM25).
alpha: peso da busca semântica (0.0 = só keyword, 1.0 = só semântica)
"""
from rank_bm25 import BM25Okapi
# Busca semântica
semantic_results = retriever.search(query, top_k=top_k * 2)
# Busca keyword (BM25)
all_docs = retriever.collection.get(include=["documents", "metadatas"])
corpus = [doc.lower().split() for doc in all_docs["documents"]]
bm25 = BM25Okapi(corpus)
tokenized_query = query.lower().split()
bm25_scores = bm25.get_scores(tokenized_query)
# Normalizar scores
max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
normalized_bm25 = bm25_scores / max_bm25
# Combinar scores
combined = {}
for result in semantic_results:
doc_id = result.content[:100]
combined[doc_id] = {
"result": result,
"semantic_score": result.score,
"bm25_score": 0.0,
}
for i, (doc, metadata) in enumerate(
zip(all_docs["documents"], all_docs["metadatas"])
):
doc_id = doc[:100]
if doc_id in combined:
combined[doc_id]["bm25_score"] = float(normalized_bm25[i])
elif normalized_bm25[i] > 0.3:
combined[doc_id] = {
"result": SearchResult(content=doc, metadata=metadata, score=0.0),
"semantic_score": 0.0,
"bm25_score": float(normalized_bm25[i]),
}
# Calcular score final
final_results = []
for entry in combined.values():
final_score = (
alpha * entry["semantic_score"]
+ (1 - alpha) * entry["bm25_score"]
)
entry["result"].score = final_score
final_results.append(entry["result"])
final_results.sort(key=lambda x: x.score, reverse=True)
return final_results[:top_k]
Cache de Embeddings
import hashlib
import json
from pathlib import Path
class EmbeddingCache:
"""Cache local de embeddings para evitar chamadas repetidas à API."""
def __init__(self, cache_dir: str = ".embedding_cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _hash(self, text: str) -> str:
return hashlib.sha256(text.encode()).hexdigest()
def get(self, text: str) -> list[float] | None:
cache_file = self.cache_dir / f"{self._hash(text)}.json"
if cache_file.exists():
return json.loads(cache_file.read_text())
return None
def set(self, text: str, embedding: list[float]) -> None:
cache_file = self.cache_dir / f"{self._hash(text)}.json"
cache_file.write_text(json.dumps(embedding))
Conclusão
Neste tutorial, construímos um sistema RAG completo que inclui:
- Embeddings: representação vetorial de textos com a API da OpenAI.
- Chunking inteligente: múltiplas estratégias para dividir documentos.
- Indexação com ChromaDB: armazenamento persistente e eficiente de vetores.
- Busca semântica: recuperação de documentos por significado, não apenas palavras-chave.
- Geração contextualizada: respostas precisas baseadas nos documentos recuperados.
- Pipeline completo: interface de linha de comando para indexação e consulta.
RAG é uma técnica transformadora que permite construir sistemas de IA que realmente conhecem seus dados. Com as ferramentas que vimos, você está pronto para implementar RAG em seus projetos.
Próximos Passos
- Explorar embeddings locais (Sentence Transformers) para reduzir custos.
- Implementar reranking com modelos cross-encoder.
- Adicionar suporte a multimodalidade (imagens e tabelas).
- Criar uma interface web com Streamlit ou Gradio.
- Integrar com LangSmith para observabilidade do pipeline.
- Experimentar com diferentes LLMs (Claude, Gemini, Llama local).
Acompanhe o Inteligência em Código para mais tutoriais sobre RAG, LangChain e aplicações práticas de IA.