config.py 2.18 KB
"""Embedding service compatibility config derived from unified app config."""

from __future__ import annotations

import os
from typing import Optional

from config.loader import get_app_config


class EmbeddingConfig(object):
    def __init__(self) -> None:
        app_config = get_app_config()
        runtime = app_config.runtime
        services = app_config.services.embedding
        text_backend = services.get_backend_config()
        image_backend = services.get_image_backend_config()

        self.HOST = runtime.embedding_host
        self.PORT = runtime.embedding_port

        self.TEXT_MODEL_ID = str(text_backend.get("model_id") or "Qwen/Qwen3-Embedding-0.6B")
        self.TEXT_MODEL_DIR = self.TEXT_MODEL_ID
        self.TEXT_DEVICE = str(text_backend.get("device") or "cuda")
        self.TEXT_BATCH_SIZE = int(text_backend.get("batch_size", 32))
        self.TEXT_NORMALIZE_EMBEDDINGS = bool(text_backend.get("normalize_embeddings", True))
        self.TEI_BASE_URL = str(text_backend.get("base_url") or "http://127.0.0.1:8080")
        self.TEI_TIMEOUT_SEC = int(text_backend.get("timeout_sec", 60))
        self.TEI_MAX_CLIENT_BATCH_SIZE = int(
            os.getenv("TEI_MAX_CLIENT_BATCH_SIZE")
            or text_backend.get("max_client_batch_size")
            or 24
        )

        self.USE_CLIP_AS_SERVICE = services.image_backend == "clip_as_service"
        self.CLIP_AS_SERVICE_SERVER = str(image_backend.get("server") or "grpc://127.0.0.1:51000")
        self.CLIP_AS_SERVICE_MODEL_NAME = str(image_backend.get("model_name") or "CN-CLIP/ViT-H-14")

        self.IMAGE_MODEL_NAME = str(image_backend.get("model_name") or "ViT-H-14")
        # Redis multimodal cache keys (image + clip_text) include this string; change model → new key space.
        self.MULTIMODAL_MODEL_NAME = str(
            image_backend.get("model_name")
            or ("CN-CLIP/ViT-H-14" if self.USE_CLIP_AS_SERVICE else "ViT-H-14")
        )
        self.IMAGE_DEVICE = image_backend.get("device")  # type: Optional[str]
        self.IMAGE_BATCH_SIZE = int(image_backend.get("batch_size", 8))
        self.IMAGE_NORMALIZE_EMBEDDINGS = bool(image_backend.get("normalize_embeddings", True))


CONFIG = EmbeddingConfig()