Spaces:
Sleeping
Sleeping
| 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 | |