130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Configuration system for chatbot instances.
|
|
Loads JSON configuration files from configs/ directory.
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
from pathlib import Path
|
|
from dataclasses import dataclass
|
|
from typing import Optional, Dict, Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Cache for loaded configs
|
|
_config_cache: Dict[str, 'ChatbotConfig'] = {}
|
|
|
|
|
|
@dataclass
|
|
class DatabaseConfig:
|
|
"""Database configuration for a chatbot instance."""
|
|
schema: Dict[str, Any]
|
|
connector: str = "preprocessor"
|
|
|
|
|
|
@dataclass
|
|
class ToolConfig:
|
|
"""Tool configuration for a chatbot instance."""
|
|
sql: Dict[str, Any]
|
|
tavily: Optional[Dict[str, Any]] = None
|
|
streaming: Dict[str, Any] = None
|
|
|
|
|
|
@dataclass
|
|
class ModelConfig:
|
|
"""Model configuration for a chatbot instance."""
|
|
operationType: str = "DATA_ANALYSE"
|
|
processingMode: str = "DETAILED"
|
|
|
|
|
|
@dataclass
|
|
class ChatbotConfig:
|
|
"""Configuration for a chatbot instance."""
|
|
id: str
|
|
name: str
|
|
systemPrompt: str
|
|
database: DatabaseConfig
|
|
tools: ToolConfig
|
|
model: ModelConfig
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'ChatbotConfig':
|
|
"""Create ChatbotConfig from dictionary."""
|
|
return cls(
|
|
id=data.get("id", "default"),
|
|
name=data.get("name", "Default Chatbot"),
|
|
systemPrompt=data.get("systemPrompt", "You are a helpful assistant."),
|
|
database=DatabaseConfig(
|
|
schema=data.get("database", {}).get("schema", {}),
|
|
connector=data.get("database", {}).get("connector", "preprocessor")
|
|
),
|
|
tools=ToolConfig(
|
|
sql=data.get("tools", {}).get("sql", {"enabled": True}),
|
|
tavily=data.get("tools", {}).get("tavily"),
|
|
streaming=data.get("tools", {}).get("streaming", {"enabled": True})
|
|
),
|
|
model=ModelConfig(
|
|
operationType=data.get("model", {}).get("operationType", "DATA_ANALYSE"),
|
|
processingMode=data.get("model", {}).get("processingMode", "DETAILED")
|
|
)
|
|
)
|
|
|
|
|
|
def load_chatbot_config(config_id: str) -> ChatbotConfig:
|
|
"""
|
|
Load chatbot configuration from JSON file.
|
|
|
|
Args:
|
|
config_id: Configuration ID (e.g., "althaus", "default")
|
|
|
|
Returns:
|
|
ChatbotConfig instance
|
|
|
|
Raises:
|
|
FileNotFoundError: If config file not found
|
|
ValueError: If config file is invalid
|
|
"""
|
|
# Check cache first
|
|
if config_id in _config_cache:
|
|
logger.debug(f"Returning cached config for {config_id}")
|
|
return _config_cache[config_id]
|
|
|
|
# Get path to configs directory
|
|
current_dir = Path(__file__).parent
|
|
configs_dir = current_dir / "configs"
|
|
config_file = configs_dir / f"{config_id}.json"
|
|
|
|
if not config_file.exists():
|
|
# Try default config if requested config not found
|
|
if config_id != "default":
|
|
logger.warning(f"Config {config_id} not found, trying default")
|
|
return load_chatbot_config("default")
|
|
raise FileNotFoundError(f"Chatbot config file not found: {config_file}")
|
|
|
|
try:
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
config = ChatbotConfig.from_dict(data)
|
|
|
|
# Cache the config
|
|
_config_cache[config_id] = config
|
|
logger.info(f"Loaded chatbot config: {config_id} ({config.name})")
|
|
|
|
return config
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Error parsing chatbot config JSON {config_file}: {e}")
|
|
raise ValueError(f"Invalid JSON in config file {config_file}: {e}")
|
|
except Exception as e:
|
|
logger.error(f"Error loading chatbot config {config_file}: {e}")
|
|
raise
|
|
|
|
|
|
def clear_config_cache():
|
|
"""Clear the configuration cache."""
|
|
global _config_cache
|
|
_config_cache.clear()
|
|
logger.debug("Cleared chatbot config cache")
|