[ALGO-3] Make MCTS Parameters Configurable Per Domain

[MVPandey]

⚙️ ADAPTIVE MCTS PARAMETERS

Priority: MEDIUM - Algorithm Optimization

Problem

Fixed MCTS parameters (UCB1 exploration constant, branching factors) don't suit all conversation domains. One-size-fits-all approach suboptimal.

Current Issues:

• Hardcoded UCB1 exploration constant (1.414) in node.py:43-49
• Fixed branching factors and simulation depth
• No adaptation based on domain requirements or performance metrics

Solution

Make MCTS parameters configurable per domain with automatic optimization based on performance feedback.

Domain-Specific Parameter Configuration

# app/services/mcts/config.py (enhanced)
@dataclass
class MCTSConfig:
    # Core MCTS parameters
    exploration_constant: float = 1.414
    branching_factor: int = 3
    simulation_depth: int = 3
    mcts_iterations: int = 10
    
    # Resource constraints
    max_tree_depth: int = 10
    max_nodes: int = 1000
    
    # Domain-specific optimizations
    domain_name: str = "general"
    confidence_threshold: float = 0.7
    early_stopping_enabled: bool = True
    
    @classmethod
    def for_domain(cls, domain_name: str) -> 'MCTSConfig':
        """Get optimized parameters for specific domain"""
        domain_configs = {
            "emotional_intelligence": cls(
                exploration_constant=1.414,  # Standard exploration
                branching_factor=3,
                simulation_depth=3,
                confidence_threshold=0.75
            ),
            "technical_support": cls(
                exploration_constant=1.0,     # More conservative
                branching_factor=2,           # Focused responses
                simulation_depth=4,           # Deeper technical analysis
                confidence_threshold=0.8
            ),
            "sales": cls(
                exploration_constant=1.8,     # More exploration for creativity
                branching_factor=4,           # More response variety
                simulation_depth=2,           # Shorter interactions
                confidence_threshold=0.6
            ),
            "education": cls(
                exploration_constant=1.2,     # Balanced exploration
                branching_factor=3,
                simulation_depth=4,           # Educational depth
                confidence_threshold=0.8
            )
        }
        return domain_configs.get(domain_name, cls())

Adaptive Parameter Optimization

# app/services/mcts/parameter_optimizer.py
class MCTSParameterOptimizer:
    def __init__(self):
        self.performance_history = {}
        self.optimization_window = 100  # Optimize every 100 requests
    
    async def optimize_parameters(
        self,
        domain: str,
        current_config: MCTSConfig,
        performance_metrics: Dict[str, float]
    ) -> MCTSConfig:
        """
        Optimize MCTS parameters based on performance feedback
        """
        # Track performance over time
        self._record_performance(domain, current_config, performance_metrics)
        
        # Check if optimization is needed
        if not self._should_optimize(domain):
            return current_config
        
        # Perform parameter optimization
        optimized_config = await self._optimize_config(domain, current_config)
        
        return optimized_config
    
    def _optimize_config(self, domain: str, config: MCTSConfig) -> MCTSConfig:
        """
        Use performance history to adjust parameters
        """
        history = self.performance_history[domain]
        
        # Adjust exploration constant based on quality vs diversity trade-off
        if history.avg_quality < 0.7 and history.avg_diversity > 0.8:
            # High diversity, low quality -> reduce exploration
            config.exploration_constant *= 0.9
        elif history.avg_quality > 0.8 and history.avg_diversity < 0.6:
            # High quality, low diversity -> increase exploration
            config.exploration_constant *= 1.1
        
        # Adjust branching factor based on performance vs cost
        if history.avg_cost > history.target_cost:
            # Too expensive -> reduce branching
            config.branching_factor = max(2, config.branching_factor - 1)
        elif history.avg_quality > 0.8 and history.avg_cost < history.target_cost * 0.8:
            # Good quality, low cost -> can afford more branching
            config.branching_factor = min(5, config.branching_factor + 1)
        
        return config

Dynamic Early Stopping

# app/services/mcts/early_stopping.py
class EarlyStoppingController:
    def should_stop_early(
        self,
        iteration: int,
        current_best_score: float,
        score_history: List[float],
        config: MCTSConfig
    ) -> bool:
        """
        Determine if MCTS should stop early based on convergence
        """
        if not config.early_stopping_enabled:
            return False
        
        # Stop if we've reached minimum iterations and score is stable
        if iteration >= 3:
            recent_scores = score_history[-3:]
            if all(abs(score - current_best_score) < 0.05 for score in recent_scores):
                return True
        
        # Stop if we've exceeded confidence threshold
        if current_best_score > config.confidence_threshold:
            return True
        
        return False

Updated MCTS Algorithm Integration

# app/services/mcts/algorithm.py (updated)
class MCTSAlgorithm:
    def __init__(self, parameter_optimizer: MCTSParameterOptimizer = None):
        self.parameter_optimizer = parameter_optimizer
        self.early_stopping = EarlyStoppingController()
    
    async def run(
        self,
        conversation: Conversation,
        config: MCTSConfig,
        domain: ConversationDomain = None
    ) -> MCTSResult:
        # Use domain-specific configuration if provided
        if domain:
            config = MCTSConfig.for_domain(domain.get_config().name)
            # Apply any runtime optimizations
            if self.parameter_optimizer:
                config = await self.parameter_optimizer.get_optimized_config(
                    domain.get_config().name, config
                )
        
        # Run MCTS with adaptive parameters
        root = MCTSNode(conversation)
        score_history = []
        
        for iteration in range(config.mcts_iterations):
            # Update exploration constant dynamically
            dynamic_exploration = self._adjust_exploration_for_iteration(
                config.exploration_constant, iteration, config.mcts_iterations
            )
            
            # Standard MCTS steps with dynamic parameters
            node = await self._select_node(root, dynamic_exploration)
            await self._expand_and_simulate(node, config)
            self._backpropagate(node)
            
            # Check for early stopping
            current_score = root.get_best_child().avg_score
            score_history.append(current_score)
            
            if self.early_stopping.should_stop_early(
                iteration, current_score, score_history, config
            ):
                break
        
        return self._create_result(root, config, len(score_history))

Implementation Steps

• Create domain-specific parameter configurations
• Implement parameter optimization system
• Add early stopping based on convergence
• Update MCTS algorithm to use dynamic parameters
• Add performance tracking and feedback loop
• Create parameter tuning dashboard

Expected Benefits

• Optimized performance per domain with appropriate exploration/exploitation balance
• Cost reduction through early stopping and right-sized branching
• Quality improvement through domain-specific parameter tuning
• Adaptive behavior that improves over time

Acceptance Criteria

• Domain-specific parameter configurations
• Automatic parameter optimization based on performance
• Early stopping reduces unnecessary iterations
• Performance improvement > 20% per domain
• Parameter tuning dashboard for monitoring

Effort: Medium (4-6 days)