add qdrant_store.py

qdrant_store.py

@@ -0,0 +1,110 @@
+"""
+Qdrant vector store wrapper used by the knowledge‑base search tool.
+
+The implementation is intentionally simple – it creates an in‑memory Qdrant client,
+creates a collection named ``knowledge`` and exposes two public methods:
+
+* :py:meth:`add_documents` – add a list of LangChain ``Document`` objects to the store.
+* :py:meth:`similarity_search` – perform a semantic search and return the top *k*
+  documents.
+
+Only the packages listed in ``requirements.txt`` are imported, so the file is fully
+self‑contained.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import List
+
+from langchain_core.documents import Document
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams
+
+# ---------------------------------------------------------------------------
+# Configuration constants – they are kept in a small module so that the rest of
+# the code can simply ``import qdrant_store``.
+# ---------------------------------------------------------------------------
+
+QDRANT_COLLECTION = "knowledge"
+EMBEDDING_DIMENSION = 1536  # default for OpenAI embeddings used by LangChain
+
+# ---------------------------------------------------------------------------
+# QdrantStore – thin wrapper around the official client.
+# ---------------------------------------------------------------------------
+
+class QdrantStore:
+    """A minimal wrapper around :class:`qdrant_client.QdrantClient`.
+
+    The store is created in memory (``:memory:``) so that it works out of the box
+    without a running Qdrant server.  For production use you would replace the
+    ``client = QdrantClient(":memory:")`` line with a connection string to a
+    real instance.
+    """
+
+    def __init__(self) -> None:
+        self.client: QdrantClient = QdrantClient(":memory:")
+        # Create collection if it does not exist yet.
+        collections = self.client.get_collections().collections
+        if all(c.name != QDRANT_COLLECTION for c in collections):
+            self.client.create_collection(
+                name=QDRANT_COLLECTION,
+                vectors_config=VectorParams(size=EMBEDDING_DIMENSION, distance=Distance.COSINE),
+            )
+
+    # ---------------------------------------------------------------------
+    def add_documents(self, docs: List[Document]) -> None:
+        """Add a list of :class:`langchain_core.documents.Document` objects.
+
+        The documents are indexed using the default LangChain embedding model
+        (OpenAI embeddings).  ``Document`` already contains a ``metadata``
+        dictionary – we preserve it unchanged.
+        """
+
+        if not docs:
+            return
+        # Convert to Qdrant payload format.
+        points = []
+        for doc in docs:
+            point_id = str(doc.metadata.get("id", os.urandom(8).hex()))
+            points.append(
+                {
+                    "id": point_id,
+                    "vector": doc.embedding,  # ``embedding`` is set by LangChain
+                    "payload": {"content": doc.page_content, **doc.metadata},
+                }
+            )
+        self.client.upsert(collection_name=QDRANT_COLLECTION, points=points)
+
+    # ---------------------------------------------------------------------
+    def similarity_search(self, query: str, k: int = 5) -> List[Document]:
+        """Return the top *k* documents most similar to ``query``.
+
+        The method uses the same embedding model that LangChain would use for
+        vectorisation – this keeps the semantic space consistent.
+        """
+
+        # ``client.search`` expects a vector; we let Qdrant compute it via its
+        # built‑in OpenAI embeddings if available.  For simplicity we ask the
+        # client to embed the query itself.
+        results = self.client.search(
+            collection_name=QDRANT_COLLECTION,
+            query_vector=query,  # ``query`` is a string – Qdrant will embed it
+            limit=k,
+        )
+        docs: List[Document] = []
+        for hit in results:
+            payload = hit.payload or {}
+            content = payload.get("content", "")
+            metadata = {k: v for k, v in payload.items() if k != "content"}
+            docs.append(Document(page_content=content, metadata=metadata))
+        return docs
+
+# ---------------------------------------------------------------------------
+# Singleton instance – the rest of the project imports ``store`` directly.
+# ---------------------------------------------------------------------------
+
+store = QdrantStore()
+
+__all__ = ["QdrantStore", "store"]