moha_mind/config.py
Rama master · réplica de lectura del repositorio original.
"""MohaMind configuration - all settings from environment variables."""
from pathlib import Path
from typing import Literal
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from moha_mind.utils.configuration import is_configured_key
LLM_PROVIDERS = {
"zai": {
"base_url": "https://api.z.ai/api/paas/v4/",
"default_model": "glm-5-turbo",
},
"openai": {
"base_url": None,
"default_model": "gpt-4o-mini",
},
}
LLMProvider = Literal["zai", "openai"]
LLMStrategy = Literal["solo", "fallback", "verify"]
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
zai_api_key: str = ""
zai_model: str = "glm-5-turbo"
zai_base_url: str = "https://api.z.ai/api/paas/v4/"
openai_api_key: str = ""
openai_model: str = "gpt-4o-mini"
openai_base_url: str = ""
@field_validator("zai_api_key", "openai_api_key")
@classmethod
def validate_api_key(cls, value: str) -> str:
return value.strip() if is_configured_key(value) else ""
primary_llm: LLMProvider = "zai"
agent_max_tool_rounds: int = Field(default=10, ge=1, le=100)
agent_max_tool_calls: int = Field(default=40, ge=1, le=500)
agent_tool_timeout_seconds: float = Field(default=60, gt=0, le=3600, allow_inf_nan=False)
agent_max_tool_result_chars: int = Field(default=16000, ge=128, le=1000000)
# How the secondary LLM is used:
# solo -> no secondary at all
# fallback -> secondary is only called if primary fails
# verify -> secondary reviews primary answers for accuracy
llm_strategy: LLMStrategy = "fallback"
# Kept for backwards compatibility with older .env files.
# If set, it acts as a shortcut to derive strategy + secondary provider.
fallback_llm: Literal["zai", "openai", "none", ""] = ""
# Which provider is used as the secondary brain (fallback or verifier)
secondary_llm: Literal["zai", "openai", "none"] = "openai"
# Verifier behavior (only used when llm_strategy == "verify")
verifier_strictness: Literal["lenient", "balanced", "strict"] = "balanced"
verifier_max_retries: int = 1
telegram_bot_token: str = ""
telegram_chat_id: str = ""
# Extra Telegram user/chat IDs (comma-separated) allowed to talk to the bot
# in addition to TELEGRAM_CHAT_ID. Everyone else is refused.
telegram_allowed_user_ids: str = ""
telegram_allow_destructive: bool = True
whatsapp_session_dir: str = "./credentials/whatsapp"
whatsapp_node_path: str = "node"
voice_enabled: bool = False
voice_model: str = "small"
voice_device: Literal["cpu", "cuda", "auto"] = "cpu"
voice_compute_type: Literal["int8", "float16", "float32", "int8_float16", "auto"] = "int8"
voice_language: Literal["auto", "ar", "en"] = "auto"
voice_cpu_threads: int = Field(default=4, ge=1, le=32)
voice_max_duration_seconds: int = Field(default=300, ge=1, le=1200)
voice_max_file_mb: int = Field(default=10, ge=1, le=20)
voice_local_files_only: bool = False
@field_validator("voice_model")
@classmethod
def validate_voice_model(cls, value: str) -> str:
value = value.strip()
if not value or value.endswith(".en") or "distil" in value.lower():
raise ValueError("Use a multilingual Whisper model such as small, medium, or large-v3")
return value
google_credentials_path: str = "./credentials/google_credentials.json"
google_token_path: str = "./credentials/google_token.json"
ms_client_id: str = ""
ms_client_secret: str = ""
ms_tenant_id: str = "common"
ms_token_path: str = "./credentials/ms_token.json"
timezone: str = "Asia/Riyadh"
# Language for proactive messages (briefings, reminders, alerts) and the
# instruction sent with Telegram commands: "ar" or "en". Free-form chat
# always mirrors whatever language the user writes in.
agent_language: str = "ar"
morning_briefing_time: str = "08:00"
weekly_review_day: str = "sun"
weekly_review_time: str = "19:00"
memory_dir: str = "./memory"
# Claude-compatible external MCP servers config ({"mcpServers": {...}}).
# See mcp_servers.example.json. Missing file = no external servers.
mcp_servers_config: str = "./mcp_servers.json"
log_level: str = "INFO"
# Rotating log files live here (gitignored), next to memory/.
log_dir: str = "./logs"
# ----- Memory router + summaries (Phase A) -----
# When enabled, the system prompt stops pasting every category in full
# and instead injects rolling summaries plus the top-K relevant categories.
memory_router_enabled: bool = True
memory_router_max_categories: int = 4
memory_summaries_enabled: bool = True
memory_summary_model: str = "" # empty = use primary LLM model
# ----- Semantic memory (Phase B) -----
# Embedding backend: 'none' | 'openai' | 'local'
# none -> semantic search disabled (FTS5 only)
# openai -> uses openai_api_key with text-embedding-3-small by default
# local -> uses sentence-transformers locally (requires optional dep)
embedding_backend: Literal["none", "openai", "local"] = "none"
embedding_model: str = "" # auto-pick based on backend when empty
semantic_top_k: int = 5
# ----- Nightly consolidator (Phase C) -----
consolidator_enabled: bool = False
# 'auto' applies all changes silently, 'confirm' queues all for approval,
# 'hybrid' auto-applies low-risk additions and asks for conflicts/sensitive writes.
consolidator_mode: Literal["auto", "confirm", "hybrid"] = "hybrid"
consolidator_time: str = "02:30"
consolidator_send_digest: bool = True
# ----- Reliability guardian -----
reliability_guardian_enabled: bool = True
reliability_guardian_time: str = "03:10"
reliability_send_digest: bool = True
memory_backup_retention_days: int = 30
memory_backup_max_count: int = 60
# ----- Privacy tiers (cross-cutting) -----
# Comma-separated category names. Sensitive categories:
# - are redacted before being sent to the verifier
# - are redacted before being stored in sessions.db (when enabled)
# - never appear verbatim in daily_log (when enabled)
sensitive_categories: str = "finances,health,documents"
privacy_redact_sessions: bool = True
privacy_redact_verifier: bool = True
privacy_redact_daily_log: bool = True
@property
def memory_path(self) -> Path:
return Path(self.memory_dir)
@property
def telegram_allowed_ids(self) -> frozenset[str]:
"""All Telegram IDs allowed to talk to the bot (owner chat + extras)."""
ids = {self.telegram_chat_id.strip()}
ids.update(part.strip() for part in self.telegram_allowed_user_ids.split(","))
return frozenset(value for value in ids if value)
def _provider_config(self, provider: str) -> dict | None:
if provider == "zai" and self.zai_api_key:
return {
"api_key": self.zai_api_key,
"model": self.zai_model,
"base_url": self.zai_base_url,
"provider": "zai",
}
if provider == "openai" and self.openai_api_key:
return {
"api_key": self.openai_api_key,
"model": self.openai_model,
"base_url": self.openai_base_url or None,
"provider": "openai",
}
return None
@property
def active_llm_config(self) -> dict:
cfg = self._provider_config(self.primary_llm)
if cfg:
return cfg
other = "openai" if self.primary_llm == "zai" else "zai"
fallback_cfg = self._provider_config(other)
if fallback_cfg:
return fallback_cfg
# No keys at all — return a dummy config that will fail politely.
if self.primary_llm == "zai":
return {
"api_key": self.zai_api_key or "missing-key",
"model": self.zai_model,
"base_url": self.zai_base_url,
"provider": "zai",
}
return {
"api_key": self.openai_api_key or "missing-key",
"model": self.openai_model,
"base_url": self.openai_base_url or None,
"provider": "openai",
}
@property
def _resolved_secondary_provider(self) -> str:
"""Decide which provider is the secondary brain.
Precedence:
1. Legacy `fallback_llm` env value (when set to a real provider) —
this keeps backwards compatibility with existing .env files.
2. Explicit `secondary_llm` env value (when not 'none').
3. The opposite of the primary provider.
"""
if self.fallback_llm and self.fallback_llm not in ("none", ""):
return self.fallback_llm
if self.secondary_llm and self.secondary_llm != "none":
return self.secondary_llm
return "openai" if self.primary_llm == "zai" else "zai"
@property
def _strategy_effective(self) -> LLMStrategy:
"""Normalize legacy config to a strategy value."""
# Legacy: fallback_llm="none" used to mean "no secondary".
if self.secondary_llm == "none" or self.fallback_llm == "none":
return "solo"
return self.llm_strategy
@property
def fallback_llm_config(self) -> dict | None:
"""Return the secondary LLM config when it should act as a FALLBACK.
Returns None when the strategy is not "fallback" or no secondary key exists.
"""
if self._strategy_effective != "fallback":
return None
return self._provider_config(self._resolved_secondary_provider)
@property
def verifier_llm_config(self) -> dict | None:
"""Return the secondary LLM config when it should act as a VERIFIER."""
if self._strategy_effective != "verify":
return None
return self._provider_config(self._resolved_secondary_provider)
@property
def secondary_llm_config(self) -> dict | None:
"""Return the secondary LLM config (either fallback or verifier role)."""
if self._strategy_effective == "solo":
return None
return self._provider_config(self._resolved_secondary_provider)
@property
def effective_strategy(self) -> LLMStrategy:
"""Public accessor for the resolved strategy."""
return self._strategy_effective
settings = Settings()