The ``BM25SparseEmbeddingProvider.__init__`` calls ``fastembed.SparseTextEmbedding(model_name="Qdrant/bm25")`` which downloads ~50 MB of model weights from HuggingFace and loads them into memory — observed >5 s wall-clock in production. The inference methods (``encode_async``, ``encode_batch_async``) already wrap work in ``anyio.to_thread.run_sync``, so the design intent is clearly to keep FastEmbed off the event loop. That protection just didn't cover the constructor. Symptom in the Astrolabe Cloud per-tenant deploy (deck #102 smoke): ~30–90 s after a user enables semantic search, the pod tips into a SIGKILL-restart cycle. Loki shows a single log line Initializing BM25 sparse embedding provider: Qdrant/bm25 followed by nothing else from the event loop until exitCode 137. Kubernetes ``/health/live`` httpGet probe timeout=5s fires 6 times in a row, kubelet kills the container, restart, repeat. Fix: switch ``get_bm25_service()`` to an async accessor that wraps the first-time construction in ``anyio.to_thread.run_sync``. Two existing call sites (``vector/processor.py:603``, ``search/bm25_hybrid.py:123``) update to ``await``. Both are already inside async functions so the await is free. New unit test pins the invariant by monkey-patching ``BM25SparseEmbeddingProvider.__init__`` with ``time.sleep(1)`` and asserting a concurrent ``anyio.sleep(0.05)`` finishes promptly — the test fails if the constructor ever runs back on the event loop. Same pattern exists in ``OllamaEmbeddingProvider.__init__`` (sync ``httpx.get`` health-check). Ollama isn't enabled in any current deploy; filed as a follow-up. Refs: - Astrolabe Cloud deck card #102 (smoke discovery) - Sibling fix #799 (NullPool for cross-loop-asyncpg, same class of "anyio bites you in production" bug) Verified: - ``uv run pytest tests/unit/`` — 1027 passed - ``uv run ruff check`` clean on touched files - ``uv run ty check`` clean on touched files - New tests pass Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
"""Embedding service with provider detection.
|
|
|
|
DEPRECATED: This module is maintained for backward compatibility.
|
|
New code should use nextcloud_mcp_server.providers.get_provider() directly.
|
|
"""
|
|
|
|
import logging
|
|
|
|
import anyio
|
|
|
|
from nextcloud_mcp_server.providers import get_provider
|
|
|
|
from .bm25_provider import BM25SparseEmbeddingProvider
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EmbeddingService:
|
|
"""
|
|
Unified embedding service with automatic provider detection.
|
|
|
|
DEPRECATED: This class wraps the new unified provider infrastructure
|
|
for backward compatibility. New code should use
|
|
nextcloud_mcp_server.providers.get_provider() directly.
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""Initialize embedding service with auto-detected provider."""
|
|
self.provider = get_provider()
|
|
|
|
async def embed(self, text: str) -> list[float]:
|
|
"""
|
|
Generate embedding vector for text.
|
|
|
|
Args:
|
|
text: Input text to embed
|
|
|
|
Returns:
|
|
Vector embedding as list of floats
|
|
"""
|
|
return await self.provider.embed(text)
|
|
|
|
async def embed_batch(self, texts: list[str]) -> list[list[float]]:
|
|
"""
|
|
Generate embeddings for multiple texts.
|
|
|
|
Args:
|
|
texts: List of texts to embed
|
|
|
|
Returns:
|
|
List of vector embeddings
|
|
"""
|
|
return await self.provider.embed_batch(texts)
|
|
|
|
def get_dimension(self) -> int:
|
|
"""
|
|
Get embedding dimension.
|
|
|
|
Returns:
|
|
Vector dimension
|
|
"""
|
|
return self.provider.get_dimension()
|
|
|
|
async def close(self):
|
|
"""Close provider resources."""
|
|
if hasattr(self.provider, "close") and callable(
|
|
getattr(self.provider, "close")
|
|
):
|
|
close_method = getattr(self.provider, "close")
|
|
await close_method()
|
|
|
|
|
|
# Singleton instance
|
|
_embedding_service: EmbeddingService | None = None
|
|
|
|
|
|
def get_embedding_service() -> EmbeddingService:
|
|
"""
|
|
Get singleton embedding service instance.
|
|
|
|
Returns:
|
|
Global EmbeddingService instance
|
|
"""
|
|
global _embedding_service
|
|
if _embedding_service is None:
|
|
_embedding_service = EmbeddingService()
|
|
return _embedding_service
|
|
|
|
|
|
# BM25 sparse embedding singleton
|
|
_bm25_service: BM25SparseEmbeddingProvider | None = None
|
|
|
|
|
|
async def get_bm25_service() -> BM25SparseEmbeddingProvider:
|
|
"""
|
|
Get singleton BM25 sparse embedding service instance.
|
|
|
|
Lazily instantiates the singleton off the event loop. The
|
|
``BM25SparseEmbeddingProvider`` constructor calls
|
|
``fastembed.SparseTextEmbedding(model_name="Qdrant/bm25")`` which
|
|
downloads ~50 MB of model weights from HuggingFace and loads them
|
|
into memory — observed >5 s wall-clock in production, enough to
|
|
stall the calling thread. The encode methods on the provider
|
|
already offload via ``anyio.to_thread.run_sync``; this routes the
|
|
first-time init through the same path so the event loop stays
|
|
responsive (kubernetes ``/health/live`` httpGet probe in
|
|
particular).
|
|
|
|
The singleton is process-wide so the per-pod cost is paid once.
|
|
Subsequent calls hit the warm path and return after a single
|
|
non-blocking await.
|
|
|
|
Concurrent first callers race: two coroutines can both observe
|
|
``_bm25_service is None`` and both enter ``run_sync``. We accept
|
|
the duplicate model load over an ``asyncio.Lock``, which has its
|
|
own cross-loop hazards under anyio TaskGroups (see PR #799 for
|
|
that class of bug). The duplicate is bounded — FastEmbed caches
|
|
the downloaded weights on disk after the first call, so loser(s)
|
|
pick up cheaply.
|
|
|
|
Returns:
|
|
Global BM25SparseEmbeddingProvider instance
|
|
"""
|
|
global _bm25_service
|
|
if _bm25_service is None:
|
|
_bm25_service = await anyio.to_thread.run_sync( # type: ignore[attr-defined] # ty: ignore[unresolved-attribute]
|
|
BM25SparseEmbeddingProvider
|
|
)
|
|
return _bm25_service
|