Vector Databases: Pinecone — Embeddings, Similarity Search & RAG

# =============================================
# FULL RAG PIPELINE
# =============================================
# pip install pinecone openai

import openai
from pinecone import Pinecone

# Setup
pc = Pinecone(api_key="YOUR_PINECONE_KEY")
index = pc.Index("knowledge-base")
openai_client = openai.OpenAI(api_key="YOUR_OPENAI_KEY")


def get_embedding(text, model="text-embedding-3-small"):
    """Buat embedding dari teks"""
    response = openai_client.embeddings.create(
        model=model, input=text
    )
    return response.data[0].embedding


def retrieve_context(query, top_k=5, namespace=None):
    """Ambil konteks relevan dari Pinecone"""
    query_vector = get_embedding(query)

    results = index.query(
        vector=query_vector,
        top_k=top_k,
        include_metadata=True,
        namespace=namespace
    )

    contexts = []
    sources = []
    for match in results.matches:
        if match.score > 0.7:  # Threshold kemiripan
            contexts.append(match.metadata.get("chunk_text", ""))
            sources.append({
                "id": match.id,
                "title": match.metadata.get("title", ""),
                "score": match.score
            })

    return contexts, sources


def generate_answer(query, contexts, sources):
    """Generate jawaban menggunakan LLM"""
    context_text = "\n\n---\n\n".join(contexts)

    system_prompt = """Anda adalah asisten AI yang membantu menjawab
pertanyaan berdasarkan konteks yang diberikan.

Aturan:
1. Jawab HANYA berdasarkan konteks yang diberikan
2. Jika informasi tidak ada di konteks, katakan "Saya tidak menemukan
   informasi tersebut dalam database kami"
3. Berikan jawaban yang jelas dan informatif
4. Sebutkan sumber jika relevan"""

    user_prompt = f"""Konteks dari database:
---
{context_text}
---

Pertanyaan pengguna: {query}

Jawab pertanyaan berdasarkan konteks di atas."""

    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.3,  # Rendah = lebih faktual
        max_tokens=1000
    )

    return response.choices[0].message.content


def rag_pipeline(query, namespace=None):
    """Full RAG pipeline"""
    # Step 1: Retrieve
    contexts, sources = retrieve_context(query, namespace=namespace)

    if not contexts:
        return "Maaf, saya tidak menemukan informasi yang relevan.", []

    # Step 2 + 3: Augment + Generate
    answer = generate_answer(query, contexts, sources)

    return answer, sources


# =============================================
# CONTOH PENGGUNAAN
# =============================================

# Pertanyaan tentang data Anda
query = "Bagaimana cara menginstall Python di Windows?"
answer, sources = rag_pipeline(query)

print("Jawaban:", answer)
print("\nSumber:")
for s in sources:
    print(f"  - {s['title']} (score: {s['score']:.3f})")


# =============================================
# INGEST DATA: Memasukkan dokumen ke Pinecone
# =============================================

def ingest_document(doc_id, text, metadata, chunk_size=500):
    """Pecah dokumen jadi chunks dan masukkan ke Pinecone"""
    # Step 1: Chunking (pecah teks panjang)
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size):
        chunk = " ".join(words[i:i + chunk_size])
        chunks.append(chunk)

    # Step 2: Embedding + Upsert
    vectors = []
    for i, chunk in enumerate(chunks):
        embedding = get_embedding(chunk)
        chunk_id = f"{doc_id}_chunk_{i}"

        vectors.append({
            "id": chunk_id,
            "values": embedding,
            "metadata": {
                **metadata,
                "chunk_text": chunk,
                "chunk_index": i,
                "parent_doc_id": doc_id
            }
        })

    # Batch upsert (Pinecone max 1000 per batch)
    batch_size = 100
    for i in range(0, len(vectors), batch_size):
        batch = vectors[i:i + batch_size]
        index.upsert(vectors=batch)

    print(f"Berhasil ingest {len(chunks)} chunks untuk doc {doc_id}")


# Contoh ingest:
ingest_document(
    doc_id="tutorial_python",
    text="Python adalah bahasa pemrograman serbaguna yang diciptakan oleh Guido van Rossum...",
    metadata={
        "title": "Tutorial Python Lengkap",
        "category": "programming",
        "source": "beebanelabs.com"
    }
)

# =============================================
# USE CASE 1: Semantic Search Engine
# =============================================

class SemanticSearchEngine:
    def __init__(self, index_name, embedding_model="text-embedding-3-small"):
        pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
        self.index = pc.Index(index_name)
        self.model = embedding_model

    def search(self, query, filters=None, top_k=10):
        vector = get_embedding(query, self.model)
        return self.index.query(
            vector=vector,
            top_k=top_k,
            include_metadata=True,
            filter=filters
        )

    def search_with_threshold(self, query, threshold=0.75, **kwargs):
        results = self.search(query, **kwargs)
        return [m for m in results.matches if m.score >= threshold]

# Usage:
engine = SemanticSearchEngine("knowledge-base")
results = engine.search("bagaimana cara deploy aplikasi ke cloud")


# =============================================
# USE CASE 2: Product Recommendation
# =============================================

def recommend_products(product_id, top_k=5):
    """Rekomendasikan produk serupa"""
    # Fetch vektor produk yang sedang dilihat
    product = index.fetch(ids=[product_id])
    product_vector = product.vectors[product_id].values

    # Cari produk dengan embedding mirip
    similar = index.query(
        vector=product_vector,
        top_k=top_k + 1,  # +1 karena produk sendiri juga muncul
        include_metadata=True,
        filter={"in_stock": {"$eq": True}}
    )

    # Exclude produk yang sedang dilihat
    return [m for m in similar.matches if m.id != product_id][:top_k]


# =============================================
# USE CASE 3: Duplicate Detection
# =============================================

def find_duplicates(texts, threshold=0.95):
    """Temukan teks yang hampir identik"""
    embeddings = [get_embedding(t) for t in texts]
    duplicates = []

    for i in range(len(texts)):
        results = index.query(
            vector=embeddings[i],
            top_k=5,
            include_metadata=True
        )
        for match in results.matches:
            if match.score >= threshold and match.id != f"doc_{i}":
                duplicates.append({
                    "original": texts[i],
                    "duplicate": match.metadata.get("text", ""),
                    "score": match.score
                })

    return duplicates


# =============================================
# USE CASE 4: Image Similarity Search
# =============================================
# pip install transformers torch
# from transformers import CLIPProcessor, CLIPModel

# def image_to_vector(image_path):
#     model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
#     processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
#
#     image = Image.open(image_path)
#     inputs = processor(images=image, return_tensors="pt")
#     vector = model.get_image_features(**inputs)
#     return vector.detach().numpy().flatten().tolist()
#
# def search_similar_images(image_path, top_k=5):
#     query_vector = image_to_vector(image_path)
#     return index.query(vector=query_vector, top_k=top_k)