[IA Series 8/n] Building a Self-Reflection LLM Agent: From Theory to Proof of Concept

ConvergenceAnalysis: {
    confidence_evolution: [0.5, 0.67, 0.75, 0.8, 0.8],
    entropy_evolution: [1.0, 0.92, 0.81, 0.72, 0.72],
    convergence_rate: 0.06,          # Confidence increasing by 6% per response
    final_stability: 1.0,            # Perfectly stable final confidence
    entropy_convergence_rate: -0.056, # Entropy decreasing by 5.6% per response
    entropy_final_stability: 1.0     # Perfectly stable final entropy
}

function SELF-REFLECTION-AGENT(percept) returns an action
    persistent: state, agent state with entropy intelligence
                responses, collection of (reasoning, answer) pairs
                question, current question to answer
                confidence_threshold, stopping threshold (default 0.8)
                entropy_threshold, entropy stopping threshold (default 0.3)
                entropy_weight, weight in combined scoring (default 0.3)
                entropy_mode, stopping mode (default "combined")
                min_responses, minimum responses required (default 5)
                min_entropy_samples, minimum samples for entropy (default 4)
                max_responses, maximum responses allowed (default 10)

    # Update state with new percept
    if percept contains question:
        state.question ← question
        state.responses ← []
        return QUERY-LLM(state.question)
    elif percept contains (reasoning, answer):
        state.responses.append((reasoning, answer))
        
        # Decision: Continue querying or provide final answer?
        if should_continue_querying():
            return QUERY-LLM(state.question)
        else:
            # Perform all internal analysis and return final result
            final_result ← perform_complete_internal_analysis()
            return REPLY-TO-USER(final_result)

function should_continue_querying() returns boolean
    current_responses ← length(state.responses)
    
    # Must have minimum responses
    if current_responses < state.min_responses:
        return true
    
    # Must not exceed maximum responses
    if current_responses ≥ state.max_responses:
        return false
    
    # Perform internal analysis to make stopping decision
    perform_internal_analysis()
    
    # Apply stopping logic based on entropy mode
    return not evaluate_stopping_criteria()

function perform_internal_analysis()
    # Internal computation: Calculate probability distribution
    answers ← extract_answers_from(state.responses)
    count ← {}
    for answer in answers:
        count[answer] ← count.get(answer, 0) + 1
    
    total ← length(answers)
    state.distribution ← {}
    for answer, freq in count:
        state.distribution[answer] ← freq / total
    
    # Internal computation: Assess confidence
    state.confidence ← max(state.distribution.values())
    
    # Internal computation: Calculate entropy
    state.entropy ← 0
    for probability in state.distribution.values():
        if probability > 0:
            state.entropy ← state.entropy - (probability × log₂(probability))
    
    # Internal computation: Calculate normalized entropy
    if length(state.distribution) ≤ 1:
        state.normalized_entropy ← 0.0
    else:
        max_entropy ← log₂(length(state.distribution))
        state.normalized_entropy ← state.entropy / max_entropy if max_entropy > 0 else 0.0
    
    # Internal computation: Classify entropy level
    if state.normalized_entropy ≤ 0.2:
        state.entropy_level ← "concentrated"
    elif state.normalized_entropy ≤ 0.7:
        state.entropy_level ← "scattered"
    else:
        state.entropy_level ← "uniform"
    
    # Internal computation: Classify consensus type
    state.consensus_type ← classify_consensus_type(state.distribution)
    
    # Internal computation: Assess convergence
    state.convergence_analysis ← assess_convergence_evolution()

function classify_consensus_type(distribution) returns consensus_type
    if distribution is empty:
        return "undefined"
    
    probabilities ← sort(distribution.values(), descending=true)
    max_prob ← probabilities[0]
    
    # Binary split: Two main answers roughly equal (check this first)
    if length(probabilities) ≥ 2 and probabilities[1] ≥ 0.35:
        if abs(probabilities[0] - probabilities[1]) ≤ 0.15:
            return "binary"
    
    # Strong consensus: One answer dominates significantly (80%+)
    if max_prob ≥ 0.8:
        return "strong"
    
    # Emerging consensus: One answer leading but not dominant (40-79%)
    if max_prob ≥ 0.4:
        return "emerging"
    
    # Divided: No clear leader (under 40%)
    return "divided"

