Bayesian Self-Escalation in Hierarchical LLM Agents: Key Algorithms and Benchmark Results in implementation

Here is the content formatted as clean, simple plain text suitable for copying into Google Blogs (no special formatting, no backticks, no markdown).

FROM DEEP SEEK AND GEMINI :

Bayesian Self-Escalation in Hierarchical LLM Agents: Key Algorithms and Benchmark Results

Source Information

- Paper Title: Knowing When to Ask for Help: Bayesian Self-Escalation in Hierarchical LLM Agents

- Author: Nadeem Shaikh

- arXiv ID: 2608.24087

- Repository & Code: github.com/nadeem-shaikh/llm-self-escalation (search on Google)

Core Idea in Simple Terms

Current LLM agent systems route tasks in two ways:

1. Static Pre-Routing: Deciding whether a query goes to a small model or a large model before generation starts.

2. Post-Hoc Retries: Generating the full response with a small model first, verifying it afterwards, and retrying with a large model if it fails.

This paper introduces Mid-Generation Bayesian Self-Escalation. The agent monitors its own confidence token-by-token during generation. As soon as its confidence drops below an optimal mathematical threshold, it immediately stops and hands off control to a stronger model—saving compute and avoiding wasted output.

Key Formulas & Mathematical Definitions

1. Token Probability Entropy (Uncertainty Signal)

   H(x_t) = - sum_{i=1}^K P(v_i) * log(P(v_i))

   - H(x_t): Shannon entropy of the vocabulary distribution at generation step t.

   - P(v_i): Probability assigned to token candidate i.

2. Likelihood Function from Uncertainty

   L(e_t) = exp(-alpha * H(x_t))

   - alpha: Learned scaling parameter calibrating raw token entropy to task success likelihood.

3. Bayesian Posterior Update Rule

   B_t = (L(e_t) * B_{t-1}) / (L(e_t) * B_{t-1} + (1 - L(e_t) + epsilon) * (1 - B_{t-1}))

   - B_t: Competence posterior at step t (P(Success | x_{1:t})).

   - B_{t-1}: Competence posterior from the previous token step.

   - epsilon: Numerical stability constant (1e-6).

4. Optimal Stopping Escalation Condition

   Trigger Handoff if B_t < theta_t*

   - theta_t*: Closed-form time-varying threshold derived from optimal-stopping dynamic programming.

5. Finite-Sample Regret Bound

   Regret(n) = O(1 / sqrt(n))

   - n: Number of labeled trajectory sequences used for offline calibration.

Python Code Algorithm

import numpy as np

class BayesianEscalationAgent:

    def __init__(self, weak_model, strong_model, prior=0.85, threshold=0.35, alpha=0.5):

        self.weak = weak_model

        self.strong = strong_model

        self.posterior = prior

        self.threshold = threshold

        self.alpha = alpha

    def compute_token_entropy(self, logit_probs):

        probs = np.array(logit_probs)

        probs = probs / np.sum(probs)

        return -np.sum(probs * np.log(probs + 1e-12))

    def update_posterior(self, entropy):

        likelihood = np.exp(-self.alpha * entropy)

        eps = 1e-6

        numerator = likelihood * self.posterior

        denominator = numerator + ((1.0 - likelihood + eps) * (1.0 - self.posterior))

        self.posterior = numerator / denominator

        return self.posterior

    def generate(self, prompt, max_tokens=100):

        tokens = []

        print("Starting generation with weak model...")

        for step in range(max_tokens):

            token, top_probs = self.weak.generate_next_token(prompt, partial_tokens=tokens)

            entropy = self.compute_token_entropy(top_probs)

            current_posterior = self.update_posterior(entropy)

            if current_posterior < self.threshold:

                print(f"[ESCALATE] Step {step+1}: Posterior {current_posterior:.3f} < Threshold {self.threshold}")

                print("Handing off partial generation to strong model...")

                return self.strong.complete(prompt, partial_context=tokens)

            tokens.append(token)

        return "".join(tokens)

Empirical Benchmark Data & Results

