Spaces:
Running
Running
| from typing import Any | |
| from cuga.config import settings, DBS_DIR | |
| from cuga.backend.llm.models import LLMManager | |
| import os | |
| from dynaconf import Dynaconf # type: ignore | |
| llm_manager = LLMManager() | |
| def _sanitize_vector_store_config(raw: dict) -> dict: | |
| cfg = dict(raw or {}) | |
| # Coerce embedding_model_dims to int when possible | |
| dims = cfg.get("embedding_model_dims") | |
| if isinstance(dims, str) and dims.isdigit(): | |
| cfg["embedding_model_dims"] = int(dims) | |
| return cfg | |
| def get_config(namespace_id: str | None = None): | |
| # Use cached Dynaconf instance to avoid duplication and key normalization issues | |
| # settings = _get_settings() | |
| # The memory.toml groups relevant settings under [mem0_config] | |
| root = dict[Any, Any](settings.get("mem0_config", {}) or {}) | |
| # Defensive deep copies and sanitization of nested configs | |
| vs = root.get("vector_store", {}) | |
| vs_cfg = _sanitize_vector_store_config((vs or {}).get("config", {})) | |
| # Default milvus url/path under DBS_DIR if not provided | |
| default_milvus = os.path.join(DBS_DIR, "milvus.db") | |
| vs_url = os.environ.get('WXO_MILVUS_URI', vs_cfg.get("url", default_milvus)) | |
| # If the configured url is a relative filesystem path, resolve it under DBS_DIR | |
| if isinstance(vs_url, str) and not (vs_url.startswith("http://") or vs_url.startswith("https://")): | |
| if not os.path.isabs(vs_url): | |
| vs_url = os.path.join(DBS_DIR, vs_url) | |
| # Ensure the parent directory exists for Milvus-lite local storage | |
| parent_dir = os.path.dirname(vs_url) | |
| if parent_dir: | |
| os.makedirs(parent_dir, exist_ok=True) | |
| vs_cfg["url"] = vs_url | |
| vs_cfg["token"] = ( | |
| f'{os.environ.get("MILVUS_USERNAME", "root")}:{os.environ.get("MILVUS_PASSWORD", "Milvus")}' | |
| ) | |
| # Prefer explicit namespace_id, then env var, then file/default | |
| vs_cfg["collection_name"] = namespace_id or os.environ.get( | |
| "MILVUS_COLLECTION", vs_cfg.get("collection_name", "memory") | |
| ) | |
| cfg: dict = {} | |
| cfg["vector_store"] = { | |
| "provider": (vs or {}).get("provider", "milvus"), | |
| "config": vs_cfg, | |
| } | |
| llm_model = llm_manager.get_model(settings.memory.mem0.model) | |
| # Keep llm config as-is (mem0 handles it), but ensure structure | |
| cfg["llm"] = { | |
| "provider": settings.memory.mem0.model.platform, | |
| "config": { | |
| "model": llm_model.model_name, | |
| "temperature": llm_model.temperature, | |
| "openai_base_url": llm_model.openai_api_base, | |
| "max_tokens": llm_model.max_tokens, | |
| }, | |
| } | |
| emb = root.get("embedder", {}) | |
| emb_cfg = (emb or {}).get("config", {}) | |
| cfg["embedder"] = { | |
| "provider": (emb or {}).get("provider", "huggingface"), | |
| "config": emb_cfg, | |
| } | |
| # Top-level defaults from [mem0_config] | |
| # Default history db under DBS_DIR if not provided; resolve relative paths under DBS_DIR | |
| history_path = root.get("history_db_path", os.path.join(DBS_DIR, "mem0_history.db")) | |
| if isinstance(history_path, str) and not os.path.isabs(history_path): | |
| history_path = os.path.join(DBS_DIR, history_path) | |
| cfg["history_db_path"] = history_path | |
| cfg["version"] = root.get("version", "v1.1") | |
| return cfg | |
| def get_milvus_config(): | |
| milvus_config = settings.milvus_config | |
| milvus_config_dict = milvus_config.to_dict() | |
| milvus_config_dict["step_processing"] = settings.memory.milvus.step_processing.model | |
| milvus_config_dict["fact_extraction"] = settings.memory.milvus.fact_extraction.model | |
| return Dynaconf(settings_files=[], environments=False, milvus_config=milvus_config_dict) | |
| def get_tips_extractor_config(): | |
| tips_extractor_config = settings.tips_extractor_config | |
| tips_extractor_config_dict = tips_extractor_config.to_dict() | |
| tips_extractor_config_dict.update(settings.memory.tips_extractor.model.to_dict()) | |
| return Dynaconf(settings_files=[], environments=False, tips_extractor_config=tips_extractor_config_dict) | |
| # Backward compatible module-level variables | |
| config = get_config() | |
| milvus_config = ( | |
| get_milvus_config() | |
| if settings.features.memory_provider == "milvus" | |
| else Dynaconf(settings_files=[], environments=False, milvus_config={}) | |
| ) | |
| tips_extractor_config = get_tips_extractor_config().tips_extractor_config | |