function assess_convergence_evolution() returns convergence_analysis
    if length(state.responses) < 2:
        return {
            confidence_evolution: [state.confidence] if state.responses else [],
            entropy_evolution: [state.normalized_entropy] if state.responses else [],
            convergence_rate: 0.0,
            final_stability: 1.0,
            entropy_convergence_rate: 0.0,
            entropy_final_stability: 1.0
        }
    
    confidences_over_time ← []
    entropies_over_time ← []
    
    # Calculate confidence and entropy evolution
    for i ← 1 to length(state.responses):
        subset_responses ← state.responses[1:i]
        subset_answers ← extract_answers_from(subset_responses)
        subset_counts ← count_occurrences(subset_answers)
        subset_total ← length(subset_answers)
        
        # Calculate confidence for subset
        if subset_total > 0:
            max_count ← max(subset_counts.values())
            confidence ← max_count / subset_total
        else:
            confidence ← 0.0
        
        # Calculate entropy for subset
        if subset_total > 0:
            subset_distribution ← {}
            for answer, count in subset_counts:
                subset_distribution[answer] ← count / subset_total
            
            entropy ← calculate_entropy(subset_distribution)
            if length(subset_distribution) > 1:
                normalized_entropy ← entropy / log₂(length(subset_distribution))
            else:
                normalized_entropy ← 0.0
        else:
            entropy ← 0.0
            normalized_entropy ← 0.0
        
        confidences_over_time.append(confidence)
        entropies_over_time.append(normalized_entropy)
    
    return {
        confidence_evolution: confidences_over_time,
        entropy_evolution: entropies_over_time,
        convergence_rate: calculate_convergence_rate(confidences_over_time),
        final_stability: assess_stability(confidences_over_time),
        entropy_convergence_rate: calculate_entropy_convergence_rate(entropies_over_time),
        entropy_final_stability: assess_entropy_stability(entropies_over_time)
    }

function evaluate_stopping_criteria() returns boolean
    current_responses ← length(state.responses)
    
    # Handle different entropy modes
    if state.entropy_mode = "off" or state.entropy_mode = "confidence_only":
        # Traditional confidence-only stopping
        return state.confidence ≥ state.confidence_threshold
    
    # Need minimum samples for entropy to be meaningful
    if current_responses < state.min_entropy_samples:
        return state.confidence ≥ state.confidence_threshold
    
    if state.entropy_mode = "entropy_only":
        # Stop only based on entropy (low entropy = concentrated)
        return state.normalized_entropy ≤ state.entropy_threshold
    
    elif state.entropy_mode = "combined":
        # Combined scoring: balance confidence and entropy
        
        # High confidence overrides entropy concerns
        if state.confidence ≥ 0.9:
            return true
        
        # Check confidence threshold first
        if state.confidence ≥ state.confidence_threshold:
            # High confidence + low entropy = strong consensus, stop early
            if state.normalized_entropy ≤ state.entropy_threshold:
                return true
            # High confidence + high entropy = check if really confident
            elif state.confidence ≥ 0.8:
                return true
        
        # Calculate combined score: confidence weighted by entropy concentration
        entropy_factor ← 1.0 - (state.entropy_weight × state.normalized_entropy)
        combined_score ← state.confidence × entropy_factor
        
        # Use a slightly lower threshold for combined scoring
        return combined_score ≥ (state.confidence_threshold × 0.9)
    
    # Default fallback
    return state.confidence ≥ state.confidence_threshold

function perform_complete_internal_analysis() returns reflection_result
    # Ensure all internal analysis is complete
    perform_internal_analysis()
    
    # Internal computation: Synthesize final reflection result
    final_answer ← argmax(state.distribution)
    
    reflection_result ← {
        final_answer: final_answer,
        consensus_confidence: state.confidence,
        answer_distribution: state.distribution,
        uncertainty_level: categorize_uncertainty(state.confidence),
        early_stopping: length(state.responses) < state.max_responses,
        total_responses: length(state.responses),
        convergence_analysis: state.convergence_analysis,
        distribution_entropy: state.entropy,
        normalized_entropy: state.normalized_entropy,
        entropy_level: state.entropy_level,
        consensus_type: state.consensus_type
    }
    
    return reflection_result

# External Actions (only these interact with environment)
function QUERY-LLM(question) returns response
    prompt ← create_chain_of_thought_prompt(question)
    response ← send_to_llm(prompt)
    return response

function REPLY-TO-USER(reflection_result) returns formatted_response
    formatted_response ← format_reflection_response(reflection_result)
    return formatted_response

# Helper Functions for Internal Computations
function categorize_uncertainty(confidence) returns uncertainty_level
    if confidence ≥ 0.8:
        return "low"
    elif confidence ≥ 0.6:
        return "medium"
    else:
        return "high"

function calculate_convergence_rate(confidences) returns rate
    if length(confidences) < 2:
        return 0.0
    return (confidences[-1] - confidences[0]) / length(confidences)

function assess_stability(confidences) returns stability
    if length(confidences) < 3:
        return 1.0
    last_three ← confidences[-3:]
    return 1.0 - (max(last_three) - min(last_three))

function calculate_entropy_convergence_rate(entropies) returns rate
    if length(entropies) < 2:
        return 0.0
    return (entropies[-1] - entropies[0]) / length(entropies)

