Spaces:
Running
Running
| """ | |
| File: /config.py | |
| Created Date: Sunday October 19th 2025 | |
| Author: Christian Nonis <alch.infoemail@gmail.com> | |
| ----- | |
| Last Modified: Sunday March 29th 2026 3:27:28 am | |
| Modified By: Christian Nonis <alch.infoemail@gmail.com> | |
| ----- | |
| """ | |
| # pylint: disable=too-few-public-methods | |
| import logging | |
| import os | |
| from typing import Literal | |
| import dotenv | |
| from pathlib import Path | |
| _project_root = Path(__file__).resolve().parent.parent | |
| _env_name = os.getenv("ENV") | |
| _env_path = _project_root / f".env{'.' + _env_name if _env_name else ''}" | |
| dotenv.load_dotenv(dotenv_path=_env_path) | |
| from src.services.redaction import install_log_redaction | |
| install_log_redaction() | |
| logger = logging.getLogger(__name__) | |
| def _normalize_openai_base_url(raw: str | None) -> str | None: | |
| if not raw or not raw.strip(): | |
| return None | |
| normalized = raw.strip().rstrip("/") | |
| if normalized in ("https://api.openai.com/v1", "https://api.openai.com"): | |
| return None | |
| return normalized | |
| def validate_pipeline_mode(mode: str | None) -> Literal["lightweight", "accurate"]: | |
| if mode not in ("lightweight", "accurate"): | |
| raise ValueError( | |
| f"Invalid PIPELINE_MODE: {mode!r}. Expected 'lightweight' or 'accurate'." | |
| ) | |
| return mode | |
| def validate_ingest_architect_mode( | |
| mode: str | None, | |
| ) -> Literal["tooler", "schema", "batch"]: | |
| normalized = (mode or "batch").strip().lower() | |
| aliases = { | |
| "tooler": "tooler", | |
| "current": "tooler", | |
| "schema": "schema", | |
| "batch": "batch", | |
| } | |
| if normalized not in aliases: | |
| raise ValueError( | |
| f"Invalid INGEST_ARCHITECT_MODE: {mode!r}. " | |
| "Expected 'tooler', 'schema', or 'batch'." | |
| ) | |
| return aliases[normalized] # type: ignore[return-value] | |
| def validate_ingest_architect_prior_context( | |
| mode: str | None, | |
| ) -> Literal["auto", "scratchpad", "raw"]: | |
| normalized = (mode or "auto").strip().lower() | |
| if normalized in ("auto", "scratchpad", "raw"): | |
| return normalized # type: ignore[return-value] | |
| raise ValueError( | |
| f"Invalid INGEST_ARCHITECT_PRIOR_CONTEXT: {mode!r}. " | |
| "Expected 'auto', 'scratchpad', or 'raw'." | |
| ) | |
| def validate_search_fusion(value: str | None) -> Literal["rrf", "cc"]: | |
| normalized = (value or "rrf").strip().lower() | |
| if normalized in ("rrf", "cc"): | |
| return normalized # type: ignore[return-value] | |
| raise ValueError( | |
| f"Invalid SEARCH_FUSION: {value!r}. Expected 'rrf' or 'cc'." | |
| ) | |
| def validate_context_passage_mode( | |
| value: str | None, | |
| ) -> Literal["hybrid", "bm25", "dense", "ilike"]: | |
| normalized = (value or "hybrid").strip().lower() | |
| if normalized in ("hybrid", "bm25", "dense", "ilike"): | |
| return normalized # type: ignore[return-value] | |
| raise ValueError( | |
| f"Invalid CONTEXT_PASSAGE_MODE: {value!r}. " | |
| "Expected 'hybrid', 'bm25', 'dense', or 'ilike'." | |
| ) | |
| def validate_search_config( | |
| *, | |
| enabled: bool, | |
| use_dense: bool, | |
| use_bm25: bool, | |
| data_db: str, | |
| bm25_k1: float, | |
| bm25_b: float, | |
| ) -> None: | |
| if bm25_k1 <= 0: | |
| raise ValueError(f"Invalid SEARCH_BM25_K1: {bm25_k1!r}. Must be > 0.") | |
| if not 0.0 <= bm25_b <= 1.0: | |
| raise ValueError( | |
| f"Invalid SEARCH_BM25_B: {bm25_b!r}. Must be between 0 and 1." | |
| ) | |
| if not enabled: | |
| return | |
| if use_bm25 and (data_db or "").strip().lower() != "postgresql": | |
| raise ValueError( | |
| "SEARCH_USE_BM25=true requires DATA_DB=postgresql " | |
| "(BM25 indexes are Postgres-only)." | |
| ) | |
| if not use_dense and not use_bm25: | |
| raise ValueError( | |
| "SEARCH_ENABLED=true requires SEARCH_USE_DENSE=true " | |
| "and/or SEARCH_USE_BM25=true." | |
| ) | |
| class AzureConfig: | |
| """ | |
| Configuration class for the Azure configuration. | |
| """ | |
| def __init__(self): | |
| self.small_llm_model = os.getenv("AZURE_SMALL_LLM_MODEL", "gpt-4o-mini") | |
| self.large_llm_model = os.getenv("AZURE_LARGE_LLM_MODEL") | |
| self.llm_api_version = os.getenv("AZURE_LLM_API_VERSION") or os.getenv( | |
| "AZURE_LARGE_LLM_API_VERSION" | |
| ) | |
| self.llm_endpoint = os.getenv("AZURE_LLM_ENDPOINT") or os.getenv( | |
| "AZURE_LARGE_LLM_ENDPOINT" | |
| ) | |
| self.llm_subscription_key = os.getenv("AZURE_LLM_SUBSCRIPTION_KEY") or os.getenv( | |
| "AZURE_LARGE_LLM_SUBSCRIPTION_KEY" | |
| ) | |
| self.embedding_model = os.getenv("AZURE_EMBEDDING_MODEL", "text-embedding-3-large") | |
| self.embedding_full_endpoint = os.getenv("AZURE_EMBEDDING_FULL_ENDPOINT") | |
| self.embedding_key = os.getenv("AZURE_EMBEDDING_KEY") | |
| def validate_llm(self): | |
| if [ | |
| self.small_llm_model, | |
| self.large_llm_model, | |
| self.llm_api_version, | |
| self.llm_endpoint, | |
| self.llm_subscription_key, | |
| ].count(None) > 0: | |
| raise ValueError("Azure LLM configuration is not complete") | |
| def validate_embeddings(self): | |
| if [self.embedding_full_endpoint, self.embedding_key].count(None) > 0: | |
| raise ValueError("Azure embeddings configuration is not complete") | |
| class OpenAIConfig: | |
| def __init__(self): | |
| self.api_key = os.getenv("OPENAI_API_KEY") | |
| self.base_url = _normalize_openai_base_url(os.getenv("OPENAI_BASE_URL")) | |
| self.small_llm_model = os.getenv("OPENAI_SMALL_LLM_MODEL", "gpt-4o-mini") | |
| self.large_llm_model = os.getenv("OPENAI_LARGE_LLM_MODEL", "gpt-4o") | |
| self.embedding_model = os.getenv( | |
| "OPENAI_EMBEDDING_MODEL", "text-embedding-3-large" | |
| ) | |
| def validate_llm(self, require_large: bool): | |
| if not self.api_key: | |
| raise ValueError("OpenAI API key is not set") | |
| if not self.small_llm_model: | |
| raise ValueError("OpenAI small LLM model is not set") | |
| if require_large and not self.large_llm_model: | |
| raise ValueError("OpenAI large LLM model is not set") | |
| def validate_embeddings(self): | |
| if not self.api_key: | |
| raise ValueError("OpenAI API key is not set") | |
| if not self.embedding_model: | |
| raise ValueError("OpenAI embedding model is not set") | |
| class AnthropicConfig: | |
| def __init__(self): | |
| self.api_key = os.getenv("ANTHROPIC_API_KEY") | |
| self.small_llm_model = os.getenv( | |
| "ANTHROPIC_SMALL_LLM_MODEL", "claude-3-5-haiku-latest" | |
| ) | |
| self.large_llm_model = os.getenv( | |
| "ANTHROPIC_LARGE_LLM_MODEL", "claude-sonnet-4-20250514" | |
| ) | |
| def validate_llm(self, require_large: bool): | |
| if not self.api_key: | |
| raise ValueError("Anthropic API key is not set") | |
| if not self.small_llm_model: | |
| raise ValueError("Anthropic small LLM model is not set") | |
| if require_large and not self.large_llm_model: | |
| raise ValueError("Anthropic large LLM model is not set") | |
| class DeepSeekConfig: | |
| def __init__(self): | |
| self.api_key = os.getenv("DEEPSEEK_API_KEY") | |
| self.base_url = "https://api.deepseek.com" | |
| self.small_llm_model = os.getenv( | |
| "DEEPSEEK_SMALL_LLM_MODEL", "deepseek-v4-flash" | |
| ) | |
| self.large_llm_model = os.getenv( | |
| "DEEPSEEK_LARGE_LLM_MODEL", "deepseek-v4-pro" | |
| ) | |
| def validate_llm(self, require_large: bool): | |
| if not self.api_key: | |
| raise ValueError("DeepSeek API key is not set") | |
| if not self.small_llm_model: | |
| raise ValueError("DeepSeek small LLM model is not set") | |
| if require_large and not self.large_llm_model: | |
| raise ValueError("DeepSeek large LLM model is not set") | |
| class GCPConfig: | |
| """ | |
| Configuration class for the GCP configuration. | |
| """ | |
| def __init__(self): | |
| credentials_path = os.getenv("GCP_CREDENTIALS_PATH") | |
| if not credentials_path: | |
| credentials_path = str( | |
| Path(__file__).parent.parent / "gcp_credentials.json" | |
| ) | |
| else: | |
| credentials_path = os.path.expanduser(credentials_path) | |
| if not os.path.isabs(credentials_path): | |
| credentials_path = str(Path(__file__).parent.parent / credentials_path) | |
| logger.debug("GCP credentials_path: %s", credentials_path) | |
| self.credentials_path = credentials_path | |
| self.project_id = os.getenv("GCP_PROJECT_ID") | |
| self.small_llm_model = os.getenv("GCP_SMALL_LLM_MODEL") | |
| self.large_llm_model = os.getenv("GCP_LARGE_LLM_MODEL") | |
| self.embedding_model = os.getenv("GCP_EMBEDDING_MODEL", "text-embedding-005") | |
| def validate_llm(self, require_large: bool): | |
| if not os.path.exists(self.credentials_path): | |
| raise FileNotFoundError( | |
| f"Credentials file not found at: {self.credentials_path}. " | |
| "Please set GCP_CREDENTIALS_PATH environment variable or place gcp_credentials.json in the project root." | |
| ) | |
| required = [self.credentials_path, self.project_id, self.small_llm_model] | |
| if require_large: | |
| required.append(self.large_llm_model) | |
| if required.count(None) > 0: | |
| raise ValueError("GCP LLM configuration is not complete") | |
| def validate_embeddings(self): | |
| if not os.path.exists(self.credentials_path): | |
| raise FileNotFoundError( | |
| f"Credentials file not found at: {self.credentials_path}. " | |
| "Please set GCP_CREDENTIALS_PATH environment variable or place gcp_credentials.json in the project root." | |
| ) | |
| if [self.credentials_path, self.project_id, self.embedding_model].count(None) > 0: | |
| raise ValueError("GCP embeddings configuration is not complete") | |
| class BedrockConfig: | |
| def __init__(self): | |
| self.region = os.getenv("BEDROCK_REGION") | |
| self.access_key_id = os.getenv("BEDROCK_ACCESS_KEY_ID") | |
| self.secret_access_key = os.getenv("BEDROCK_SECRET_ACCESS_KEY") | |
| self.session_token = os.getenv("BEDROCK_SESSION_TOKEN") | |
| self.small_llm_model = os.getenv("BEDROCK_SMALL_LLM_MODEL") | |
| self.large_llm_model = os.getenv("BEDROCK_LARGE_LLM_MODEL") | |
| self.embedding_model = os.getenv("BEDROCK_EMBEDDING_MODEL") | |
| def _validate_auth(self): | |
| if [self.region, self.access_key_id, self.secret_access_key].count(None) > 0: | |
| raise ValueError("Bedrock authentication is not complete") | |
| def validate_llm(self, require_large: bool): | |
| self._validate_auth() | |
| required = [self.small_llm_model] | |
| if require_large: | |
| required.append(self.large_llm_model) | |
| if required.count(None) > 0: | |
| raise ValueError("Bedrock LLM configuration is not complete") | |
| def validate_embeddings(self): | |
| self._validate_auth() | |
| if [self.embedding_model].count(None) > 0: | |
| raise ValueError("Bedrock embeddings configuration is not complete") | |
| class RedisConfig: | |
| """ | |
| Configuration class for the Redis configuration. | |
| """ | |
| def __init__(self): | |
| self.host = os.getenv("REDIS_HOST", "localhost") | |
| port_str = os.getenv("REDIS_PORT", "6379") | |
| self.port = int(port_str) if port_str else 6379 | |
| self.password = os.getenv("REDIS_PASSWORD") or None | |
| class Neo4jConfig: | |
| """ | |
| Configuration class for the Neo4j configuration. | |
| """ | |
| def __init__(self): | |
| self.host = os.getenv("NEO4J_HOST", "localhost") | |
| self.port = os.getenv("NEO4J_PORT", "7687") | |
| self.username = os.getenv("NEO4J_USERNAME", "neo4j") | |
| self.password = os.getenv("NEO4J_PASSWORD", "password") | |
| class PostgreSQLConfig: | |
| """ | |
| Configuration class for the PostgreSQL graph / data / vector stores. | |
| BrainAPI uses one Postgres database per brain. Two cluster-level names are | |
| therefore needed beyond the connection credentials: | |
| * ``system_database`` — the registry database (mirrors Mongo's | |
| ``MONGO_SYSTEM_DATABASE``). It stores the ``data_brains`` table that | |
| catalogs every brain alongside its PAT and metadata. | |
| * ``maintenance_database`` — the database used solely for | |
| ``CREATE DATABASE`` statements when a new brain is provisioned. Almost | |
| every cluster ships with ``postgres`` available for this role. | |
| ``POSTGRES_DATABASE`` is retained as the source of truth for the system | |
| database to preserve backwards compatibility with existing deployments. | |
| ``POSTGRES_SYSTEM_DATABASE`` takes precedence when set explicitly. | |
| """ | |
| def __init__(self): | |
| self.host = os.getenv("POSTGRES_HOST") | |
| port_str = os.getenv("POSTGRES_PORT") | |
| self.port = int(port_str) if port_str else None | |
| self.username = os.getenv("POSTGRES_USERNAME") | |
| self.password = os.getenv("POSTGRES_PASSWORD") | |
| system_db = os.getenv("POSTGRES_SYSTEM_DATABASE") or os.getenv( | |
| "POSTGRES_DATABASE" | |
| ) | |
| self.system_database = system_db | |
| self.database = system_db | |
| self.maintenance_database = os.getenv( | |
| "POSTGRES_MAINTENANCE_DATABASE", "postgres" | |
| ) | |
| def validate_credentials(self) -> None: | |
| """Validate only what is required to open a Postgres connection.""" | |
| if [self.host, self.port, self.username, self.password].count(None) > 0: | |
| raise ValueError("PostgreSQL connection configuration is not complete") | |
| def validate(self) -> None: | |
| self.validate_credentials() | |
| if not self.system_database: | |
| raise ValueError( | |
| "PostgreSQL system database is not configured " | |
| "(set POSTGRES_DATABASE or POSTGRES_SYSTEM_DATABASE)" | |
| ) | |
| class MilvusConfig: | |
| """ | |
| Configuration class for the Milvus configuration. | |
| """ | |
| def __init__(self): | |
| self.host = os.getenv("MILVUS_HOST", "localhost") | |
| port_str = os.getenv("MILVUS_PORT", "19530") | |
| self.port = int(port_str) if port_str else 19530 | |
| self.uri = os.getenv("MILVUS_URI") | |
| self.token = os.getenv("MILVUS_TOKEN") | |
| class EmbeddingsConfig: | |
| """ | |
| Configuration class for the Embeddings configuration. | |
| """ | |
| def __init__(self, mode: str, *, provider: str = "", model: str = ""): | |
| self.local_model = os.getenv("EMBEDDINGS_LOCAL_MODEL", "BAAI/bge-small-en-v1.5") | |
| self.small_model = os.getenv("EMBEDDINGS_SMALL_MODEL", "BAAI/bge-small-en-v1.5") | |
| gemma = provider == "ollama" and model == "embeddinggemma-2:270m" | |
| self.profile = os.getenv("EMBEDDING_PROFILE", "embeddinggemma2-text-v1" if gemma else "") | |
| default_dimension = 512 if gemma else 384 | |
| self.embedding_nodes_dimension = ( | |
| int(os.getenv("EMBEDDING_NODES_DIMENSION")) | |
| if os.getenv("EMBEDDING_NODES_DIMENSION") | |
| else default_dimension | |
| ) | |
| self.embedding_triplets_dimension = ( | |
| int(os.getenv("EMBEDDING_TRIPLETS_DIMENSION")) | |
| if os.getenv("EMBEDDING_TRIPLETS_DIMENSION") | |
| else default_dimension | |
| ) | |
| self.embedding_observations_dimension = ( | |
| int(os.getenv("EMBEDDING_OBSERVATIONS_DIMENSION")) | |
| if os.getenv("EMBEDDING_OBSERVATIONS_DIMENSION") | |
| else default_dimension | |
| ) | |
| self.embedding_data_dimension = ( | |
| int(os.getenv("EMBEDDING_DATA_DIMENSION")) | |
| if os.getenv("EMBEDDING_DATA_DIMENSION") | |
| else default_dimension | |
| ) | |
| self.embedding_relationships_dimension = ( | |
| int(os.getenv("EMBEDDING_RELATIONSHIPS_DIMENSION")) | |
| if os.getenv("EMBEDDING_RELATIONSHIPS_DIMENSION") | |
| else default_dimension | |
| ) | |
| self.native_dimension = int(os.getenv("EMBEDDING_NATIVE_DIMENSION", "768" if gemma else str(self.embedding_data_dimension))) | |
| self.mrl_dimension = int(os.getenv("EMBEDDING_MRL_DIMENSION", "512" if gemma else "0")) | |
| dimensions = { | |
| self.embedding_nodes_dimension, self.embedding_triplets_dimension, | |
| self.embedding_observations_dimension, self.embedding_data_dimension, | |
| self.embedding_relationships_dimension, | |
| } | |
| if min(dimensions) < 1 or self.native_dimension < 1 or self.mrl_dimension < 0: | |
| raise ValueError("Embedding dimensions must be positive (MRL may be zero to disable it)") | |
| if gemma or self.profile == "embeddinggemma2-text-v1": | |
| if not (gemma and self.profile == "embeddinggemma2-text-v1" and self.native_dimension == 768 | |
| and self.mrl_dimension == 512 and dimensions == {512}): | |
| raise ValueError("embeddinggemma2-text-v1 requires ollama embeddinggemma-2:270m, native 768, MRL 512 and all stored dimensions 512; use a new index when changing profiles") | |
| elif self.mrl_dimension: | |
| raise ValueError("MRL is only supported by the qualified embeddinggemma2-text-v1 profile") | |
| class MongoConfig: | |
| """ | |
| Configuration class for the Mongo configuration. | |
| """ | |
| def __init__(self): | |
| self.host = os.getenv("MONGO_HOST", "localhost") | |
| self.port = int(os.getenv("MONGO_PORT")) if os.getenv("MONGO_PORT") else 27017 | |
| self.username = os.getenv("MONGO_USERNAME", "admin") | |
| self.password = os.getenv("MONGO_PASSWORD", "password") | |
| self.connection_string = os.getenv("MONGO_CONNECTION_STRING") | |
| self.system_database = os.getenv("MONGO_SYSTEM_DATABASE", "system") | |
| class CeleryConfig: | |
| """ | |
| Configuration class for the Celery configuration. | |
| """ | |
| def __init__(self): | |
| """ | |
| Initialize CeleryConfig by loading the worker concurrency setting from the environment. | |
| Sets `self.worker_concurrency` from the `CELERY_WORKER_CONCURRENCY` environment variable and validates its presence. | |
| """ | |
| self.worker_concurrency = os.getenv("CELERY_WORKER_CONCURRENCY", "2") | |
| class OllamaConfig: | |
| """ | |
| Configuration class for the Ollama configuration. | |
| """ | |
| def __init__(self): | |
| self.host = os.getenv("OLLAMA_HOST", "localhost") | |
| self.port = os.getenv("OLLAMA_PORT", "11434") | |
| self.llm_small_model = os.getenv("OLLAMA_LLM_SMALL_MODEL", "llama3.2") | |
| self.llm_large_model = os.getenv("OLLAMA_LLM_LARGE_MODEL", "llama3.2") | |
| self.embedding_model = os.getenv("OLLAMA_EMBEDDING_MODEL") or os.getenv( | |
| "EMBEDDINGS_LOCAL_MODEL", "embeddinggemma-2:270m" | |
| ) | |
| self.embedding_keep_alive = os.getenv("OLLAMA_EMBEDDING_KEEP_ALIVE", "10m") | |
| self.num_threads = int(os.getenv("OLLAMA_NUM_THREADS", "2")) | |
| if self.num_threads < 1: | |
| raise ValueError("OLLAMA_NUM_THREADS must be positive") | |
| class SpacyConfig: | |
| """ | |
| Configuration class for the Spacy configuration. | |
| """ | |
| def __init__(self): | |
| self.keep_models_in_memory = ( | |
| os.getenv("SPACY_KEEP_MODELS_IN_MEMORY", "false") == "true" | |
| ) | |
| SEARCH_FTS_REGCONFIGS = frozenset({"italian", "spanish", "simple"}) | |
| _MODES = ("local", "remote") | |
| _PROVIDERS = ( | |
| "ollama", | |
| "azure", | |
| "openai", | |
| "anthropic", | |
| "deepseek", | |
| "gcp_vertex", | |
| "amazon_bedrock", | |
| "fastembed", | |
| "mock", | |
| ) | |
| class Config: | |
| """ | |
| Configuration class for the application. | |
| """ | |
| def brainpat_token(self): | |
| """Deprecated compatibility view of the sole administrator credential.""" | |
| return self.system_pat | |
| def __init__(self): | |
| """ | |
| Initialize the application's central configuration by composing environment-backed sub-configurations and loading runtime flags. | |
| This constructor instantiates Azure, Redis, Neo4j, Milvus, Embeddings, Mongo, GCP, and Celery configuration objects, reads the RUN_GRAPH_CONSOLIDATOR flag into `run_graph_consolidator`, and resolves the canonical administrator credential. | |
| Raises: | |
| ValueError: If administrator credentials conflict or are missing in production. | |
| """ | |
| from src.services.auth_config import resolve_system_pat | |
| self.system_pat = resolve_system_pat() | |
| self.models_mode = os.getenv("MODELS_MODE", "local") | |
| if self.models_mode not in _MODES: | |
| self.models_mode = "local" | |
| default_small_provider = "ollama" if self.models_mode == "local" else "gcp_vertex" | |
| default_large_provider = "ollama" if self.models_mode == "local" else "azure" | |
| default_embeddings_provider = ( | |
| "ollama" if self.models_mode == "local" else "azure" | |
| ) | |
| self.llm_small_provider = os.getenv("LLM_SMALL_PROVIDER", default_small_provider) | |
| self.llm_large_provider = os.getenv("LLM_LARGE_PROVIDER", default_large_provider) | |
| self.embeddings_provider = os.getenv( | |
| "EMBEDDINGS_PROVIDER", default_embeddings_provider | |
| ) | |
| for provider in ( | |
| self.llm_small_provider, | |
| self.llm_large_provider, | |
| self.embeddings_provider, | |
| ): | |
| if provider not in _PROVIDERS: | |
| raise ValueError(f"Invalid provider: {provider}") | |
| use_azure_llm = self.llm_small_provider == "azure" or self.llm_large_provider == "azure" | |
| use_azure_embeddings = self.embeddings_provider == "azure" | |
| use_gcp_llm = self.llm_small_provider == "gcp_vertex" or self.llm_large_provider == "gcp_vertex" | |
| use_gcp_large = self.llm_large_provider == "gcp_vertex" | |
| use_gcp_embeddings = self.embeddings_provider == "gcp_vertex" | |
| use_bedrock_llm = ( | |
| self.llm_small_provider == "amazon_bedrock" | |
| or self.llm_large_provider == "amazon_bedrock" | |
| ) | |
| use_bedrock_large = self.llm_large_provider == "amazon_bedrock" | |
| use_bedrock_embeddings = self.embeddings_provider == "amazon_bedrock" | |
| use_openai_llm = ( | |
| self.llm_small_provider == "openai" or self.llm_large_provider == "openai" | |
| ) | |
| use_openai_large = self.llm_large_provider == "openai" | |
| use_openai_embeddings = self.embeddings_provider == "openai" | |
| use_anthropic_llm = ( | |
| self.llm_small_provider == "anthropic" | |
| or self.llm_large_provider == "anthropic" | |
| ) | |
| use_anthropic_large = self.llm_large_provider == "anthropic" | |
| use_deepseek_llm = ( | |
| self.llm_small_provider == "deepseek" | |
| or self.llm_large_provider == "deepseek" | |
| ) | |
| use_deepseek_large = self.llm_large_provider == "deepseek" | |
| use_ollama = ( | |
| self.llm_small_provider == "ollama" | |
| or self.llm_large_provider == "ollama" | |
| or self.embeddings_provider == "ollama" | |
| ) | |
| self.azure = AzureConfig() if use_azure_llm or use_azure_embeddings else None | |
| if self.azure is not None: | |
| if use_azure_llm: | |
| self.azure.validate_llm() | |
| if use_azure_embeddings: | |
| self.azure.validate_embeddings() | |
| self.gcp = GCPConfig() if use_gcp_llm or use_gcp_embeddings else None | |
| if self.gcp is not None: | |
| if use_gcp_llm: | |
| self.gcp.validate_llm(require_large=use_gcp_large) | |
| if use_gcp_embeddings: | |
| self.gcp.validate_embeddings() | |
| self.bedrock = BedrockConfig() if use_bedrock_llm or use_bedrock_embeddings else None | |
| if self.bedrock is not None: | |
| if use_bedrock_llm: | |
| self.bedrock.validate_llm(require_large=use_bedrock_large) | |
| if use_bedrock_embeddings: | |
| self.bedrock.validate_embeddings() | |
| self.openai = OpenAIConfig() if use_openai_llm or use_openai_embeddings else None | |
| if self.openai is not None: | |
| if use_openai_llm: | |
| self.openai.validate_llm(require_large=use_openai_large) | |
| if use_openai_embeddings: | |
| self.openai.validate_embeddings() | |
| self.anthropic = AnthropicConfig() if use_anthropic_llm else None | |
| if self.anthropic is not None: | |
| self.anthropic.validate_llm(require_large=use_anthropic_large) | |
| self.deepseek = DeepSeekConfig() if use_deepseek_llm else None | |
| if self.deepseek is not None: | |
| self.deepseek.validate_llm(require_large=use_deepseek_large) | |
| self.ollama = OllamaConfig() if use_ollama else None | |
| self.embeddings = EmbeddingsConfig( | |
| mode=self.models_mode, provider=self.embeddings_provider, | |
| model=self.ollama.embedding_model if self.embeddings_provider == "ollama" else "", | |
| ) | |
| self.vector_db = os.getenv("VECTOR_DB", "milvus") | |
| self.data_db = os.getenv("DATA_DB", "mongo") | |
| self.graph_db = os.getenv("GRAPH_DB", "neo4j") | |
| need_mongo = self.data_db.strip().lower() == "mongo" | |
| need_milvus = self.vector_db.strip().lower() == "milvus" | |
| need_neo4j = self.graph_db.strip().lower() == "neo4j" | |
| need_postgres = ( | |
| self.data_db.strip().lower() == "postgresql" | |
| or self.vector_db.strip().lower() == "postgresql" | |
| or self.graph_db.strip().lower() == "networkx" | |
| ) | |
| self.redis = RedisConfig() | |
| self.neo4j = Neo4jConfig() if need_neo4j else None | |
| self.postgresql = PostgreSQLConfig() if need_postgres else None | |
| self.milvus = MilvusConfig() if need_milvus else None | |
| self.mongo = MongoConfig() if need_mongo else None | |
| self.celery = CeleryConfig() | |
| self.spacy = SpacyConfig() | |
| self.run_graph_consolidator = ( | |
| os.getenv("RUN_GRAPH_CONSOLIDATOR", "false") == "true" | |
| ) | |
| self.run_observations = os.getenv("RUN_OBSERVATIONS", "false") == "true" | |
| self.janitor_batch_size = int(os.getenv("JANITOR_BATCH_SIZE", "20")) | |
| self.ingest_janitor_max_llm_calls = int( | |
| os.getenv("INGEST_JANITOR_MAX_LLM_CALLS", "2") | |
| ) | |
| # Accurate-mode hotpath: default to cheap/fast. Set false to restore legacy | |
| # Architect-full-text + per-create Janitor for A/B measurement. | |
| self.ingest_architect_per_unit = ( | |
| os.getenv("INGEST_ARCHITECT_PER_UNIT", "true") == "true" | |
| ) | |
| self.ingest_defer_janitor = ( | |
| os.getenv("INGEST_DEFER_JANITOR", "true") == "true" | |
| ) | |
| self.ingest_architect_mode: Literal["tooler", "schema", "batch"] = ( | |
| validate_ingest_architect_mode( | |
| os.getenv("INGEST_ARCHITECT_MODE", "batch") | |
| ) | |
| ) | |
| self.ingest_architect_max_schema_calls = int( | |
| os.getenv("INGEST_ARCHITECT_MAX_SCHEMA_CALLS", "3") | |
| ) | |
| self.ingest_architect_escalate_max_turns = int( | |
| os.getenv("INGEST_ARCHITECT_ESCALATE_MAX_TURNS", "3") | |
| ) | |
| self.ingest_architect_escalate = ( | |
| os.getenv("INGEST_ARCHITECT_ESCALATE", "true") == "true" | |
| ) | |
| self.ingest_architect_dense_entity_threshold = int( | |
| os.getenv("INGEST_ARCHITECT_DENSE_ENTITY_THRESHOLD", "12") | |
| ) | |
| self.ingest_architect_dense_max_chars = int( | |
| os.getenv("INGEST_ARCHITECT_DENSE_MAX_CHARS", "1200") | |
| ) | |
| self.ingest_architect_prior_context: Literal[ | |
| "auto", "scratchpad", "raw" | |
| ] = validate_ingest_architect_prior_context( | |
| os.getenv("INGEST_ARCHITECT_PRIOR_CONTEXT", "auto") | |
| ) | |
| self.ingest_architect_scratchpad_token_cap = int( | |
| os.getenv("INGEST_ARCHITECT_SCRATCHPAD_TOKEN_CAP", "500") | |
| ) | |
| self.docparser_endpoint = os.getenv("DOCPARSER_ENDPOINT") | |
| self.docparser_token = os.getenv("DOCPARSER_TOKEN") | |
| self.app_host = os.getenv("APP_HOST") | |
| self.pipeline_mode: Literal["lightweight", "accurate"] = validate_pipeline_mode( | |
| os.getenv("PIPELINE_MODE", "accurate") | |
| ) | |
| self.ocr_mode: Literal["docling", "docparser"] = os.getenv( | |
| "OCR_MODE", "docling" | |
| ) | |
| self.agentic_architecture: Literal["custom", "langchain"] = os.getenv( | |
| "AGENTIC_ARCHITECTURE", "custom" | |
| ) | |
| self.search_enabled = os.getenv("SEARCH_ENABLED", "false") == "true" | |
| self.search_use_dense = os.getenv("SEARCH_USE_DENSE", "true") == "true" | |
| self.search_use_bm25 = os.getenv("SEARCH_USE_BM25", "true") == "true" | |
| community_raw = os.getenv("SEARCH_COMMUNITY_LABELS", "TYPE,CLASS,TOPIC") | |
| community_labels = [ | |
| part.strip() for part in community_raw.split(",") if part.strip() | |
| ] | |
| self.search_community_labels = community_labels or [ | |
| "TYPE", | |
| "CLASS", | |
| "TOPIC", | |
| ] | |
| self.search_neighbor_fanout = int(os.getenv("SEARCH_NEIGHBOR_FANOUT", "50")) | |
| self.search_fusion: Literal["rrf", "cc"] = validate_search_fusion( | |
| os.getenv("SEARCH_FUSION", "rrf") | |
| ) | |
| self.search_fusion_alpha = float(os.getenv("SEARCH_FUSION_ALPHA", "0.5")) | |
| self.search_literal_fill = os.getenv("SEARCH_LITERAL_FILL", "false") == "true" | |
| self.search_bm25_k1 = float(os.getenv("SEARCH_BM25_K1", "1.2")) | |
| self.search_bm25_b = float(os.getenv("SEARCH_BM25_B", "0.75")) | |
| self.search_fts_regconfig = os.getenv("SEARCH_FTS_REGCONFIG", "").strip().lower() | |
| self.search_fts_brains = frozenset( | |
| part.strip() | |
| for part in os.getenv("SEARCH_FTS_BRAINS", "").split(",") | |
| if part.strip() | |
| ) | |
| self.context_passage_mode: Literal[ | |
| "hybrid", "bm25", "dense", "ilike" | |
| ] = validate_context_passage_mode( | |
| os.getenv("CONTEXT_PASSAGE_MODE", "hybrid") | |
| ) | |
| validate_search_config( | |
| enabled=self.search_enabled, | |
| use_dense=self.search_use_dense, | |
| use_bm25=self.search_use_bm25, | |
| data_db=self.data_db, | |
| bm25_k1=self.search_bm25_k1, | |
| bm25_b=self.search_bm25_b, | |
| ) | |
| def search_fts_regconfig_for_brain(self, brain_id: str) -> str | None: | |
| bid = (brain_id or "").strip() | |
| if not bid.lower().startswith("searchbench"): | |
| return None | |
| if bid not in self.search_fts_brains: | |
| return None | |
| if self.search_fts_regconfig not in SEARCH_FTS_REGCONFIGS: | |
| return None | |
| return self.search_fts_regconfig | |
| config = Config() | |