Spaces:
Running
Running
File size: 4,311 Bytes
0646b18 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
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
|