Coverage for src / lexigram / contracts / ai / vector.py: 0%
53 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-15 18:57 +0800
1"""Vector store and document protocols."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
8if TYPE_CHECKING:
9 from lexigram.contracts.core import HealthCheckResult
10 from lexigram.contracts.core.result import Result
11 from lexigram.contracts.data.vector.exceptions import VectorError
14@runtime_checkable
15class DocumentProtocol(Protocol):
16 """Structural protocol for unified document representation.
18 Structural protocol satisfied by any class with the required
19 fields — used across lexigram-vector and lexigram-ai-rag.
20 """
22 @property
23 def id(self) -> str | None:
24 """Unique document identifier, or None if unassigned."""
25 ...
27 @property
28 def text(self) -> str:
29 """Text content of the document."""
30 ...
32 @property
33 def metadata(self) -> dict[str, Any]:
34 """Arbitrary metadata key-value pairs."""
35 ...
38@dataclass(frozen=True)
39class EmbeddingResult:
40 """Result of an embedding operation."""
42 vectors: list[list[float]]
43 model: str
44 tokens: int
47@runtime_checkable
48class SearchResultProtocol(Protocol):
49 """Protocol for a vector search hit."""
51 @property
52 def text(self) -> str:
53 """The text content of the search result."""
54 ...
56 @property
57 def document(self) -> DocumentProtocol:
58 """The retrieved document."""
59 ...
61 @property
62 def score(self) -> float:
63 """Relevance score."""
64 ...
66 @property
67 def metadata(self) -> dict[str, Any]:
68 """Additional search-specific metadata."""
69 ...
72@runtime_checkable
73class BatchProcessorProtocol(Protocol):
74 """Protocol for batch operations on documents."""
76 async def process_batch(
77 self,
78 documents: list[DocumentProtocol],
79 **kwargs: Any,
80 ) -> Result[int, VectorError]:
81 """Process a batch of documents."""
82 ...
85@runtime_checkable
86class DocumentVectorStoreProtocol(Protocol):
87 """Document-centric vector store API for RAG and AI packages.
89 This is the AI-layer contract for add/search/delete against embedded
90 documents. For connection lifecycle and collection management, use
91 :class:`~lexigram.contracts.data.vector.protocols.VectorStoreProtocol`
92 (infrastructure layer).
94 All vector database drivers (Qdrant, ChromaDB, PGVector, etc.) are adapted
95 to this shape via ``lexigram-vector`` so callers can swap backends.
96 """
98 async def add(
99 self, documents: list[DocumentProtocol]
100 ) -> Result[list[str], VectorError]:
101 """Add documents to the vector store.
103 Args:
104 documents: List of documents with text and metadata.
106 Returns:
107 ``Ok(list[str])`` with document IDs on success, or
108 ``Err(VectorStoreError)`` on failure.
109 """
110 ...
112 async def batch_upsert(
113 self,
114 documents: list[DocumentProtocol],
115 batch_size: int = 100,
116 ) -> Result[int, VectorError]:
117 """Upsert documents in batches.
119 Args:
120 documents: List of documents to upsert.
121 batch_size: Number of documents per batch. defaults to 100.
123 Returns:
124 ``Ok(int)`` with the total count of documents upserted, or
125 ``Err(VectorStoreError)`` on failure.
126 """
127 ...
129 async def search(
130 self,
131 query: list[float],
132 *,
133 top_k: int = 10,
134 filters: dict[str, Any] | None = None,
135 score_threshold: float | None = None,
136 ) -> Result[list[SearchResultProtocol], VectorError]:
137 """Search for similar documents using a vector.
139 Args:
140 query: Query embedding vector.
141 top_k: Number of results to return.
142 filters: Metadata filters.
143 score_threshold: Minimum relevance score.
145 Returns:
146 ``Ok(list of search results)`` on success, or
147 ``Err(VectorStoreError)`` on failure.
148 """
149 ...
151 async def search_text(
152 self,
153 query: str,
154 *,
155 top_k: int = 10,
156 filters: dict[str, Any] | None = None,
157 ) -> Result[list[SearchResultProtocol], VectorError]:
158 """Search for similar documents using text.
160 Args:
161 query: Query text.
162 top_k: Number of results to return.
163 filters: Metadata filters.
165 Returns:
166 ``Ok(list of search results)`` on success, or
167 ``Err(VectorStoreError)`` on failure.
168 """
169 ...
171 async def delete(self, ids: list[str]) -> Result[int, VectorError]:
172 """Delete documents by ID.
174 Args:
175 ids: List of document IDs to delete.
177 Returns:
178 ``Ok(int)`` with the count of deleted documents on success, or
179 ``Err(VectorStoreError)`` on failure.
180 """
181 ...
183 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
184 """Perform a lightweight connectivity check.
186 Returns:
187 Structured health check result.
188 """
189 ...
192@runtime_checkable
193class ChunkerProtocol(Protocol):
194 """Protocol for document chunking strategies.
196 Implementations split document text into smaller, overlapping
197 or non-overlapping chunks suitable for embedding and retrieval.
198 """
200 def chunk(
201 self,
202 text: str,
203 metadata: dict[str, Any] | None = None,
204 ) -> list[Any]:
205 """Split text into chunks.
207 Args:
208 text: The document text to chunk.
209 metadata: Optional metadata to attach to each chunk.
211 Returns:
212 List of Chunk objects.
213 """
214 ...
217@dataclass(frozen=True)
218class Document:
219 """Concrete document data class shared across AI packages.
221 Satisfies ``DocumentProtocol``. Use this when constructing documents
222 in packages that must not depend on ``lexigram-vector``.
223 """
225 text: str
226 metadata: dict[str, Any] = field(default_factory=dict)
227 id: str | None = None
228 embedding: list[float] | None = None
231@dataclass(frozen=True)
232class RAGSearchResult:
233 """Enriched search result for RAG applications.
235 Unlike the generic SearchResult from the storage layer (data/vector/types.py),
236 this result is mutable and focused on RAG workflows:
237 - Wraps a full Document object (not just id/score/vector/content)
238 - Includes ranking information useful for re-ranking pipelines
239 - Mutable to allow post-processing
241 Satisfies ``SearchResultProtocol``.
242 """
244 document: Document
245 score: float
246 rank: int = 0
247 metadata: dict[str, Any] = field(default_factory=dict)
250__all__ = [
251 "BatchProcessorProtocol",
252 "ChunkerProtocol",
253 "Document",
254 "DocumentProtocol",
255 "DocumentVectorStoreProtocol",
256 "EmbeddingResult",
257 "RAGSearchResult",
258 "SearchResultProtocol",
259]