scripts/create_pinecone_index.py

#!/usr/bin/env python3
"""One-time script to create the Pinecone index for the multi-agent RAG system.

This script avoids the need to install the Pinecone CLI by using the Python SDK
directly for this single administrative task.

The index is a plain dense-vector serverless index. Embeddings are generated
by the app using NVIDIA models (nvidia/nv-embed-v1 or similar).

Usage:
    uv run python scripts/create_pinecone_index.py

Environment Variables:
    PINECONE_API_KEY  - Required: Your Pinecone API key
    PINECONE_INDEX    - Optional: Index name (default: multi-agent-index)
"""

import os
import sys
from pathlib import Path

# Add project root to path so app imports work
sys.path.insert(0, str(Path(__file__).parent.parent))

from dotenv import load_dotenv

load_dotenv()


def get_embedding_dimension() -> int:
    """Probe the configured embeddings model to determine output dimension."""
    from app.core.llm_factory import get_embeddings_model

    embeddings = get_embeddings_model()
    vector = embeddings.embed_query("dimension probe")
    return len(vector)


def main() -> int:
    api_key = os.getenv("PINECONE_API_KEY")
    if not api_key:
        print("Error: PINECONE_API_KEY environment variable not set")
        return 1

    index_name = os.getenv("PINECONE_INDEX", "multi-agent-index")

    print("Pinecone index setup")
    print(f"  Index name : {index_name}")

    # Determine embedding dimension from the app's configured model
    try:
        dimension = get_embedding_dimension()
        print(f"  Dimension  : {dimension} (from configured embedding model)")
    except Exception as exc:
        print(f"Error: Could not determine embedding dimension: {exc}")
        print("You can also set it manually by editing this script.")
        return 1

    # Create index via SDK (one-time admin task)
    try:
        from pinecone import Pinecone

        pc = Pinecone(api_key=api_key)

        if pc.has_index(index_name):
            print(f"\nIndex '{index_name}' already exists. Skipping creation.")
            info = pc.describe_index(index_name)
            if isinstance(info, dict):
                print(f"  Status   : {info.get('status', 'unknown')}")
                print(f"  Dimension: {info.get('dimension', 'unknown')}")
                print(f"  Metric   : {info.get('metric', 'unknown')}")
            else:
                print(f"  Status   : {getattr(info, 'status', 'unknown')}")
                print(f"  Dimension: {getattr(info, 'dimension', 'unknown')}")
                print(f"  Metric   : {getattr(info, 'metric', 'unknown')}")
            return 0

        print(f"\nCreating serverless dense-vector index '{index_name}' ...")
        print(f"  Cloud  : aws")
        print(f"  Region : us-east-1")
        print(f"  Metric : cosine")
        print(f"  Dim    : {dimension}")

        pc.create_index(
            name=index_name,
            dimension=dimension,
            metric="cosine",
            spec={
                "serverless": {
                    "cloud": "aws",
                    "region": "us-east-1",
                }
            },
        )

        # Wait briefly and verify
        import time

        time.sleep(5)

        if pc.has_index(index_name):
            info = pc.describe_index(index_name)
            if isinstance(info, dict):
                print(f"\n  Status   : {info.get('status', 'unknown')}")
                print(f"  Dimension: {info.get('dimension', 'unknown')}")
                print(f"  Metric   : {info.get('metric', 'unknown')}")
            else:
                print(f"\n  Status   : {getattr(info, 'status', 'unknown')}")
                print(f"  Dimension: {getattr(info, 'dimension', 'unknown')}")
                print(f"  Metric   : {getattr(info, 'metric', 'unknown')}")
            print(f"\nIndex '{index_name}' created successfully.")
            print("You can now run: uv run python scripts/seed_rag_data.py")
            return 0
        else:
            print("\nWarning: Index creation initiated but not yet visible.")
            print("Check the Pinecone console for status.")
            return 0

    except Exception as exc:
        print(f"\nError: Failed to create index: {exc}")
        return 1


if __name__ == "__main__":
    raise SystemExit(main())