pinecone_utils.py

# pinecone_utils.py

from pinecone import Pinecone, ServerlessSpec
from config import PINECONE_API_KEY, PINECONE_ENVIRONMENT, INDEX_NAME, CONTEXT_FIELDS
from sentence_transformers import SentenceTransformer
import torch

# Conectar a Pinecone
def connect_to_pinecone():
    # Crear una instancia de Pinecone
    pc = Pinecone(api_key=PINECONE_API_KEY)

    # Verificar si el índice existe
    index_names = pc.list_indexes().names()
    if INDEX_NAME not in index_names:
        # Si el índice no existe, crearlo
        pc.create_index(
            name=INDEX_NAME,
            dimension=1024,  # Asegúrate de que esta dimensión coincida con la de tus embeddings
            metric='cosine',  # Puedes cambiar el métrico según tus necesidades
            spec=ServerlessSpec(
                cloud='aws',
                region=PINECONE_ENVIRONMENT  # Asegúrate de que este sea el entorno correcto
            )
        )

    # Conectar al índice
    index = pc.Index(INDEX_NAME)
    return index

# Función para realizar la búsqueda vectorial
def vector_search(query, embedding_model, index):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    # Generar el embedding utilizando el modelo de embeddings
    xq = embedding_model.encode(query, convert_to_tensor=True, device=device)

    # Convertir el tensor a lista
    xq = xq.cpu().tolist()

    # Realizar búsqueda vectorial en el índice de Pinecone
    res = index.query(vector=xq, top_k=3, include_metadata=True)
    if res and res['matches']:
        return [
            {
                'content': ' '.join(f"{k}: {v}" for k, v in match['metadata'].items() if k in CONTEXT_FIELDS and k != 'Tag'),
                'metadata': match['metadata'],
                'score': match.get('score', 0)
            }
            for match in res['matches']
            if 'metadata' in match
        ]
    return []