Decision Point:

- Without Paper (Post-Hoc Verification): After 100% tokens generated

- Without Paper (Static Pre-Routing): Before generation starts

- With Paper (Bayesian Self-Escalation): Mid-generation (Token-by-token)

Operational Cost Reduction:

- Without Paper (Post-Hoc Verification): Baseline (0% savings on failed runs)

- Without Paper (Static Pre-Routing): Baseline (High false routing)

- With Paper (Bayesian Self-Escalation): ~34.2% lower total compute cost

Wasted Token Ratio on Failures:

- Without Paper (Post-Hoc Verification): 100% of tokens generated

- Without Paper (Static Pre-Routing): N/A

- With Paper (Bayesian Self-Escalation): < 8% of expected trajectory length

Accuracy at Fixed Budget:

- Without Paper (Post-Hoc Verification): ~76%

- Without Paper (Static Pre-Routing): ~78%

- With Paper (Bayesian Self-Escalation): ~88% (+10% to +12% improvement)

Convergence Rate Guarantee:

- Without Paper (Post-Hoc Verification): None

- Without Paper (Static Pre-Routing): None

- With Paper (Bayesian Self-Escalation): O(1/sqrt(n)) finite-sample rate

Fixing Mid-Model in Standalone Gemini

1. API Logprobs vs. Raw Model Logits

In closed API services like Gemini, full model logit distributions are not exposed. To fix this, estimate token probability distribution entropy using a normalized Top-K Shannon Entropy approximation over top_k logprobs (logprobs=5):

   H_K(x_t) = - sum_{i=1}^K [ p_hat_i * ln(p_hat_i) ]

   where p_hat_i = exp(logprob_i) / sum_{j=1}^K exp(logprob_j)

2. Token Streaming & Early Exit Fix

Standard API calls wait for full completion. Use generate_content_stream to evaluate logprobs token-by-token in real time. Once the calculated competence posterior B_t drops below theta_t*, break the streaming loop immediately to halt execution and stop incurring token costs on gemini-2.5-flash.

Code snippet:

config = types.GenerateContentConfig(

    max_output_tokens=300,

    response_logprobs=True,

    logprobs=5

)

response_stream = client.models.generate_content_stream(

    model="gemini-2.5-flash",

    contents=prompt,

    config=config

)

for chunk in response_stream:

    entropy = compute_top_k_entropy(chunk)

    posterior = update_competence_posterior(posterior, entropy)

    if posterior < THRESHOLD:

        print("Uncertainty spike detected. Halting Flash model.")

        break

3. Preserving State During Escalation (Zero-Token Loss)

Passing only the original prompt to the strong model wastes the work already completed by the weaker model. Use structured prompt stitching or system context injection to pass both the original task and partial response so gemini-3.1-pro can continue seamlessly without restarting from scratch.

Code snippet:

def complete_with_strong_model(original_prompt, partial_text):

    handoff_prompt = (

        f"Task: {original_prompt}\n\n"

        f"Partial Response Generated So Far:\n{partial_text}\n\n"

        f"Instruction: Continue the response cleanly from where it left off."

    )

    response = client.models.generate_content(

        model="gemini-3.1-pro",

        contents=handoff_prompt

    )

    return partial_text + response.text

4. Optimal Parameter Calibration

Uncalibrated raw token entropy triggers false positives during creative or synonym generation. Calibrate scaling parameter alpha and threshold theta_t* offline on validation samples:

- Collect 100-200 execution trajectories on sample tasks.

- Compute average logprob entropy across successful vs. failed runs.

- Fit alpha using logistic regression so L(e_t) = exp(-alpha * H(x_t)) aligns with actual task success rates.

- Set theta_t* dynamically based on remaining budget or expected task length.

Junior Model Implementation in Gemini

Python implementation file: gemini_junior_agent.py

import math

from google import genai

from google.genai import types

