"""Factory that returns a LangChain-compatible Embeddings object. This replaces the custom EmbeddingClient interface with the standard LangChain ``langchain_core.embeddings.Embeddings`` protocol, enabling seamless integration with LangChain vector stores, retrievers, and chains. Default priority (with ``prefer_local_embeddings=True``): 1. HuggingFace — ``langchain_huggingface.HuggingFaceEmbeddings`` (free, on-device). 2. OpenAI fallback only if HuggingFace import fails AND an API key is present. 3. Fake — deterministic ``FakeEmbeddings`` for unit tests only. Setting ``prefer_local_embeddings=False`` restores the legacy behaviour (OpenAI first if a key is configured, HuggingFace as fallback). """ import logging from langchain_core.embeddings import Embeddings from app.config import settings logger = logging.getLogger(__name__) _instance: Embeddings | None = None def _build_huggingface() -> Embeddings | None: """Try to construct a HuggingFace embeddings client, returning ``None`` on failure.""" try: from langchain_huggingface import HuggingFaceEmbeddings except ImportError: logger.warning( "langchain-huggingface not installed — cannot use local embeddings. " "Run: pip install langchain-huggingface sentence-transformers" ) return None logger.info( "Using LangChain HuggingFaceEmbeddings (model=%s, free/local)", settings.local_embedding_model, ) return HuggingFaceEmbeddings( model_name=settings.local_embedding_model, model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, ) def _build_openai() -> Embeddings | None: """Try to construct an OpenAI embeddings client, returning ``None`` on failure.""" if not settings.openai_api_key: return None try: from langchain_openai import OpenAIEmbeddings except ImportError: return None logger.info("Using LangChain OpenAIEmbeddings (model=%s)", settings.embedding_model) return OpenAIEmbeddings( model=settings.embedding_model, openai_api_key=settings.openai_api_key, ) def get_embedding_client() -> Embeddings: """Return a singleton LangChain :class:`Embeddings` instance. The client is constructed once and reused — critical for the HuggingFace backend which loads a ~90 MB model at construction time. Default order (``prefer_local_embeddings=True``): 1. :class:`langchain_huggingface.HuggingFaceEmbeddings` (free, local). 2. :class:`langchain_openai.OpenAIEmbeddings` (only if HF unavailable). 3. :class:`langchain_community.embeddings.FakeEmbeddings` as a test fallback. Set ``prefer_local_embeddings=False`` to invert the first two. Returns: A configured LangChain :class:`Embeddings` object. """ global _instance if _instance is not None: return _instance if getattr(settings, "prefer_local_embeddings", False): _instance = _build_huggingface() or _build_openai() else: _instance = _build_openai() or _build_huggingface() if _instance is None: logger.warning( "No embedding backend available — falling back to FakeEmbeddings (TEST ONLY)." ) from langchain_community.embeddings import FakeEmbeddings _instance = FakeEmbeddings(size=384) return _instance def reset_embedding_client() -> None: """Reset the cached singleton — useful for tests or hot config reloads.""" global _instance _instance = None