Dynamic Per-User Tool Generation MCP Server

""" Decision Engine Implements score-based decisioning logic for slot resolution. Uses dual-threshold approach: 1. Min Score Threshold: Minimum confidence to accept a match 2. Score Gap Delta: Minimum gap between top 2 candidates for auto-resolution """ import logging from typing import List, Optional, Dict, Any from slot_resolution.core.models import ( Candidate, SlotResolutionResponse, ResolutionStatus, ResolvedEntity, ResolutionMethod ) logger = logging.getLogger(__name__) class DecisionEngine: """ Applies score-based decisioning logic to determine resolution outcome. The engine uses configurable thresholds to decide whether to: - Auto-resolve to a single match - Request user disambiguation - Report no match """ # Default thresholds per entity type DEFAULT_MIN_SCORES = { "impact": 0.65, "urgency": 0.65, "priority": 0.65, "status": 0.60, "category": 0.60, "source": 0.60, "location": 0.55, "department": 0.55, "user": 0.70, "usergroup": 0.65, "vendor": 0.70 } DEFAULT_SCORE_GAP_DELTA = 0.15 DEFAULT_MAX_CANDIDATES = 5 def __init__( self, min_scores: Optional[Dict[str, float]] = None, score_gap_delta: float = DEFAULT_SCORE_GAP_DELTA, max_candidates: int = DEFAULT_MAX_CANDIDATES ): """ Initialize the decision engine. Args: min_scores: Dictionary mapping entity types to minimum scores score_gap_delta: Minimum score gap for auto-resolution max_candidates: Maximum candidates to return for disambiguation """ self.min_scores = min_scores or self.DEFAULT_MIN_SCORES self.score_gap_delta = score_gap_delta self.max_candidates = max_candidates logger.info( f"DecisionEngine initialized with score_gap_delta={score_gap_delta}, " f"max_candidates={max_candidates}" ) def get_min_score(self, entity_type: str) -> float: """ Get minimum score threshold for an entity type. Args: entity_type: Type of entity Returns: Minimum score threshold """ return self.min_scores.get(entity_type, 0.60) def decide( self, candidates: List[Candidate], entity_type: str, input_query: str, normalized_query: str, min_score_override: Optional[float] = None, score_gap_delta_override: Optional[float] = None ) -> SlotResolutionResponse: """ Apply decisioning logic to candidate list. Args: candidates: List of candidate matches from search entity_type: Type of entity being resolved input_query: Original user input normalized_query: Normalized input min_score_override: Override default min score score_gap_delta_override: Override default score gap delta Returns: SlotResolutionResponse with appropriate status """ # Get thresholds min_score = min_score_override or self.get_min_score(entity_type) score_gap_delta = score_gap_delta_override or self.score_gap_delta # Filter candidates above min_score qualified_candidates = [ c for c in candidates if c.confidence >= min_score ] logger.debug( f"Decisioning for '{input_query}': " f"{len(candidates)} total candidates, " f"{len(qualified_candidates)} above threshold {min_score}" ) # No matches above threshold if len(qualified_candidates) == 0: return self._create_no_match_response( entity_type=entity_type, input_query=input_query, normalized_query=normalized_query, min_score=min_score ) # Single match above threshold if len(qualified_candidates) == 1: return self._create_resolved_response( candidate=qualified_candidates[0], input_query=input_query, normalized_query=normalized_query, method=ResolutionMethod.FUZZY_MATCH, config={"minScore": min_score, "scoreGapDelta": score_gap_delta} ) # Multiple matches - check score gap top1 = qualified_candidates[0] top2 = qualified_candidates[1] score_gap = top1.confidence - top2.confidence logger.debug( f"Top 2 candidates: " f"1st={top1.canonical_name} ({top1.confidence:.3f}), " f"2nd={top2.canonical_name} ({top2.confidence:.3f}), " f"gap={score_gap:.3f}" ) # Clear winner (large score gap) if score_gap >= score_gap_delta: return self._create_resolved_response( candidate=top1, input_query=input_query, normalized_query=normalized_query, method=ResolutionMethod.FUZZY_MATCH_WITH_GAP, config={ "minScore": min_score, "scoreGapDelta": score_gap_delta, "actualGap": score_gap } ) # Ambiguous - require disambiguation return self._create_disambiguation_response( candidates=qualified_candidates[:self.max_candidates], entity_type=entity_type, input_query=input_query, normalized_query=normalized_query, config={"minScore": min_score, "scoreGapDelta": score_gap_delta} ) def _create_resolved_response( self, candidate: Candidate, input_query: str, normalized_query: str, method: ResolutionMethod, config: Dict[str, Any] ) -> SlotResolutionResponse: """Create a RESOLVED response.""" resolved_entity = ResolvedEntity( id=candidate.id, canonical_name=candidate.canonical_name, entity_type=candidate.entity_type, confidence=candidate.confidence, method=method, attributes=candidate.attributes ) return SlotResolutionResponse( status=ResolutionStatus.RESOLVED, resolved=resolved_entity, input_echo=input_query, normalization=normalized_query, method=method, config=config ) def _create_disambiguation_response( self, candidates: List[Candidate], entity_type: str, input_query: str, normalized_query: str, config: Dict[str, Any] ) -> SlotResolutionResponse: """Create a MULTIPLE_MATCHES response.""" guidance_text = ( f"Multiple {entity_type}s match '{input_query}'. " f"Please select one:" ) return SlotResolutionResponse( status=ResolutionStatus.MULTIPLE_MATCHES, candidates=candidates, input_echo=input_query, normalization=normalized_query, guidance_text=guidance_text, config=config ) def _create_no_match_response( self, entity_type: str, input_query: str, normalized_query: str, min_score: float ) -> SlotResolutionResponse: """Create a NO_MATCH response.""" error_message = f"No {entity_type} found matching '{input_query}'" return SlotResolutionResponse( status=ResolutionStatus.NO_MATCH, input_echo=input_query, normalization=normalized_query, error=error_message, config={"minScore": min_score} )