class GeminiJuniorAgent:

    def __init__(

        self,

        junior_model: str = "gemini-2.5-flash",

        senior_model: str = "gemini-3.1-pro",

        confidence_threshold: float = 0.35,

        prior_success_rate: float = 0.80

    ):

        self.client = genai.Client()

        self.junior_model = junior_model

        self.senior_model = senior_model

        self.threshold = confidence_threshold

        self.prior = prior_success_rate

    def _calculate_chunk_entropy(self, top_candidates: list) -> float:

        if not top_candidates:

            return 0.0

        probs = [math.exp(candidate.log_probability) for candidate in top_candidates]

        total_p = sum(probs)

        if total_p == 0:

            return 0.0

        entropy = 0.0

        for p in probs:

            p_norm = p / total_p

            if p_norm > 0:

                entropy -= p_norm * math.log(p_norm)

        return entropy

    def _update_posterior(self, current_posterior: float, entropy: float) -> float:

        scaled_uncertainty = math.exp(-0.5 * entropy)

        likelihood_success = scaled_uncertainty

        likelihood_failure = 1.0 - scaled_uncertainty + 1e-6

        num = likelihood_success * current_posterior

        den = num + (likelihood_failure * (1.0 - current_posterior))

        return num / den

    def process_request(self, user_prompt: str, max_tokens: int = 300) -> str:

        config = types.GenerateContentConfig(

            max_output_tokens=max_tokens,

            response_logprobs=True,

            logprobs=5

        )

        posterior = self.prior

        generated_text = ""

        should_escalate = False

        print(f"--- [Junior Model ({self.junior_model})] Executing Task ---")

        stream = self.client.models.generate_content_stream(

            model=self.junior_model,

            contents=user_prompt,

            config=config

        )

        for chunk in stream:

            chunk_text = chunk.text or ""

            generated_text += chunk_text

            if chunk.candidates and chunk.candidates[0].logprobs_result:

                logprob_res = chunk.candidates[0].logprobs_result

                chosen_tokens = logprob_res.chosen_candidates

                top_tokens = logprob_res.top_candidates

                for i, token_info in enumerate(chosen_tokens):

                    top_k_candidates = top_tokens[i].candidates if i < len(top_tokens) else []

                    entropy = self._calculate_chunk_entropy(top_k_candidates)

                    posterior = self._update_posterior(posterior, entropy)

                    if posterior < self.threshold:

                        print(f"\n\n[JUNIOR MODEL UNCERTAINTY SPIKE]")

                        print(f"Confidence score ({posterior:.3f}) fell below threshold ({self.threshold}).")

                        should_escalate = True

                        break

            if should_escalate:

                break

            print(chunk_text, end="", flush=True)

        if should_escalate:

            print(f"\n--- [Escalating to Senior Model ({self.senior_model})] ---")

            

            senior_prompt = (

                f"Task: {user_prompt}\n\n"

                f"Partial Work Completed by Junior Assistant:\n{generated_text}\n\n"

                f"Instruction: Cleanly continue and complete the output from where it stopped."

            )

            senior_response = self.client.models.generate_content(

                model=self.senior_model,

                contents=senior_prompt

            )

            return generated_text + "\n" + senior_response.text

        return generated_text

if __name__ == "__main__":

    agent = GeminiJuniorAgent(

        junior_model="gemini-2.5-flash",

        senior_model="gemini-3.1-pro",

        confidence_threshold=0.35

    )

    prompt = (

        "Draft a high-level overview of a web app, then write a low-level "

        "custom C++ lock-free queue using atomic operations and memory order memory barriers."

    )

    result = agent.process_request(prompt)

    print("\n\n=== Final Output ===")

    print(result)

Core Responsibilities of the Junior Model

1. Filtering Low-Complexity Work: Handles ~70-80% of routine, high-volume tasks directly without consuming senior model API budget.

2. Real-time Entropy Streaming: Evaluates token logprob distributions using generate_content_stream with logprobs=5.

3. Mid-Generation Exit: Immediately halts streaming generation upon encountering an uncertainty spike, avoiding wasted output tokens.

4. Context Preservation: Decodes its partially completed output so the Senior Model can pick up from the exact point of exit without restarting from scratch.