Skill RAG do Pinecone

Esta skill te guiará na construção de um pipeline RAG de produção ou um sistema de memória persistente para agentes usando Pinecone. Siga o fluxo de trabalho do início ao fim — não pule etapas nem vá direto para o código antes de entender o que o usuário realmente precisa.

Antes de começar — faça uma pergunta

Antes de escrever qualquer código, identifique qual destes dois casos de uso se aplica:

A — RAG sobre documentos: O usuário quer indexar um corpus (PDFs, documentos, código, páginas da web) e recuperar trechos relevantes para fundamentar as respostas do LLM.

B — Memória de agente: O usuário quer que um agente se lembre de fatos, decisões ou contexto entre sessões ou entre múltiplos agentes que compartilham uma base de conhecimento.

A configuração é semelhante, mas a estratégia de namespace e os padrões de recuperação diferem. Se o usuário não disse, pergunte: "Isso é para recuperação de documentos, memória de agente, ou ambos?" Em seguida, siga o fluxo de trabalho relevante abaixo.

Etapa 1 — Escolha a configuração do seu índice

Escolha o tipo de índice antes de escrever qualquer código. Errar aqui significa recriar o índice mais tarde.

Serverless (recomendado para a maioria dos casos)

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="PINECONE_API_KEY")

if "my-index" not in pc.list_indexes().names():
    pc.create_index(
        name="my-index",
        dimension=1536,        # must match your embedding model exactly
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
index = pc.Index("my-index")

Baseado em Pods (para produção de alto rendimento consistente)

from pinecone import PodSpec

pc.create_index(
    name="my-index-prod",
    dimension=1536,
    metric="cosine",
    spec=PodSpec(environment="us-east1-gcp", pod_type="p1.x1")
)

Referência rápida de dimensão — combine isso exatamente com seu modelo de embedding:

Checkpoint: Índice existe, dimensão corresponde ao modelo de embedding, index.describe_index_stats() retorna sem erro.

Etapa 2 — Crie embeddings e insira documentos

Sempre agrupe as inserções — nunca insira um vetor por vez.

from openai import OpenAI

client = OpenAI()

def embed(texts: list[str]) -> list[list[float]]:
    res = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [r.embedding for r in res.data]

def upsert_docs(index, docs: list[dict], namespace: str = "default"):
    """docs = [{"id": "...", "text": "...", "metadata": {...}}]"""
    BATCH = 100
    for i in range(0, len(docs), BATCH):
        batch = docs[i:i + BATCH]
        vecs = [
            {
                "id": d["id"],
                "values": emb,
                "metadata": {**d.get("metadata", {}), "text": d["text"]}
            }
            for d, emb in zip(batch, embed([d["text"] for d in batch]))
        ]
        index.upsert(vectors=vecs, namespace=namespace)

Sempre armazene o texto original nos metadados — isso evita uma segunda consulta no momento da recuperação.

Checkpoint: index.describe_index_stats() mostra a contagem de vetores > 0 no namespace de destino.

Etapa 3 — Escolha a estratégia de recuperação

Busca densa (semântica) — use para a maioria dos casos

def search(index, query: str, top_k: int = 5, namespace: str = "default",
           filter: dict = None) -> list[dict]:
    [q_emb] = embed([query])
    results = index.query(
        vector=q_emb, top_k=top_k, namespace=namespace,
        include_metadata=True, filter=filter
    )
    return [{"text": m.metadata["text"], "score": m.score, "id": m.id}
            for m in results.matches]

Busca híbrida (semântica + palavra-chave BM25) — use quando o corpus tiver terminologia exata

Use híbrida quando o domínio tiver termos precisos que a busca semântica perde: citações legais, códigos médicos, SKUs de produtos, nomes de métodos de API.

from pinecone_text.sparse import BM25Encoder

bm25 = BM25Encoder().default()
bm25.fit([d["text"] for d in docs])  # ajuste uma vez no seu corpus

def hybrid_search(index, query: str, top_k: int = 5, alpha: float = 0.7):
    """alpha=1.0 é puramente denso; alpha=0.0 é puramente esparso."""
    dense = [v * alpha for v in embed([query])[0]]
    sparse_raw = bm25.encode_queries(query)
    sparse = {
        "indices": sparse_raw["indices"],
        "values": [v * (1 - alpha) for v in sparse_raw["values"]]
    }

…