feat: add qdrant vector adapter
This commit is contained in:
@@ -0,0 +1,98 @@
|
||||
# Task 5 Report — Implement the Qdrant VectorStore adapter
|
||||
|
||||
## Status
|
||||
|
||||
Implemented on 2026-08-08 in `/Users/mp/projects/ThothII/.worktrees/git-workspace-registry`.
|
||||
|
||||
## RED evidence
|
||||
|
||||
Command:
|
||||
|
||||
```bash
|
||||
cd harness
|
||||
./.venv/bin/pytest tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py -q
|
||||
```
|
||||
|
||||
Observed before implementation:
|
||||
|
||||
- exit code `2`
|
||||
- collection failed during import because the adapter did not exist yet
|
||||
|
||||
Representative failures:
|
||||
|
||||
- `ModuleNotFoundError: No module named 'tht.adapters.vector.qdrant'`
|
||||
|
||||
## GREEN evidence
|
||||
|
||||
Focused behavior suite:
|
||||
|
||||
```bash
|
||||
cd harness
|
||||
./.venv/bin/pytest tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py -q
|
||||
```
|
||||
|
||||
- exit code `0`
|
||||
- `31 passed, 1 warning`
|
||||
|
||||
Touched-file lint:
|
||||
|
||||
```bash
|
||||
cd harness
|
||||
./.venv/bin/ruff check tht/adapters/vector/qdrant.py tht/adapters/vector/__init__.py \
|
||||
tht/ports/vector.py tht/vectorstore/records.py tht/vectorstore/store.py \
|
||||
tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py
|
||||
```
|
||||
|
||||
- exit code `0`
|
||||
- `All checks passed!`
|
||||
|
||||
Patch hygiene:
|
||||
|
||||
```bash
|
||||
git diff --check
|
||||
```
|
||||
|
||||
- exit code `0`
|
||||
|
||||
## What changed
|
||||
|
||||
- added `QdrantVectorStore` with direct `requests`-based REST calls for:
|
||||
- `GET /collections/{collection}`
|
||||
- `PUT /collections/{collection}`
|
||||
- `PUT /collections/{collection}/index`
|
||||
- `PUT /collections/{collection}/points?wait=true`
|
||||
- `POST /collections/{collection}/points/query`
|
||||
- `POST /collections/{collection}/points/scroll`
|
||||
- `POST /collections/{collection}/points/delete?wait=true`
|
||||
- implemented idempotent collection provisioning for `1024` dimensions and `Cosine` distance
|
||||
- created deterministic UUIDv5 point IDs from workspace, semantic kind, and canonical record key
|
||||
- preserved canonical record identity and only upserted/deleted points matching the exact workspace
|
||||
and generation filters
|
||||
- added Qdrant payload helpers so stored payloads carry:
|
||||
- `workspace_id`
|
||||
- grouped semantic `kind` (`schema`, `evidence`, `memory`)
|
||||
- original `record_kind`
|
||||
- canonical `record_key`
|
||||
- `content_hash`
|
||||
- existing Thoth metadata fields
|
||||
- mapped Qdrant payloads back into existing `VectorHit` objects without losing the original
|
||||
Thoth kind
|
||||
- exported the new adapter from the public vector adapter package and added focused contract tests
|
||||
- sanitized timeout and malformed-response failures so CLI-facing callers do not leak raw endpoint
|
||||
details
|
||||
|
||||
## Self-review
|
||||
|
||||
- confirmed collection mismatch fails without any delete/recreate path
|
||||
- confirmed every query/scroll/delete operation includes a workspace filter
|
||||
- confirmed the adapter never deletes or rewrites unrelated Qdrant points
|
||||
- added keyword payload indexes for all filter-critical fields used here, including `document_id`
|
||||
for exact Evidence filtering
|
||||
|
||||
## Concerns
|
||||
|
||||
- the requested `adversarial-review` skill could not run its full external reviewer flow in this
|
||||
environment because the skill’s referenced `brain/` files are missing at
|
||||
`/Users/mp/.agents/skills/adversarial-review`; I performed a manual adversarial self-review
|
||||
instead
|
||||
- the focused suite still emits one pre-existing warning from `testcontainers.postgres`
|
||||
@@ -0,0 +1,312 @@
|
||||
import json
|
||||
from uuid import NAMESPACE_URL, uuid5
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from tht.adapters.vector.qdrant import QdrantVectorStore, point_id
|
||||
from tht.ports.vector import VectorStoreError, VectorWriteRecord
|
||||
from tht.vectorstore.records import VectorRecord
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, status_code: int, payload=None, text: str | None = None):
|
||||
self.status_code = status_code
|
||||
self._payload = payload
|
||||
self.text = text if text is not None else (
|
||||
"" if payload is None else json.dumps(payload)
|
||||
)
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return 200 <= self.status_code < 300
|
||||
|
||||
def json(self):
|
||||
if isinstance(self._payload, Exception):
|
||||
raise self._payload
|
||||
return self._payload
|
||||
|
||||
|
||||
class FakeQdrantHttp:
|
||||
def __init__(self, *, dimension=1024, distance="Cosine"):
|
||||
self.dimension = dimension
|
||||
self.distance = distance
|
||||
self.collection = None
|
||||
self.payload_indexes: set[str] = set()
|
||||
self.points: dict[str, dict] = {}
|
||||
self.calls: list[tuple[str, str, dict | None]] = []
|
||||
self.fail_request: Exception | None = None
|
||||
self.malformed_query = False
|
||||
self.malformed_scroll = False
|
||||
|
||||
def request(self, method, url, *, json=None, timeout=None):
|
||||
self.calls.append((method, url, json))
|
||||
if self.fail_request is not None:
|
||||
raise self.fail_request
|
||||
|
||||
path = url.split("://", 1)[-1].split("/", 1)[-1]
|
||||
path = "/" + path.split("?", 1)[0]
|
||||
|
||||
if method == "GET" and path == "/collections/workspace-semantic":
|
||||
if self.collection is None:
|
||||
return FakeResponse(404, {"status": "error"})
|
||||
return FakeResponse(200, {
|
||||
"result": {
|
||||
"config": {
|
||||
"params": {
|
||||
"vectors": {"size": self.dimension, "distance": self.distance}
|
||||
}
|
||||
},
|
||||
"payload_schema": {
|
||||
field: {"data_type": "keyword"} for field in sorted(self.payload_indexes)
|
||||
},
|
||||
}
|
||||
})
|
||||
|
||||
if method == "PUT" and path == "/collections/workspace-semantic":
|
||||
self.collection = json
|
||||
self.dimension = json["vectors"]["size"]
|
||||
self.distance = json["vectors"]["distance"]
|
||||
return FakeResponse(200, {"status": "ok"})
|
||||
|
||||
if method == "PUT" and path == "/collections/workspace-semantic/index":
|
||||
self.payload_indexes.add(json["field_name"])
|
||||
return FakeResponse(200, {"status": "ok"})
|
||||
|
||||
if method == "PUT" and path == "/collections/workspace-semantic/points":
|
||||
for point in json["points"]:
|
||||
self.points[point["id"]] = point
|
||||
return FakeResponse(200, {"result": {"status": "acknowledged"}})
|
||||
|
||||
if method == "POST" and path == "/collections/workspace-semantic/points/query":
|
||||
if self.malformed_query:
|
||||
return FakeResponse(200, {"result": {"points": "nope"}})
|
||||
wanted = _match_points(self.points.values(), json["filter"])
|
||||
scored = sorted(
|
||||
(
|
||||
{
|
||||
"id": point["id"],
|
||||
"score": point.get("score", 0.9),
|
||||
"payload": point["payload"],
|
||||
}
|
||||
for point in wanted
|
||||
),
|
||||
key=lambda point: (-point["score"], point["payload"]["record_key"]),
|
||||
)
|
||||
return FakeResponse(200, {"result": {"points": scored[: json["limit"]]}})
|
||||
|
||||
if method == "POST" and path == "/collections/workspace-semantic/points/scroll":
|
||||
if self.malformed_scroll:
|
||||
return FakeResponse(200, {"result": {"points": "bad"}})
|
||||
wanted = sorted(
|
||||
_match_points(self.points.values(), json["filter"]),
|
||||
key=lambda point: point["payload"]["record_key"],
|
||||
)
|
||||
return FakeResponse(200, {"result": {"points": wanted}})
|
||||
|
||||
if method == "POST" and path == "/collections/workspace-semantic/points/delete":
|
||||
doomed = [point["id"] for point in _match_points(self.points.values(), json["filter"])]
|
||||
for point_id_value in doomed:
|
||||
self.points.pop(point_id_value, None)
|
||||
return FakeResponse(200, {"result": {"status": "acknowledged"}})
|
||||
|
||||
raise AssertionError((method, path, json))
|
||||
|
||||
|
||||
def _match_points(points, flt):
|
||||
matches = []
|
||||
must = flt["must"]
|
||||
for point in points:
|
||||
payload = point["payload"]
|
||||
if all(_match_clause(payload, clause) for clause in must):
|
||||
matches.append(point)
|
||||
return matches
|
||||
|
||||
|
||||
def _match_clause(payload, clause):
|
||||
key = clause["key"]
|
||||
match = clause["match"]
|
||||
if "value" in match:
|
||||
return payload.get(key) == match["value"]
|
||||
if "any" in match:
|
||||
return payload.get(key) in set(match["any"])
|
||||
raise AssertionError(clause)
|
||||
|
||||
|
||||
def _write_record(record_id: str, kind: str, *, metadata=None):
|
||||
return VectorWriteRecord(
|
||||
record=VectorRecord(
|
||||
id=record_id,
|
||||
kind=kind,
|
||||
ref=f"ref:{record_id}",
|
||||
title=f"title:{record_id}",
|
||||
content=f"content:{record_id}",
|
||||
metadata=metadata or {},
|
||||
),
|
||||
embedding=[0.1] * 1024,
|
||||
content_hash="sha256:" + "a" * 64,
|
||||
)
|
||||
|
||||
|
||||
def _store(fake: FakeQdrantHttp) -> QdrantVectorStore:
|
||||
return QdrantVectorStore(
|
||||
base_url="http://qdrant:6333",
|
||||
collection="workspace-semantic",
|
||||
workspace_id="demo",
|
||||
expected_dimension=1024,
|
||||
request=fake.request,
|
||||
)
|
||||
|
||||
|
||||
def test_point_id_is_deterministic_uuidv5():
|
||||
assert point_id("demo", "memory", "memory:1") == str(
|
||||
uuid5(NAMESPACE_URL, "thothii:demo:memory:memory:1")
|
||||
)
|
||||
|
||||
|
||||
def test_upsert_creates_collection_and_keyword_indexes_idempotently():
|
||||
fake = FakeQdrantHttp()
|
||||
store = _store(fake)
|
||||
|
||||
assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1
|
||||
assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1
|
||||
|
||||
creates = [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")]
|
||||
assert len(creates) == 1
|
||||
assert creates[0][2] == {"vectors": {"size": 1024, "distance": "Cosine"}}
|
||||
assert fake.payload_indexes == {
|
||||
"content_hash",
|
||||
"document_id",
|
||||
"kind",
|
||||
"record_key",
|
||||
"record_kind",
|
||||
"vector_generation",
|
||||
"workspace_id",
|
||||
}
|
||||
|
||||
|
||||
def test_upsert_refuses_collection_dimension_or_distance_mismatch_without_recreating():
|
||||
fake = FakeQdrantHttp(dimension=384, distance="Dot")
|
||||
fake.collection = {"vectors": {"size": 384, "distance": "Dot"}}
|
||||
store = _store(fake)
|
||||
|
||||
with pytest.raises(VectorStoreError, match="Qdrant collection configuration mismatch"):
|
||||
store.upsert("memory", [_write_record("memory:1", "memory")])
|
||||
|
||||
creates = [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")]
|
||||
assert creates == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("record", "semantic_kind"),
|
||||
[
|
||||
(_write_record("schema_column:patients.id", "schema_column"), "schema"),
|
||||
(
|
||||
_write_record(
|
||||
"demo:gen:11111111111111111111111111111111:chunk:1",
|
||||
"evidence",
|
||||
metadata={
|
||||
"workspace_id": "demo",
|
||||
"vector_generation": "gen:11111111111111111111111111111111",
|
||||
"document_id": "doc:abc",
|
||||
},
|
||||
),
|
||||
"evidence",
|
||||
),
|
||||
(_write_record("memory:1", "memory"), "memory"),
|
||||
],
|
||||
)
|
||||
def test_upsert_serializes_qdrant_point_payloads(record, semantic_kind):
|
||||
fake = FakeQdrantHttp()
|
||||
store = _store(fake)
|
||||
|
||||
store.upsert("memory" if semantic_kind == "memory" else "evidence" if semantic_kind == "evidence" else "schema_records", [record])
|
||||
|
||||
point = next(iter(fake.points.values()))
|
||||
assert point["id"] == point_id("demo", semantic_kind, record.record.id)
|
||||
assert point["vector"] == record.embedding
|
||||
assert point["payload"]["workspace_id"] == "demo"
|
||||
assert point["payload"]["kind"] == semantic_kind
|
||||
assert point["payload"]["record_kind"] == record.record.kind
|
||||
assert point["payload"]["record_key"] == record.record.id
|
||||
assert point["payload"]["content_hash"] == record.content_hash
|
||||
|
||||
|
||||
def test_search_filters_by_workspace_and_allowed_record_kinds():
|
||||
fake = FakeQdrantHttp()
|
||||
store = _store(fake)
|
||||
store.upsert("memory", [_write_record("memory:1", "memory")])
|
||||
other = next(iter(fake.points.values())).copy()
|
||||
other["id"] = point_id("other", "memory", "memory:2")
|
||||
other["payload"] = {**other["payload"], "workspace_id": "other", "record_key": "memory:2"}
|
||||
fake.points[other["id"]] = other
|
||||
solved = next(iter(fake.points.values())).copy()
|
||||
solved["id"] = point_id("demo", "memory", "solved:1")
|
||||
solved["payload"] = {**solved["payload"], "record_key": "solved:1", "record_kind": "solved_question"}
|
||||
fake.points[solved["id"]] = solved
|
||||
|
||||
hits = store.search(["memory"], [0.2] * 1024, limit=5, kinds=["memory"])
|
||||
|
||||
assert [hit.id for hit in hits] == ["memory:1"]
|
||||
query_call = next(call for call in fake.calls if call[0] == "POST" and call[1].endswith("/points/query?wait=true") is False and call[1].endswith("/points/query"))
|
||||
assert query_call[2]["filter"] == {
|
||||
"must": [
|
||||
{"key": "workspace_id", "match": {"value": "demo"}},
|
||||
{"key": "record_kind", "match": {"any": ["memory"]}},
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def test_existing_hashes_health_and_exact_generation_inventory_and_delete():
|
||||
fake = FakeQdrantHttp()
|
||||
store = _store(fake)
|
||||
generation = "gen:" + "1" * 32
|
||||
keep = "gen:" + "2" * 32
|
||||
store.upsert("evidence", [
|
||||
_write_record(
|
||||
f"demo:{generation}:chunk:1",
|
||||
"evidence",
|
||||
metadata={"workspace_id": "demo", "vector_generation": generation, "document_id": "doc:1"},
|
||||
),
|
||||
_write_record(
|
||||
f"demo:{keep}:chunk:2",
|
||||
"evidence",
|
||||
metadata={"workspace_id": "demo", "vector_generation": keep, "document_id": "doc:2"},
|
||||
),
|
||||
])
|
||||
|
||||
assert store.existing_hashes("evidence", ["evidence"]) == {
|
||||
f"demo:{generation}:chunk:1": "sha256:" + "a" * 64,
|
||||
f"demo:{keep}:chunk:2": "sha256:" + "a" * 64,
|
||||
}
|
||||
assert store.list_evidence_generations("evidence", "demo") == [generation, keep]
|
||||
assert store.delete_generation("evidence", generation, "demo") == 1
|
||||
assert store.list_evidence_generations("evidence", "demo") == [keep]
|
||||
|
||||
health = store.health()
|
||||
assert health.ok is True
|
||||
assert health.read_reachable is True
|
||||
assert health.write_reachable is True
|
||||
assert health.observed_dimensions == (1024,)
|
||||
assert health.dimension_compatible is True
|
||||
|
||||
|
||||
def test_sanitizes_timeout_and_malformed_responses():
|
||||
fake = FakeQdrantHttp()
|
||||
store = _store(fake)
|
||||
fake.fail_request = requests.Timeout("dial tcp 10.0.0.9:6333: i/o timeout")
|
||||
|
||||
with pytest.raises(VectorStoreError, match="Qdrant request failed") as timeout:
|
||||
store.search(["memory"], [0.2] * 1024, limit=1)
|
||||
assert "10.0.0.9" not in str(timeout.value)
|
||||
|
||||
fake.fail_request = None
|
||||
store.upsert("memory", [_write_record("memory:1", "memory")])
|
||||
fake.malformed_query = True
|
||||
with pytest.raises(VectorStoreError, match="Qdrant returned malformed query response"):
|
||||
store.search(["memory"], [0.2] * 1024, limit=1)
|
||||
|
||||
fake.malformed_query = False
|
||||
fake.malformed_scroll = True
|
||||
with pytest.raises(VectorStoreError, match="Qdrant returned malformed scroll response"):
|
||||
store.existing_hashes("memory", ["memory"])
|
||||
@@ -3,14 +3,15 @@ from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from tht.adapters.vector.thoth_http import ThothHttpVectorStore
|
||||
from tht.adapters.vector.legacy_direct import LegacyDirectVectorStore
|
||||
from tht.adapters.vector.qdrant import QdrantVectorStore
|
||||
from tht.adapters.vector.thoth_http import ThothHttpVectorStore
|
||||
from tht.evidence.model import EvidenceDoc
|
||||
from tht.ports.vector import (
|
||||
VectorHit,
|
||||
VectorReadUnavailable,
|
||||
VectorRecord,
|
||||
VectorStore,
|
||||
VectorReadUnavailable,
|
||||
VectorWriteRecord,
|
||||
VectorWriteUnavailable,
|
||||
)
|
||||
@@ -158,11 +159,13 @@ def test_http_store_is_runtime_vector_store():
|
||||
|
||||
|
||||
def test_vector_contract_is_exported_from_public_packages():
|
||||
from tht.adapters.vector import QdrantVectorStore as PublicQdrantStore
|
||||
from tht.adapters.vector import ThothHttpVectorStore as PublicHttpStore
|
||||
from tht.ports import VectorReadUnavailable as PublicVectorReadUnavailable
|
||||
from tht.ports import VectorStore as PublicVectorStore
|
||||
from tht.ports import VectorWriteRecord as PublicVectorWriteRecord
|
||||
from tht.ports import VectorReadUnavailable as PublicVectorReadUnavailable
|
||||
|
||||
assert PublicQdrantStore is QdrantVectorStore
|
||||
assert PublicHttpStore is ThothHttpVectorStore
|
||||
assert PublicVectorStore is VectorStore
|
||||
assert PublicVectorWriteRecord is VectorWriteRecord
|
||||
@@ -248,3 +251,24 @@ def test_legacy_direct_search_requires_a_strict_positive_integer_limit(limit):
|
||||
|
||||
with pytest.raises(ValueError, match="positive integer"):
|
||||
store.search(["memory"], [0.1], limit=limit)
|
||||
|
||||
|
||||
def test_qdrant_store_is_runtime_vector_store():
|
||||
store = QdrantVectorStore(
|
||||
base_url="http://qdrant:6333",
|
||||
collection="workspace-semantic",
|
||||
workspace_id="demo",
|
||||
expected_dimension=1024,
|
||||
request=lambda *args, **kwargs: MagicMock(
|
||||
ok=True,
|
||||
status_code=200,
|
||||
text='{"result":{"config":{"params":{"vectors":{"size":1024,"distance":"Cosine"}}},"payload_schema":{}}}',
|
||||
json=lambda: {
|
||||
"result": {
|
||||
"config": {"params": {"vectors": {"size": 1024, "distance": "Cosine"}}},
|
||||
"payload_schema": {},
|
||||
}
|
||||
},
|
||||
),
|
||||
)
|
||||
assert isinstance(store, VectorStore)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from tht.adapters.vector.legacy_direct import LegacyDirectVectorStore
|
||||
from tht.adapters.vector.pgvector import PgVectorStore
|
||||
from tht.adapters.vector.qdrant import QdrantVectorStore
|
||||
from tht.adapters.vector.thoth_http import ThothHttpVectorStore
|
||||
|
||||
__all__ = ["LegacyDirectVectorStore", "PgVectorStore", "ThothHttpVectorStore"]
|
||||
__all__ = ["LegacyDirectVectorStore", "PgVectorStore", "QdrantVectorStore", "ThothHttpVectorStore"]
|
||||
|
||||
@@ -0,0 +1,363 @@
|
||||
"""Qdrant-backed vector store for one workspace-owned semantic collection."""
|
||||
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from uuid import NAMESPACE_URL, uuid5
|
||||
|
||||
import requests
|
||||
|
||||
from tht.adapters.vector.pgvector import (
|
||||
COLLECTION_KINDS,
|
||||
_collection,
|
||||
_validate_collection_kinds,
|
||||
_validate_known_kinds,
|
||||
)
|
||||
from tht.ports.vector import (
|
||||
VectorCapabilities,
|
||||
VectorHealth,
|
||||
VectorStoreError,
|
||||
VectorWriteRecord,
|
||||
require_positive_limit,
|
||||
)
|
||||
from tht.vectorstore.records import qdrant_payload, qdrant_semantic_kind
|
||||
from tht.vectorstore.store import VectorHit, hit_from_metadata
|
||||
|
||||
_GENERATION = re.compile(r"gen:[0-9a-f]{32}")
|
||||
_WORKSPACE = re.compile(r"[a-z][a-z0-9_-]{0,63}")
|
||||
_KEYWORD_INDEXES = (
|
||||
"content_hash",
|
||||
"document_id",
|
||||
"kind",
|
||||
"record_key",
|
||||
"record_kind",
|
||||
"vector_generation",
|
||||
"workspace_id",
|
||||
)
|
||||
|
||||
|
||||
def point_id(workspace_id: str, kind: str, record_key: str) -> str:
|
||||
return str(uuid5(NAMESPACE_URL, f"thothii:{workspace_id}:{kind}:{record_key}"))
|
||||
|
||||
|
||||
def _sanitize_exception(exc: Exception) -> str:
|
||||
if isinstance(exc, requests.Timeout):
|
||||
return "Qdrant request failed: timeout"
|
||||
return f"Qdrant request failed: {type(exc).__name__}"
|
||||
|
||||
|
||||
class QdrantVectorStore:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
base_url: str,
|
||||
collection: str,
|
||||
workspace_id: str,
|
||||
expected_dimension: int | None = None,
|
||||
request: Callable[..., object] | None = None,
|
||||
connect_timeout: float = 2.0,
|
||||
read_timeout: float = 10.0,
|
||||
):
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._collection = collection
|
||||
self._workspace_id = workspace_id
|
||||
self._expected_dimension = expected_dimension
|
||||
self._request = request or requests.request
|
||||
self._timeout = (connect_timeout, read_timeout)
|
||||
|
||||
@property
|
||||
def capabilities(self) -> VectorCapabilities:
|
||||
return VectorCapabilities(
|
||||
search=True,
|
||||
existing_hashes=True,
|
||||
upsert=True,
|
||||
metadata_filter=True,
|
||||
delete_generation=True,
|
||||
list_evidence_generations=True,
|
||||
)
|
||||
|
||||
def health(self) -> VectorHealth:
|
||||
try:
|
||||
info = self._ensure_collection(strict=False)
|
||||
except VectorStoreError as exc:
|
||||
return VectorHealth(
|
||||
ok=False,
|
||||
detail=str(exc),
|
||||
read_configured=True,
|
||||
read_reachable=False,
|
||||
read_detail=str(exc),
|
||||
write_configured=True,
|
||||
write_reachable=False,
|
||||
write_detail=str(exc),
|
||||
expected_dimension=self._expected_dimension,
|
||||
)
|
||||
|
||||
dimensions = ()
|
||||
compatible = None
|
||||
if info is not None:
|
||||
dimension = info["config"]["params"]["vectors"]["size"]
|
||||
dimensions = (dimension,)
|
||||
compatible = (
|
||||
None if self._expected_dimension is None else dimensions == (self._expected_dimension,)
|
||||
)
|
||||
return VectorHealth(
|
||||
ok=compatible is not False,
|
||||
read_configured=True,
|
||||
read_reachable=True,
|
||||
write_configured=True,
|
||||
write_reachable=True,
|
||||
expected_dimension=self._expected_dimension,
|
||||
observed_dimensions=dimensions,
|
||||
dimension_compatible=compatible,
|
||||
)
|
||||
|
||||
def search(
|
||||
self,
|
||||
collections: list[str],
|
||||
embedding: list[float],
|
||||
*,
|
||||
limit: int,
|
||||
kinds: list[str] | None = None,
|
||||
metadata_filter: dict[str, object] | None = None,
|
||||
) -> list[VectorHit]:
|
||||
require_positive_limit(limit)
|
||||
self._validate_embedding(embedding, query=True)
|
||||
allowed_record_kinds = self._allowed_record_kinds(collections, kinds)
|
||||
if not allowed_record_kinds:
|
||||
return []
|
||||
filter_must = self._workspace_filter()
|
||||
filter_must.append({"key": "record_kind", "match": {"any": allowed_record_kinds}})
|
||||
if metadata_filter is not None:
|
||||
if set(metadata_filter) != {"vector_generation", "document_ids", "workspace_id"}:
|
||||
raise VectorStoreError("Unsupported vector metadata filter")
|
||||
generation = metadata_filter["vector_generation"]
|
||||
document_ids = metadata_filter["document_ids"]
|
||||
workspace_id = metadata_filter["workspace_id"]
|
||||
if (
|
||||
not isinstance(generation, str)
|
||||
or not isinstance(document_ids, list)
|
||||
or not all(isinstance(item, str) for item in document_ids)
|
||||
or not isinstance(workspace_id, str)
|
||||
):
|
||||
raise VectorStoreError("Invalid vector metadata filter")
|
||||
filter_must = [
|
||||
{"key": "workspace_id", "match": {"value": workspace_id}},
|
||||
{"key": "record_kind", "match": {"any": allowed_record_kinds}},
|
||||
{"key": "vector_generation", "match": {"value": generation}},
|
||||
{"key": "document_id", "match": {"any": document_ids}},
|
||||
]
|
||||
response = self._call(
|
||||
"POST",
|
||||
f"/collections/{self._collection}/points/query",
|
||||
{
|
||||
"vector": embedding,
|
||||
"limit": limit,
|
||||
"with_payload": True,
|
||||
"filter": {"must": filter_must},
|
||||
},
|
||||
)
|
||||
points = response.get("result", {}).get("points")
|
||||
if not isinstance(points, list):
|
||||
raise VectorStoreError("Qdrant returned malformed query response")
|
||||
hits = [self._hit_from_point(point) for point in points]
|
||||
return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:limit]
|
||||
|
||||
def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]:
|
||||
_collection("vectors", collection)
|
||||
_validate_collection_kinds(collection, kinds)
|
||||
points = self._scroll(
|
||||
[
|
||||
*self._workspace_filter(),
|
||||
{"key": "record_kind", "match": {"any": sorted(kinds)}},
|
||||
]
|
||||
)
|
||||
hashes: dict[str, str] = {}
|
||||
for point in points:
|
||||
payload = point.get("payload")
|
||||
if not isinstance(payload, dict):
|
||||
raise VectorStoreError("Qdrant returned malformed scroll response")
|
||||
record_key = payload.get("record_key")
|
||||
content_hash = payload.get("content_hash")
|
||||
if not isinstance(record_key, str) or not isinstance(content_hash, str):
|
||||
raise VectorStoreError("Qdrant returned malformed scroll response")
|
||||
hashes[record_key] = content_hash
|
||||
return hashes
|
||||
|
||||
def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int:
|
||||
_collection("vectors", collection)
|
||||
self._ensure_collection(strict=True)
|
||||
points = []
|
||||
for write_record in records:
|
||||
_validate_collection_kinds(collection, [write_record.record.kind])
|
||||
self._validate_embedding(write_record.embedding, query=False)
|
||||
semantic_kind = qdrant_semantic_kind(write_record.record.kind)
|
||||
points.append(
|
||||
{
|
||||
"id": point_id(self._workspace_id, semantic_kind, write_record.record.id),
|
||||
"vector": write_record.embedding,
|
||||
"payload": qdrant_payload(
|
||||
write_record.record,
|
||||
content_hash=write_record.content_hash,
|
||||
workspace_id=self._workspace_id,
|
||||
),
|
||||
}
|
||||
)
|
||||
self._call(
|
||||
"PUT",
|
||||
f"/collections/{self._collection}/points?wait=true",
|
||||
{"points": points},
|
||||
)
|
||||
return len(records)
|
||||
|
||||
def delete_generation(self, collection: str, generation: str, workspace_id: str) -> int:
|
||||
if collection != "evidence" or _GENERATION.fullmatch(generation) is None:
|
||||
raise VectorStoreError("Only exact Evidence generations may be deleted")
|
||||
if _WORKSPACE.fullmatch(workspace_id) is None:
|
||||
raise VectorStoreError("Invalid Evidence workspace namespace")
|
||||
before = len(
|
||||
self._scroll(
|
||||
[
|
||||
{"key": "workspace_id", "match": {"value": workspace_id}},
|
||||
{"key": "record_kind", "match": {"any": ["evidence"]}},
|
||||
{"key": "vector_generation", "match": {"value": generation}},
|
||||
]
|
||||
)
|
||||
)
|
||||
self._call(
|
||||
"POST",
|
||||
f"/collections/{self._collection}/points/delete?wait=true",
|
||||
{
|
||||
"filter": {
|
||||
"must": [
|
||||
{"key": "workspace_id", "match": {"value": workspace_id}},
|
||||
{"key": "record_kind", "match": {"any": ["evidence"]}},
|
||||
{"key": "vector_generation", "match": {"value": generation}},
|
||||
]
|
||||
}
|
||||
},
|
||||
)
|
||||
return before
|
||||
|
||||
def list_evidence_generations(self, collection: str, workspace_id: str) -> list[str]:
|
||||
if collection != "evidence":
|
||||
raise VectorStoreError("Only exact Evidence generations may be listed")
|
||||
if _WORKSPACE.fullmatch(workspace_id) is None:
|
||||
raise VectorStoreError("Invalid Evidence workspace namespace")
|
||||
points = self._scroll(
|
||||
[
|
||||
{"key": "workspace_id", "match": {"value": workspace_id}},
|
||||
{"key": "record_kind", "match": {"any": ["evidence"]}},
|
||||
]
|
||||
)
|
||||
generations = {
|
||||
payload["vector_generation"]
|
||||
for point in points
|
||||
if isinstance((payload := point.get("payload")), dict)
|
||||
and isinstance(payload.get("vector_generation"), str)
|
||||
and _GENERATION.fullmatch(payload["vector_generation"]) is not None
|
||||
}
|
||||
return sorted(generations)
|
||||
|
||||
def _workspace_filter(self) -> list[dict]:
|
||||
return [{"key": "workspace_id", "match": {"value": self._workspace_id}}]
|
||||
|
||||
def _allowed_record_kinds(
|
||||
self, collections: list[str], kinds: list[str] | None
|
||||
) -> list[str]:
|
||||
selected: set[str] = set()
|
||||
for collection in collections:
|
||||
_collection("vectors", collection)
|
||||
selected.update(COLLECTION_KINDS[collection])
|
||||
if kinds:
|
||||
_validate_known_kinds(kinds)
|
||||
selected &= set(kinds)
|
||||
return sorted(selected)
|
||||
|
||||
def _validate_embedding(self, embedding: list[float], *, query: bool) -> None:
|
||||
if self._expected_dimension is not None and len(embedding) != self._expected_dimension:
|
||||
raise VectorStoreError(
|
||||
"Query embedding dimension does not match configured dimension"
|
||||
if query
|
||||
else "Embedding dimension does not match configured dimension"
|
||||
)
|
||||
|
||||
def _ensure_collection(self, *, strict: bool) -> dict | None:
|
||||
response = self._call("GET", f"/collections/{self._collection}", None, allow_missing=True)
|
||||
if response is None:
|
||||
if not strict:
|
||||
return None
|
||||
self._call(
|
||||
"PUT",
|
||||
f"/collections/{self._collection}",
|
||||
{"vectors": {"size": self._expected_dimension or 1024, "distance": "Cosine"}},
|
||||
)
|
||||
for field_name in _KEYWORD_INDEXES:
|
||||
self._call(
|
||||
"PUT",
|
||||
f"/collections/{self._collection}/index",
|
||||
{"field_name": field_name, "field_schema": "keyword"},
|
||||
)
|
||||
response = self._call("GET", f"/collections/{self._collection}", None)
|
||||
result = response.get("result") if isinstance(response, dict) else None
|
||||
config = result.get("config", {}).get("params", {}).get("vectors") if isinstance(result, dict) else None
|
||||
if not isinstance(config, dict):
|
||||
raise VectorStoreError("Qdrant returned malformed collection response")
|
||||
size = config.get("size")
|
||||
distance = config.get("distance")
|
||||
if (
|
||||
self._expected_dimension is not None
|
||||
and (size != self._expected_dimension or distance != "Cosine")
|
||||
):
|
||||
raise VectorStoreError("Qdrant collection configuration mismatch")
|
||||
for field_name in _KEYWORD_INDEXES:
|
||||
if field_name not in result.get("payload_schema", {}):
|
||||
self._call(
|
||||
"PUT",
|
||||
f"/collections/{self._collection}/index",
|
||||
{"field_name": field_name, "field_schema": "keyword"},
|
||||
)
|
||||
return result
|
||||
|
||||
def _scroll(self, must: list[dict]) -> list[dict]:
|
||||
response = self._call(
|
||||
"POST",
|
||||
f"/collections/{self._collection}/points/scroll",
|
||||
{"with_payload": True, "limit": 10000, "filter": {"must": must}},
|
||||
)
|
||||
points = response.get("result", {}).get("points")
|
||||
if not isinstance(points, list):
|
||||
raise VectorStoreError("Qdrant returned malformed scroll response")
|
||||
return points
|
||||
|
||||
def _hit_from_point(self, point: dict) -> VectorHit:
|
||||
payload = point.get("payload")
|
||||
score = point.get("score")
|
||||
if not isinstance(payload, dict) or not isinstance(score, (int, float)):
|
||||
raise VectorStoreError("Qdrant returned malformed query response")
|
||||
return hit_from_metadata(float(score), payload)
|
||||
|
||||
def _call(self, method: str, path: str, payload: dict | None, allow_missing: bool = False) -> dict | None:
|
||||
try:
|
||||
response = self._request(
|
||||
method,
|
||||
f"{self._base_url}{path}",
|
||||
json=payload,
|
||||
timeout=self._timeout,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise VectorStoreError(_sanitize_exception(exc)) from exc
|
||||
if response.status_code == 404 and allow_missing:
|
||||
return None
|
||||
if not response.ok:
|
||||
raise VectorStoreError(f"Qdrant request failed: HTTP {response.status_code}")
|
||||
if response.status_code == 204 or not getattr(response, "text", ""):
|
||||
return {}
|
||||
try:
|
||||
data = response.json()
|
||||
except Exception as exc:
|
||||
raise VectorStoreError("Qdrant returned malformed JSON response") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise VectorStoreError("Qdrant returned malformed JSON response")
|
||||
return data
|
||||
|
||||
|
||||
__all__ = ["QdrantVectorStore", "point_id"]
|
||||
@@ -89,10 +89,11 @@ __all__ = [
|
||||
"VectorCapabilities",
|
||||
"VectorHealth",
|
||||
"VectorHit",
|
||||
"VectorRecord",
|
||||
"VectorReadUnavailable",
|
||||
"VectorRecord",
|
||||
"VectorStore",
|
||||
"VectorStoreError",
|
||||
"VectorWriteRecord",
|
||||
"VectorWriteUnavailable",
|
||||
"require_positive_limit",
|
||||
]
|
||||
|
||||
@@ -17,6 +17,31 @@ class VectorRecord(BaseModel):
|
||||
metadata: dict = {}
|
||||
|
||||
|
||||
def qdrant_semantic_kind(kind: str) -> str:
|
||||
if kind in {"schema_table", "schema_column"}:
|
||||
return "schema"
|
||||
if kind in {"memory", "solved_question"}:
|
||||
return "memory"
|
||||
if kind == "evidence":
|
||||
return "evidence"
|
||||
raise ValueError(f"Unsupported vector kind: {kind}")
|
||||
|
||||
|
||||
def qdrant_payload(record: VectorRecord, *, content_hash: str, workspace_id: str) -> dict:
|
||||
semantic_kind = qdrant_semantic_kind(record.kind)
|
||||
return {
|
||||
"workspace_id": workspace_id,
|
||||
"kind": semantic_kind,
|
||||
"record_kind": record.kind,
|
||||
"record_key": record.id,
|
||||
"ref": record.ref,
|
||||
"title": record.title,
|
||||
"content": record.content,
|
||||
"content_hash": content_hash,
|
||||
**record.metadata,
|
||||
}
|
||||
|
||||
|
||||
def split_markdown(text: str, max_chars: int) -> list[str]:
|
||||
"""Spezza un markdown: intero se sta nel limite, altrimenti per heading '##',
|
||||
e in ultima istanza per accumulo greedy di righe."""
|
||||
|
||||
@@ -42,7 +42,7 @@ def hit_from_metadata(similarity: float, metadata: dict | None) -> VectorHit:
|
||||
md = metadata or {}
|
||||
return VectorHit(
|
||||
id=md.get("record_key", ""),
|
||||
kind=md.get("kind", ""),
|
||||
kind=md.get("record_kind", md.get("kind", "")),
|
||||
ref=md.get("ref", ""),
|
||||
title=md.get("title", ""),
|
||||
content=md.get("content", ""),
|
||||
|
||||
Reference in New Issue
Block a user