jpmschweitzerandClaude Fable 5
8348b4bf92
- embed_batch now issues one batched /api/embed request (the old loop made one /api/embeddings round-trip per chunk) with a per-text fallback that preserves None-for-failed semantics - update_from_page embeds all chunks in that single call and stores them in one Qdrant batch upsert (upsert_points) - reindex order reversed: upsert new points first, then prune stale ids (deterministic uuid5 ids make overwrite safe) so a mid-way failure no longer leaves the page with zero vectors - VectorUpdateSummary gains status (success/partial/failed) and chunks_skipped; all-embeddings-failed keeps old vectors and reports failure instead of success=True Measured on a 7-chunk page ingest (local server, llm_tester): ~375ms -> ~181ms median over 3 runs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
135 lines
4.5 KiB
Python
135 lines
4.5 KiB
Python
"""
|
|
Offline unit tests for the batched, delete-last vector reindex path.
|
|
|
|
Pins the Phase D fixes in VectorService.update_from_page:
|
|
- embeddings come from ONE batched embed_batch call, not per-chunk embed()
|
|
- new points are upserted BEFORE stale points are deleted (deterministic
|
|
uuid5 ids make the overwrite safe), so a failure can no longer leave
|
|
the page with zero vectors
|
|
- the summary reports partial/failed status instead of unconditional
|
|
success=True when chunks are skipped
|
|
"""
|
|
|
|
import uuid
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from src.services.vector_service import VectorService
|
|
|
|
TENANT = "llm_tester"
|
|
TENANT_COLLECTION = "library_desk_llm_tester"
|
|
PAGE = {
|
|
"id": 44,
|
|
"title": "T",
|
|
"path": "users/llm_tester/page",
|
|
"content": " ".join(f"word{i}" for i in range(1200)), # 3 chunks @ 500/50
|
|
"tags": [],
|
|
}
|
|
|
|
|
|
def _chunk_id(page_id: int, idx: int) -> str:
|
|
return str(uuid.uuid5(uuid.NAMESPACE_DNS, f"page_{page_id}_chunk_{idx}"))
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_qdrant():
|
|
qdrant = MagicMock()
|
|
qdrant.collection_exists = AsyncMock(return_value=True)
|
|
qdrant.ensure_collection = AsyncMock()
|
|
qdrant.upsert_points = AsyncMock(side_effect=lambda collection_name, points: len(points))
|
|
qdrant.scroll_all_points = AsyncMock(return_value=[])
|
|
qdrant.delete_by_ids = AsyncMock(side_effect=lambda collection_name, point_ids: len(point_ids))
|
|
qdrant.delete_by_filter = AsyncMock(return_value=0)
|
|
return qdrant
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ollama():
|
|
ollama = MagicMock()
|
|
ollama.embed_batch = AsyncMock(
|
|
side_effect=lambda texts, **kw: [[0.1] * 768 for _ in texts]
|
|
)
|
|
return ollama
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_wiki():
|
|
wiki = MagicMock()
|
|
wiki.get_page = AsyncMock(return_value=dict(PAGE))
|
|
return wiki
|
|
|
|
|
|
@pytest.fixture
|
|
def service(mock_qdrant, mock_wiki, mock_ollama):
|
|
return VectorService(mock_qdrant, mock_wiki, mock_ollama)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestBatchedReindex:
|
|
async def test_single_batched_embed_and_single_upsert(
|
|
self, service, mock_qdrant, mock_ollama
|
|
):
|
|
summary = await service.update_from_page(page_id=44, user=TENANT)
|
|
|
|
assert summary.success is True
|
|
assert summary.status == "success"
|
|
assert summary.chunks_created == 3
|
|
mock_ollama.embed_batch.assert_awaited_once()
|
|
mock_qdrant.upsert_points.assert_awaited_once()
|
|
|
|
async def test_upsert_happens_before_stale_delete(self, service, mock_qdrant):
|
|
order = []
|
|
mock_qdrant.upsert_points = AsyncMock(
|
|
side_effect=lambda collection_name, points: order.append("upsert") or len(points)
|
|
)
|
|
mock_qdrant.delete_by_ids = AsyncMock(
|
|
side_effect=lambda collection_name, point_ids: order.append("delete") or len(point_ids)
|
|
)
|
|
# One stale point from a previous, longer version of the page
|
|
mock_qdrant.scroll_all_points = AsyncMock(return_value=[
|
|
{"id": _chunk_id(44, i), "payload": {}} for i in range(4)
|
|
])
|
|
|
|
summary = await service.update_from_page(page_id=44, user=TENANT)
|
|
|
|
assert order == ["upsert", "delete"]
|
|
assert summary.chunks_deleted == 1 # only the stale 4th chunk
|
|
deleted = mock_qdrant.delete_by_ids.await_args.kwargs["point_ids"]
|
|
assert deleted == [_chunk_id(44, 3)]
|
|
|
|
async def test_partial_embedding_failure_marks_partial(
|
|
self, service, mock_qdrant, mock_ollama
|
|
):
|
|
mock_ollama.embed_batch = AsyncMock(
|
|
side_effect=lambda texts, **kw: [
|
|
[0.1] * 768 if i != 1 else None for i in range(len(texts))
|
|
]
|
|
)
|
|
|
|
summary = await service.update_from_page(page_id=44, user=TENANT)
|
|
|
|
assert summary.status == "partial"
|
|
assert summary.success is True
|
|
assert summary.chunks_created == 2
|
|
assert summary.chunks_skipped == 1
|
|
assert "failed" in (summary.error_message or "")
|
|
|
|
async def test_total_embedding_failure_keeps_old_vectors(
|
|
self, service, mock_qdrant, mock_ollama
|
|
):
|
|
mock_ollama.embed_batch = AsyncMock(
|
|
side_effect=lambda texts, **kw: [None for _ in texts]
|
|
)
|
|
mock_qdrant.scroll_all_points = AsyncMock(return_value=[
|
|
{"id": _chunk_id(44, 0), "payload": {}}
|
|
])
|
|
|
|
summary = await service.update_from_page(page_id=44, user=TENANT)
|
|
|
|
assert summary.status == "failed"
|
|
assert summary.success is False
|
|
# Old vectors are NOT wiped when nothing new was stored
|
|
mock_qdrant.upsert_points.assert_not_awaited()
|
|
mock_qdrant.delete_by_ids.assert_not_awaited()
|