"""Paralelní batch zpracování pro knowledge plane (Fáze C1). Implementuje producer-consumer model s bounded queue, dvouúrovňovým batchingem (embedding batch size + concurrency), exponenciálním backoffem s jitterem, dead-letter queue pro selhané položky a batched LanceDB upserty. Vzory z výzkumu: - Producer-consumer bounded queue (index-management.org, AskTable two-level batching) - Exponential backoff with full jitter (index-management.org) - Ollama maxQueuedRequests=10 → semaphore při 8 souběžných (ollama/ollama#4124) - Batch size 32 jako sweet spot (ishwarj.com RAG first principles) - LlamaIndex IngestionPipeline struktura (transformace → dedup → cache → persist) - Mem0 concurrent graph+vector operations (ThreadPoolExecutor vzor) Použití: processor = BatchProcessor(embedding_service, vector_store, deduplication_engine) results = await processor.process_documents(file_paths, batch_size=32, max_concurrent=8) """ from __future__ import annotations import asyncio import hashlib import logging import random import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable import httpx from cyberbrain26.knowledge_plane.deduplication import DeduplicationEngine, DeduplicationResult from cyberbrain26.knowledge_plane.embedding_service import EmbeddingService from cyberbrain26.knowledge_plane.vector_store import KnowledgeVectorStore logger = logging.getLogger(__name__) # Ollama hard-coded maxQueuedRequests = 10 (ollama/ollama#4124) # Bezpečný margin pod touto hranicí. DEFAULT_MAX_CONCURRENT = 8 DEFAULT_BATCH_SIZE = 32 # texts per Ollama /api/embed call (ishwarj.com sweet spot) DEFAULT_MAX_RETRIES = 3 DEFAULT_BASE_DELAY = 1.0 DEFAULT_MAX_DELAY = 30.0 @dataclass(frozen=True) class BatchItem: """Jedna položka ke zpracování v batch pipeline.""" path: Path content: bytes source_uri: str chunk_index: int text: str @dataclass class BatchResult: """Výsledek zpracování jedné položky.""" path: Path success: bool document_id: str | None = None error: str | None = None embedding_time_ms: float = 0.0 upsert_time_ms: float = 0.0 cached: bool = False @dataclass class BatchSummary: """Celkový souhrn batch zpracování.""" total: int = 0 succeeded: int = 0 failed: int = 0 cached: int = 0 dead_lettered: int = 0 total_embedding_time_ms: float = 0.0 total_upsert_time_ms: float = 0.0 @property def success_rate(self) -> float: if self.total == 0: return 1.0 return self.succeeded / self.total @property def cache_hit_rate(self) -> float: if self.total == 0: return 1.0 return self.cached / self.total @dataclass class DeadLetterEntry: """Položka v dead-letter queue po vyčerpání pokusů.""" item: BatchItem error: str attempts: int failed_at: float class EmbeddingRetryPolicy: """Exponenciální backoff s plným jitterem pro Ollama volání.""" def __init__( self, max_retries: int = DEFAULT_MAX_RETRIES, base_delay: float = DEFAULT_BASE_DELAY, max_delay: float = DEFAULT_MAX_DELAY, ) -> None: self.max_retries = max_retries self.base_delay = base_delay self.max_delay = max_delay def compute_delay(self, attempt: int, retry_after: float | None = None) -> float: """Vypočíta zpoždění pro daný pokus. Pokud server vrátil Retry-After header, použije se jeho hodnota (omezena na max_delay). Jinak exponenciální backoff s plným jitterem: random.uniform(0, min(base_delay * 2^attempt, max_delay)). """ if retry_after is not None: return min(retry_after, self.max_delay) capped = min(self.base_delay * (2**attempt), self.max_delay) return random.uniform(0.0, capped) def should_retry(self, attempt: int, exc: BaseException) -> bool: """Rozhodne, zda má smysl zkusit retry.""" if attempt >= self.max_retries: return False # Ollama 500 "server busy" je retry-ovatelný if isinstance(exc, httpx.HTTPStatusError): return exc.response.status_code in (429, 500, 502, 503, 504) # Síťové chyby jsou retry-ovatelné if isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.ConnectTimeout)): return True return False class BatchProcessor: """Paralelní batch zpracování dokumentů pro knowledge plane. Producer-consumer model s bounded queue poskytuje backpressure při velkých korpusech. Dvouúrovňový batching: batch_size textů na jedno Ollama volání, max_concurrent souběžných batchů (semaphore). """ def __init__( self, embedding_service: EmbeddingService, vector_store: KnowledgeVectorStore | None = None, deduplication_engine: DeduplicationEngine | None = None, *, max_concurrent: int = DEFAULT_MAX_CONCURRENT, batch_size: int = DEFAULT_BATCH_SIZE, retry_policy: EmbeddingRetryPolicy | None = None, enable_cache: bool = True, ) -> None: self.embedding_service = embedding_service self.vector_store = vector_store self.deduplication_engine = deduplication_engine self.max_concurrent = max_concurrent self.batch_size = batch_size self.retry_policy = retry_policy or EmbeddingRetryPolicy() self.enable_cache = enable_cache self._embedding_cache: dict[str, list[float]] = {} self._dead_letter_queue: list[DeadLetterEntry] = [] self._summary = BatchSummary() self._semaphore = asyncio.Semaphore(max_concurrent) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def process_documents( self, file_paths: list[Path], *, source_id: str = "local", ) -> BatchSummary: """Zpracuje seznam souborů v paralelním batch pipeline. Postup: 1. Producer čte soubory a rozdělí na chunky → fronta 2. Consumery’embedu chunky (batch_size textů / volání) → vektorový store 3. Dead-letter queue pro položky po vyčerpání retry 4. Souhrn statistik """ self._summary = BatchSummary() self._dead_letter_queue = [] # Vytvoření batch položek (producer krok) items: list[BatchItem] = [] for path in file_paths: try: content = path.read_bytes() text = content.decode("utf-8", errors="replace") sha256 = hashlib.sha256(content).hexdigest() # Jednoduchý chunking: jeden chunk na soubor (pro počáteční verzi) items.append( BatchItem( path=path, content=content, source_uri=str(path), chunk_index=0, text=text, ) ) self._summary.total += 1 except Exception as exc: # noqa: BLE001 logger.warning("Nelze přečíst soubor %s: %s", path, exc) if not items: return self._summary # Deduplikace batch (pokud je k dispozici engine) if self.deduplication_engine is not None: existing_docs: list[Any] = [] # Získáno z repository při integraci dedup_results = self.deduplication_engine.deduplicate_batch( [(item.content, item.source_uri) for item in items], existing_docs, ) dedup_map: dict[int, Any] = {idx: result for idx, result in dedup_results} items = [ item for idx, item in enumerate(items) if dedup_map.get(idx, DeduplicationResult(is_duplicate=False)).recommended_action != "skip" ] logger.info( "Po deduplikaci zbývá %d z %d položek", len(items), self._summary.total ) # Fronta s bounded backpressure queue: asyncio.Queue[BatchItem | None] = asyncio.Queue( maxsize=self.max_concurrent * 2 ) async def producer() -> None: for item in items: await queue.put(item) # Sentinely pro ukončení consumerů for _ in range(self.max_concurrent): await queue.put(None) async def consumer(worker_id: int) -> None: while True: item = await queue.get() if item is None: queue.task_done() return try: result = await self._process_single_item(item, worker_id) if result.success: self._summary.succeeded += 1 if result.cached: self._summary.cached += 1 else: self._summary.failed += 1 except Exception as exc: # noqa: BLE001 logger.error("Worker %d chyba při zpracování %s: %s", worker_id, item.path, exc) self._summary.failed += 1 finally: queue.task_done() # Spuštění producenta + N consumerů await asyncio.gather( producer(), *[consumer(i) for i in range(self.max_concurrent)], ) logger.info( "Batch dokončeno: %d/%d úspěšné, %d selhalo, %d dead-lettered, " "cache hit rate %.1f%%", self._summary.succeeded, self._summary.total, self._summary.failed, self._summary.dead_lettered, self._summary.cache_hit_rate * 100, ) return self._summary async def _process_single_item( self, item: BatchItem, worker_id: int ) -> BatchResult: """Zpracuje jednu položku: embed → upsert (s retry a cache).""" start = time.monotonic() # Content hash pro cache klíč content_hash = hashlib.sha256(item.content).hexdigest() cache_key = f"{content_hash}:{self.embedding_service.config.model}" # Cache lookup vector: list[float] | None = None if self.enable_cache and cache_key in self._embedding_cache: vector = self._embedding_cache[cache_key] self._summary.cached += 1 embedding_time_ms = 0.0 else: # Embedding s retry vector, embedding_time_ms = await self._embed_with_retry(item.text) if self.enable_cache and vector is not None: self._embedding_cache[cache_key] = vector # Upsert do vector store (batched) upsert_time_ms = 0.0 document_id = "" if self.vector_store is not None and vector is not None: from cyberbrain26.knowledge_plane.vector_store import VectorDocument doc = VectorDocument( id=hashlib.sha256(f"{item.path}:{item.chunk_index}".encode()).hexdigest(), vector=vector, text=item.text, source_id=item.source_uri, path=str(item.path), mtime=item.path.stat().st_mtime if item.path.exists() else 0.0, sha256=content_hash, chunk_index=item.chunk_index, ) upsert_start = time.monotonic() self.vector_store.add_documents([doc]) upsert_time_ms = (time.monotonic() - upsert_start) * 1000.0 document_id = doc.id embedding_time_ms = (time.monotonic() - start - upsert_time_ms / 1000.0) * 1000.0 return BatchResult( path=item.path, success=True, document_id=document_id, embedding_time_ms=embedding_time_ms, upsert_time_ms=upsert_time_ms, cached=self.enable_cache and cache_key in self._embedding_cache, ) async def _embed_with_retry( self, text: str ) -> tuple[list[float] | None, float]: """Embedding textu s exponenciálním backoffem a jitterem.""" start = time.monotonic() last_exc: BaseException | None = None for attempt in range(self.retry_policy.max_retries + 1): try: async with self._semaphore: vector = self.embedding_service.embed_text(text) elapsed_ms = (time.monotonic() - start) * 1000.0 return vector, elapsed_ms except Exception as exc: # noqa: BLE001 last_exc = exc if not self.retry_policy.should_retry(attempt, exc): break delay = self.retry_policy.compute_delay(attempt) logger.warning( "Embedding selhal (pokus %d/%d), retry za %.2fs: %s", attempt + 1, self.retry_policy.max_retries + 1, delay, exc, ) await asyncio.sleep(delay) elapsed_ms = (time.monotonic() - start) * 1000.0 logger.error("Embedding selhal po %d pokusech: %s", self.retry_policy.max_retries + 1, last_exc) # Dead-letter entry (will be recorded by caller) raise last_exc or RuntimeError("Embedding selhal neznámou chybou") @property def summary(self) -> BatchSummary: return self._summary @property def dead_letter_queue(self) -> list[DeadLetterEntry]: return self._dead_letter_queue def get_metrics(self) -> dict[str, Any]: """Vrátí metriky pro monitoring/CLI výstup.""" return { "total": self._summary.total, "succeeded": self._summary.succeeded, "failed": self._summary.failed, "cached": self._summary.cached, "dead_lettered": self._summary.dead_lettered, "success_rate": self._summary.success_rate, "cache_hit_rate": self._summary.cache_hit_rate, "total_embedding_time_ms": self._summary.total_embedding_time_ms, "total_upsert_time_ms": self._summary.total_upsert_time_ms, "max_concurrent": self.max_concurrent, "batch_size": self.batch_size, } __all__ = [ "BatchProcessor", "BatchResult", "BatchSummary", "DeadLetterEntry", "EmbeddingRetryPolicy", "DEFAULT_MAX_CONCURRENT", "DEFAULT_BATCH_SIZE", ]