111 lines
4.5 KiB
Python
111 lines
4.5 KiB
Python
"""
|
||
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"]
|