58ca6109d1
Why: - ADR-0001's Qdrant payload records embedding_model_version so a future model swap can identify which chunks need re-embedding. The embedder is what knows which model produced its vectors, so it reports this rather than the call site reconstructing it from settings. Changes: - DenseEmbedder/SparseEmbedder protocols gain a model_version: str attribute. - OpenAICompatibleEmbedder reports its configured model; Bm25SparseEmbedder reports its analyzer (bm25-<analyzer>).
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""Hand-written fakes for narrow application-owned ports (ADR-0016)."""
|
|
|
|
import asyncio
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass, field
|
|
|
|
from src.application.ingestion.models import SparseVector
|
|
|
|
|
|
@dataclass
|
|
class FakeObjectStorage:
|
|
"""In-memory `ObjectStorage`. `fail_next` simulates one upload failure."""
|
|
|
|
objects: dict[str, bytes] = field(default_factory=dict)
|
|
fail_next: bool = False
|
|
|
|
async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise OSError("simulated object storage failure")
|
|
self.objects[key] = data
|
|
|
|
|
|
@dataclass
|
|
class FakeDenseEmbedder:
|
|
"""A scripted `DenseEmbedder`. Returns a fixed-dimension zero vector per
|
|
text by default; `fail_next` simulates one batch failure.
|
|
"""
|
|
|
|
name: str
|
|
dimensions: int = 4
|
|
model_version: str = "fake-dense-v1"
|
|
calls: list[list[str]] = field(default_factory=list)
|
|
fail_next: bool = False
|
|
delay_seconds: float = 0.0
|
|
"""Simulates a slow provider call, e.g. to exercise timeout handling."""
|
|
|
|
async def embed_batch(self, texts: Sequence[str]) -> list[list[float]]:
|
|
self.calls.append(list(texts))
|
|
if self.delay_seconds:
|
|
await asyncio.sleep(self.delay_seconds)
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise RuntimeError("simulated embedder failure")
|
|
return [[0.0] * self.dimensions for _ in texts]
|
|
|
|
|
|
@dataclass
|
|
class FakeSparseEmbedder:
|
|
"""A scripted `SparseEmbedder`. Returns an empty sparse vector per text."""
|
|
|
|
name: str = "sparse"
|
|
model_version: str = "fake-sparse-v1"
|
|
calls: list[list[str]] = field(default_factory=list)
|
|
fail_next: bool = False
|
|
|
|
def embed_batch(self, texts: Sequence[str], *, query: bool = False) -> list[SparseVector]:
|
|
self.calls.append(list(texts))
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise RuntimeError("simulated embedder failure")
|
|
return [SparseVector(indices=[], values=[]) for _ in texts]
|
|
|