# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Chatbot instance configuration management. Handles loading and applying instance-specific configurations. """ import logging from typing import Optional, Dict, Any, List from modules.interfaces.interfaceFeatures import getFeatureInterface from modules.interfaces.interfaceDbApp import getRootInterface logger = logging.getLogger(__name__) class ChatbotConfig: """ Chatbot instance configuration structure. Provides defaults and validation for chatbot instance configs. """ # Default configuration DEFAULT_CONFIG = { "connector": { "types": ["preprocessor"], # Array of database connector types: "preprocessor", "custom" "type": "preprocessor", # Legacy: single connector type (for backward compatibility) "customConnectorClass": None # For custom connectors }, "prompts": { "useCustomPrompts": False, "customAnalysisPrompt": None, "customFinalAnswerPrompt": None, "customSystemPrompt": None # For LangGraph workflow (single system prompt) }, "behavior": { "maxQueries": 5, "enableWebResearch": True, "enableRetryOnEmpty": True, "maxRetryAttempts": 2 }, "database": { "schema": None, # Custom schema info if needed "tablePrefix": None # Custom table prefix if needed } } def __init__(self, config: Optional[Dict[str, Any]] = None): """ Initialize chatbot config with defaults and overrides. Args: config: Instance-specific config dict (from FeatureInstance.config) """ self.config = self._merge_config(config or {}) def _merge_config(self, instance_config: Dict[str, Any]) -> Dict[str, Any]: """ Merge instance config with defaults, handling nested dicts. Args: instance_config: Instance-specific config Returns: Merged configuration dict """ merged = self.DEFAULT_CONFIG.copy() # Deep merge nested dicts for key, value in instance_config.items(): if key in merged and isinstance(merged[key], dict) and isinstance(value, dict): merged[key] = {**merged[key], **value} else: merged[key] = value return merged @property def connector_types(self) -> List[str]: """Get connector types as list (supports multiple connectors).""" connector_config = self.config.get("connector", {}) # Support new array format types = [] if "types" in connector_config and isinstance(connector_config["types"], list): types = connector_config["types"] # Fallback to legacy single type format elif "type" in connector_config: types = [connector_config["type"]] else: types = ["preprocessor"] # Filter out 'websearch' (not a database connector, handled separately via enableWebResearch) types = [t for t in types if t != "websearch"] # Ensure at least one connector if not types: types = ["preprocessor"] return types @property def connector_type(self) -> str: """Get primary connector type (preprocessor, custom).""" # For backward compatibility, return first connector type types = self.connector_types return types[0] if types else "preprocessor" @property def custom_connector_class(self) -> Optional[str]: """Get custom connector class name if using custom connector.""" return self.config.get("connector", {}).get("customConnectorClass") @property def use_custom_prompts(self) -> bool: """Check if custom prompts should be used. Always true since prompts are required.""" # Prompts are now required, so this is always true if prompts are configured return bool(self.config.get("prompts", {}).get("customAnalysisPrompt") or self.config.get("prompts", {}).get("customFinalAnswerPrompt")) @property def custom_analysis_prompt(self) -> Optional[str]: """Get custom analysis prompt (required for chatbot instances).""" prompt = self.config.get("prompts", {}).get("customAnalysisPrompt") if not prompt: logger.warning("custom_analysis_prompt is not configured - this is required for chatbot instances") return prompt @property def custom_final_answer_prompt(self) -> Optional[str]: """Get custom final answer prompt (required for chatbot instances).""" prompt = self.config.get("prompts", {}).get("customFinalAnswerPrompt") if not prompt: logger.warning("custom_final_answer_prompt is not configured - this is required for chatbot instances") return prompt @property def custom_system_prompt(self) -> Optional[str]: """Get custom system prompt for LangGraph workflow.""" # Prefer customSystemPrompt, fallback to customAnalysisPrompt prompt = self.config.get("prompts", {}).get("customSystemPrompt") if not prompt: prompt = self.config.get("prompts", {}).get("customAnalysisPrompt") return prompt @property def max_queries(self) -> int: """Get maximum number of queries allowed.""" return self.config.get("behavior", {}).get("maxQueries", 5) @property def enable_web_research(self) -> bool: """Check if web research is enabled.""" return self.config.get("behavior", {}).get("enableWebResearch", True) @property def enable_retry_on_empty(self) -> bool: """Check if retry on empty results is enabled.""" return self.config.get("behavior", {}).get("enableRetryOnEmpty", True) @property def max_retry_attempts(self) -> int: """Get maximum retry attempts.""" return self.config.get("behavior", {}).get("maxRetryAttempts", 2) def get_connector_instance(self): """ Get connector instance based on configuration. Uses the primary (first) connector type from the configured connectors. Returns: Connector instance (PreprocessorConnector, or custom connector if configured) """ # Use primary connector type (first in the list) connector_type = self.connector_type.lower() if connector_type == "preprocessor": from modules.connectors.connectorPreprocessor import PreprocessorConnector return PreprocessorConnector() elif connector_type == "custom" and self.custom_connector_class: # Dynamic import for custom connectors try: module_path, class_name = self.custom_connector_class.rsplit(".", 1) module = __import__(module_path, fromlist=[class_name]) connector_class = getattr(module, class_name) return connector_class() except Exception as e: logger.error(f"Failed to load custom connector {self.custom_connector_class}: {e}") raise ValueError(f"Invalid custom connector: {self.custom_connector_class}") else: # Default to PreprocessorConnector logger.warning(f"Unknown connector type '{connector_type}', using PreprocessorConnector") from modules.connectors.connectorPreprocessor import PreprocessorConnector return PreprocessorConnector() def get_chatbot_config(instance_id: Optional[str]) -> ChatbotConfig: """ Load chatbot configuration for a feature instance. Args: instance_id: FeatureInstance ID (None for default config) Returns: ChatbotConfig instance with merged defaults and instance config """ if not instance_id: # Return default config if no instance ID provided return ChatbotConfig() try: rootInterface = getRootInterface() featureInterface = getFeatureInterface(rootInterface.db) instance = featureInterface.getFeatureInstance(instance_id) if not instance: logger.warning(f"Feature instance {instance_id} not found, using default config") return ChatbotConfig() # Verify it's a chatbot instance if instance.featureCode != "chatbot": logger.warning(f"Instance {instance_id} is not a chatbot instance, using default config") return ChatbotConfig() # Load config from instance instance_config = instance.config if hasattr(instance, 'config') and instance.config else {} return ChatbotConfig(instance_config) except Exception as e: logger.error(f"Error loading chatbot config for instance {instance_id}: {e}") # Return default config on error return ChatbotConfig()