Indexes documents end-to-end: chunk -> embed (batched) -> upsert.
Continues past per-batch failures (logged + counted in errors).
Yields IndexingProgress after each batch so callers can stream UI updates.
Failure model: an embedder or store exception on a single batch does NOT
abort the run. The failing batch's chunks are dropped (not retried inside
the pipeline), the error counter increments, and the next batch proceeds.
Callers that need stricter "all or nothing" semantics should check
final.errors == 0 and retry the document set themselves.
Source code in src/cenote/pipeline/indexing.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98 | class IndexingPipeline:
"""Indexes documents end-to-end: chunk -> embed (batched) -> upsert.
Continues past per-batch failures (logged + counted in `errors`).
Yields IndexingProgress after each batch so callers can stream UI updates.
Failure model: an embedder or store exception on a single batch does NOT
abort the run. The failing batch's chunks are dropped (not retried inside
the pipeline), the error counter increments, and the next batch proceeds.
Callers that need stricter "all or nothing" semantics should check
`final.errors == 0` and retry the document set themselves.
"""
def __init__(
self,
chunker: Chunker,
embedder: Embedder,
store: VectorStore,
*,
namespace: str,
batch_size: int = 32,
) -> None:
if batch_size <= 0:
raise ConfigurationError("batch_size must be positive")
self._chunker = chunker
self._embedder = embedder
self._store = store
self._namespace = namespace
self._batch_size = batch_size
async def index(self, documents: Iterable[Document]) -> AsyncIterator[IndexingProgress]:
"""Index documents, yielding progress after each batch.
Empty document iterables yield no progress events. Documents that
produce zero chunks (e.g., empty content) still count toward
`documents_done` but contribute no chunks.
"""
docs_done = 0
chunks_done = 0
errors = 0
pending: list[Chunk] = []
for doc in documents:
chunks = self._chunker.chunk(doc)
docs_done += 1
pending.extend(chunks)
while len(pending) >= self._batch_size:
batch = pending[: self._batch_size]
pending = pending[self._batch_size :]
try:
embedded = await self._embedder.embed(batch)
await self._store.upsert(embedded, namespace=self._namespace)
chunks_done += len(batch)
except Exception as exc:
logger.warning("IndexingPipeline batch failed: %s", exc)
errors += 1
yield IndexingProgress(docs_done, chunks_done, errors)
if pending:
try:
embedded = await self._embedder.embed(pending)
await self._store.upsert(embedded, namespace=self._namespace)
chunks_done += len(pending)
except Exception as exc:
logger.warning("IndexingPipeline final batch failed: %s", exc)
errors += 1
yield IndexingProgress(docs_done, chunks_done, errors)
|
index(documents: Iterable[Document]) -> AsyncIterator[IndexingProgress]
async
Index documents, yielding progress after each batch.
Empty document iterables yield no progress events. Documents that
produce zero chunks (e.g., empty content) still count toward
documents_done but contribute no chunks.
Source code in src/cenote/pipeline/indexing.py
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98 | async def index(self, documents: Iterable[Document]) -> AsyncIterator[IndexingProgress]:
"""Index documents, yielding progress after each batch.
Empty document iterables yield no progress events. Documents that
produce zero chunks (e.g., empty content) still count toward
`documents_done` but contribute no chunks.
"""
docs_done = 0
chunks_done = 0
errors = 0
pending: list[Chunk] = []
for doc in documents:
chunks = self._chunker.chunk(doc)
docs_done += 1
pending.extend(chunks)
while len(pending) >= self._batch_size:
batch = pending[: self._batch_size]
pending = pending[self._batch_size :]
try:
embedded = await self._embedder.embed(batch)
await self._store.upsert(embedded, namespace=self._namespace)
chunks_done += len(batch)
except Exception as exc:
logger.warning("IndexingPipeline batch failed: %s", exc)
errors += 1
yield IndexingProgress(docs_done, chunks_done, errors)
if pending:
try:
embedded = await self._embedder.embed(pending)
await self._store.upsert(embedded, namespace=self._namespace)
chunks_done += len(pending)
except Exception as exc:
logger.warning("IndexingPipeline final batch failed: %s", exc)
errors += 1
yield IndexingProgress(docs_done, chunks_done, errors)
|