Class EmbeddingService

skills/embedding/scripts/embed.py:234–556

Configurable embedding backend with caching and safe fallbacks.

class EmbeddingService:
    """Configurable embedding backend with caching and safe fallbacks."""

    def __init__(
        self,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        dimensions: Optional[int] = None,
        cache_dir: Optional[Path] = None,
    ):
        requested_provider = (
            provider
            or _get_first_env_value("PAPERFLOW_EMBED_PROVIDER", "EMBEDDING_PROVIDER")
            or "hash"
        ).strip().lower()
        self.dimensions = int(
            dimensions
            or _get_first_env_value("PAPERFLOW_EMBED_DIMENSIONS", "EMBEDDING_DIMENSIONS")
            or "768"
        )
        self.cache_dir = Path(cache_dir or DEFAULT_CACHE_DIR)
        self.cache_dir.mkdir(parents=True, exist_ok=True)

        self.client = None
        self.local_model = None
        self.model_source = None

        if requested_provider in {"openai", "dashscope", "aliyun", "bailian"}:
            self.provider = "openai"
            provider_hint = requested_provider
            self.model = model or _get_openai_embedding_model(provider_hint)
            api_key = _get_openai_api_key(provider_hint)
            if OPENAI_AVAILABLE and not _is_placeholder_openai_key(api_key):
                self.client = OpenAI(
                    api_key=api_key,
                    base_url=_get_openai_base_url(provider_hint),
                    timeout=_get_openai_timeout(provider_hint),
                )
            else:
                self.provider = "hash"
                self.model = "hash"
        elif requested_provider in {"nscale_api", "nscale"}:
            self.provider = "nscale_api"
            self.model = model or os.environ.get("NSCALE_EMBEDDING_MODEL") or os.environ.get("HF_EMBEDDING_MODEL") or "Qwen3-Embedding-8B"
            api_key = (
                os.environ.get("NSCALE_API_KEY")
                or os.environ.get("NSCALE_SERVICE_TOKEN")
                or ""
            )
            if _is_placeholder_nscale_key(api_key):
                api_key = ""
            base_url = (
                os.environ.get("NSCALE_BASE_URL")
                or "https://aiproxy.infaas-amd-dev.glo1.nscale.com"
            ).strip().rstrip("/")
            timeout = float(os.environ.get("NSCALE_API_TIMEOUT", os.environ.get("HF_API_TIMEOUT", "60")))
            if api_key:
                self.client = {