function assess_entropy_stability(entropies) returns stability
    if length(entropies) < 3:
        return 1.0
    last_three ← entropies[-3:]
    return 1.0 - (max(last_three) - min(last_three))

function calculate_entropy(distribution) returns entropy
    H ← 0
    for probability in distribution.values():
        if probability > 0:
            H ← H - (probability × log₂(probability))
    return H

function extract_answers_from(responses) returns answers
    answers ← []
    for (reasoning, answer) in responses:
        answers.append(answer)
    return answers

function count_occurrences(answers) returns counts
    counts ← {}
    for answer in answers:
        counts[answer] ← counts.get(answer, 0) + 1
    return counts

function argmax(distribution) returns max_key
    max_prob ← 0
    max_key ← ""
    for key, prob in distribution:
        if prob > max_prob:
            max_prob ← prob
            max_key ← key
    return max_key

function create_chain_of_thought_prompt(question) returns prompt
    return "Please think step by step and provide your reasoning: " + question

function format_reflection_response(result) returns formatted_response
    return "Final Answer: " + result.final_answer + 
           " (Confidence: " + result.consensus_confidence + 
           ", Uncertainty: " + result.uncertainty_level + ")"

@dataclass(frozen=True)
class LLMResponse:
    """Immutable Domain entity representing a single LLM response."""
    reasoning: str
    answer: str
    timestamp: datetime  # For convergence analysis
    
@dataclass(frozen=True)
class ConvergenceAnalysis:
    """Immutable analysis of response evolution."""
    confidence_evolution: List[float]
    entropy_evolution: List[float]
    convergence_rate: float
    final_stability: float
    entropy_convergence_rate: float
    entropy_final_stability: float

@dataclass(frozen=True)
class ReflectionResult:
    """Comprehensive result object with entropy intelligence."""
    final_answer: str
    consensus_confidence: float
    answer_distribution: Dict[str, float]
    uncertainty_level: str
    early_stopping: bool
    total_responses: int
    convergence_analysis: ConvergenceAnalysis
    distribution_entropy: float
    normalized_entropy: float
    entropy_level: str
    consensus_type: str

@dataclass
class ReflectionConfig:
    """Configuration for entropy-based self-reflection."""
    llm_interface: LLMInterface
    confidence_threshold: float = 0.8
    entropy_threshold: float = 0.3
    entropy_weight: float = 0.3
    entropy_mode: str = "combined"  # "off", "confidence_only", "entropy_only", "combined"
    min_responses: int = 5
    min_entropy_samples: int = 4
    max_responses: int = 10
    prompt_template: str = ""

@dataclass
class AgentState:
    """Comprehensive state tracking for self-reflection."""
    question: str
    responses: List[LLMResponse]
    distribution: Dict[str, float]
    confidence: float
    entropy: float
    normalized_entropy: float
    entropy_level: str
    consensus_type: str
    convergence_analysis: ConvergenceAnalysis

class LLMInterface(ABC):
    """Abstract interface for LLM interactions with entropy support."""
    
    @abstractmethod
    async def generate_response(self, prompt: str, question: str) -> LLMResponse:
        """Generate a single LLM response for entropy analysis."""
        pass
    
    @abstractmethod
    async def generate_batch_responses(self, prompt: str, question: str, count: int) -> List[LLMResponse]:
        """Generate multiple responses for parallel processing."""
        pass

class EnhancedLiteLLMAdapter(LLMInterface):
    """Enhanced LiteLLM adapter with entropy-optimized parameters."""
    
    def __init__(self, model: str, temperature: float = 0.7, **kwargs):
        self.model = model
        self.temperature = temperature
        self.entropy_optimized = kwargs.get('entropy_optimized', True)
        self.kwargs = kwargs

class EntropyCalculator:
    """Service for entropy calculations and analysis."""
    
    @staticmethod
    def calculate_shannon_entropy(distribution: Dict[str, float]) -> float:
        """Calculate Shannon entropy for answer distribution."""
        pass
    
    @staticmethod
    def normalize_entropy(entropy: float, unique_answers: int) -> float:
        """Normalize entropy to [0,1] range."""
        pass
    
    @staticmethod
    def classify_entropy_level(normalized_entropy: float) -> str:
        """Classify entropy as concentrated/scattered/uniform."""
        pass

class ConsensusClassifier:
    """Service for consensus pattern recognition."""
    
    @staticmethod
    def classify_consensus(distribution: Dict[str, float]) -> str:
        """Classify consensus type with binary split detection."""
        pass
    
    @staticmethod
    def detect_binary_split(distribution: Dict[str, float]) -> bool:
        """Detect binary consensus patterns."""
        pass

