Innova_Hackthon / services /pinecone_service.py
GitHub Actions
Deploy from GitHub: 68bd5c9838c0ba7d11f5a74b54e80085410bc33a
6543cd2
Raw
History Blame Contribute Delete
2.03 kB
import os
import logging
from pinecone import Pinecone
logger = logging.getLogger(__name__)
class PineconeService:
def __init__(self):
api_key = os.environ.get("PINECONE_API_KEY")
index_name = os.environ.get("PINECONE_INDEX_NAME", os.environ.get("PINECONE_INDEX", "fashion-products"))
self.client = Pinecone(api_key=api_key)
self.index = self.client.Index(index_name)
logger.info(f"Pinecone connected to index: {index_name}")
def query(self, vector: list[float], top_k: int = 10, filter: dict | None = None) -> list[dict]:
logger.info(f"Pinecone β€” query: top_k={top_k}, filter={filter}")
kwargs = {
"vector": vector,
"top_k": top_k,
"include_metadata": True,
"include_values": False,
}
if filter:
kwargs["filter"] = filter
res = self.index.query(**kwargs)
results = [
{
"id": m["id"],
"score": m.get("score", 0),
"metadata": m.get("metadata", {}),
}
for m in (res.get("matches") or [])
]
logger.info(
f"Pinecone β€” {len(results)} result(s):\n" +
"\n".join(f" {r['metadata'].get('name', r['id'])} | score: {r['score']:.3f}" for r in results)
)
return results
def fetch_by_id(self, record_id: str) -> dict | None:
logger.info(f"Pinecone β€” fetch_by_id: \"{record_id}\"")
res = self.index.fetch(ids=[record_id])
records = res.get("vectors") or res.get("records") or {}
record = records.get(record_id)
if not record:
logger.warning(f"Pinecone β€” record \"{record_id}\" not found")
return None
result = {
"id": record.get("id", record_id),
"score": 1,
"metadata": record.get("metadata", {}),
}
logger.info(f"Pinecone β€” found: \"{result['metadata'].get('name', record_id)}\"")
return result