class SelfReflectionAgent:
    """Main agent implementing entropy-based self-reflection."""
    
    def __init__(self, config: ReflectionConfig, question: str):
        self._config = config
        self._state = AgentState(question=question, responses=[], ...)
        self._entropy_calculator = EntropyCalculator()
        self._consensus_classifier = ConsensusClassifier()
        self._convergence_analyzer = ConvergenceAnalyzer()
    
    def process_question(self) -> ReflectionResult:
        """Main processing loop with entropy-based stopping."""
        while self._should_continue_querying():
            response = await self._query_llm()
            self._state.responses.append(response)
            self._update_internal_state()
        
        return self._synthesize_final_result()
    
    def _should_continue_querying(self) -> bool:
        """Multi-mode stopping decision with entropy intelligence."""
        if len(self._state.responses) < self._config.min_responses:
            return True
        
        if len(self._state.responses) >= self._config.max_responses:
            return False
        
        return not self._evaluate_stopping_criteria()
    
    def _evaluate_stopping_criteria(self) -> bool:
        """Stopping logic with four entropy modes."""
        # Implementation of entropy modes: off, confidence_only, entropy_only, combined
        pass
    
    def _update_internal_state(self):
        """Update all internal state with latest response."""
        self._calculate_distribution()
        self._assess_confidence()
        self._calculate_entropy()
        self._classify_consensus()
        self._assess_convergence()
    
    def _calculate_distribution(self):
        """Calculate probability distribution - O(m) complexity."""
        answers = [response.answer for response in self._state.responses]
        counts = Counter(answers)
        total = len(answers)
        self._state.distribution = {answer: count/total for answer, count in counts.items()}
    
    def _assess_confidence(self):
        """Assess consensus confidence - O(k) complexity."""
        self._state.confidence = max(self._state.distribution.values()) if self._state.distribution else 0.0
    
    def _calculate_entropy(self):
        """Calculate Shannon and normalized entropy - O(k) complexity."""
        self._state.entropy = self._entropy_calculator.calculate_shannon_entropy(self._state.distribution)
        unique_answers = len(self._state.distribution)
        self._state.normalized_entropy = self._entropy_calculator.normalize_entropy(
            self._state.entropy, unique_answers
        )
        self._state.entropy_level = self._entropy_calculator.classify_entropy_level(
            self._state.normalized_entropy
        )
    
    def _classify_consensus(self):
        """Classify consensus type - O(k log k) complexity."""
        self._state.consensus_type = self._consensus_classifier.classify_consensus(
            self._state.distribution
        )
    
    def _assess_convergence(self):
        """Assess convergence evolution - O(m) complexity."""
        self._state.convergence_analysis = self._convergence_analyzer.analyze_evolution(
            self._state.responses
        )

class ConvergenceAnalyzer:
    """Service for analyzing response convergence patterns."""
    
    def analyze_evolution(self, responses: List[LLMResponse]) -> ConvergenceAnalysis:
        """Analyze dual-track confidence and entropy evolution."""
        if len(responses) < 2:
            return self._create_minimal_analysis(responses)
        
        confidences = []
        entropies = []
        
        for i in range(1, len(responses) + 1):
            subset = responses[:i]
            confidence, entropy = self._calculate_subset_metrics(subset)
            confidences.append(confidence)
            entropies.append(entropy)
        
        return ConvergenceAnalysis(
            confidence_evolution=confidences,
            entropy_evolution=entropies,
            convergence_rate=self._calculate_convergence_rate(confidences),
            final_stability=self._assess_stability(confidences),
            entropy_convergence_rate=self._calculate_convergence_rate(entropies),
            entropy_final_stability=self._assess_stability(entropies)
        )

class StoppingCriteriaEvaluator:
    """Evaluator for multi-mode stopping decisions."""
    
    def __init__(self, config: ReflectionConfig):
        self._config = config
    
    def evaluate(self, state: AgentState) -> bool:
        """Evaluate stopping criteria based on entropy mode."""
        handlers = {
            "off": self._evaluate_off_mode,
            "confidence_only": self._evaluate_confidence_only,
            "entropy_only": self._evaluate_entropy_only,
            "combined": self._evaluate_combined_mode
        }
        
        handler = handlers.get(self._config.entropy_mode, self._evaluate_combined_mode)
        return handler(state)
    
    def _evaluate_combined_mode(self, state: AgentState) -> bool:
        """Combined scoring with entropy weighting."""
        # High confidence override
        if state.confidence >= 0.9:
            return True
        
        # Dual-threshold check
        if (state.confidence >= self._config.confidence_threshold and 
            state.normalized_entropy <= self._config.entropy_threshold):
            return True
        
        # Combined scoring
        entropy_factor = 1.0 - (self._config.entropy_weight * state.normalized_entropy)
        combined_score = state.confidence * entropy_factor
        
        return combined_score >= (self._config.confidence_threshold * 0.9)