Speculative Execution in AI Agents: Predicting and Pre-Computing Likely Next Steps

What Is Speculative Execution in AI Agents

Speculative execution is a performance optimization borrowed from CPU design. The idea is simple: instead of waiting to know exactly what the next step is, predict the most likely next step and start computing it immediately. If the prediction is correct, you save the entire computation time. If it is wrong, you discard the result and compute the correct one.

In AI agents, this means predicting which tool the agent will call next, what data it will need, or what type of response it will generate — and beginning that work before the LLM has finished deciding.

Predicting the Next Tool Call

Many agent workflows follow predictable patterns. A customer service agent almost always looks up the customer record first. A coding agent usually reads a file before editing it. You can exploit these patterns.

from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class ToolPrediction:
    tool_name: str
    confidence: float
    predicted_args: dict

class ToolPredictor:
    """Predicts the next tool call based on historical patterns."""

    def __init__(self):
        # Tracks: given last tool called, what tool typically follows
        self._transitions: dict[str, dict[str, int]] = defaultdict(
            lambda: defaultdict(int)
        )
        self._total: dict[str, int] = defaultdict(int)

    def record(self, prev_tool: str, next_tool: str):
        self._transitions[prev_tool][next_tool] += 1
        self._total[prev_tool] += 1

    def predict(self, current_tool: str) -> ToolPrediction | None:
        if current_tool not in self._transitions:
            return None

        candidates = self._transitions[current_tool]
        if not candidates:
            return None

        best_tool = max(candidates, key=candidates.get)
        confidence = candidates[best_tool] / self._total[current_tool]

        if confidence < 0.6:
            return None  # Not confident enough

        return ToolPrediction(
            tool_name=best_tool,
            confidence=confidence,
            predicted_args={},
        )

# Usage: after observing many runs, the predictor learns patterns
predictor = ToolPredictor()
predictor.record("search_customer", "get_order_history")
predictor.record("search_customer", "get_order_history")
predictor.record("search_customer", "get_order_history")
predictor.record("search_customer", "update_address")

prediction = predictor.predict("search_customer")
# ToolPrediction(tool_name='get_order_history', confidence=0.75, predicted_args={})

Speculative Tool Execution

Once you have a prediction, you can execute the predicted tool call speculatively — in parallel with the LLM deciding what to do next.

import asyncio
from typing import Any

class SpeculativeExecutor:
    def __init__(self, predictor: ToolPredictor, tool_registry: dict):
        self.predictor = predictor
        self.tools = tool_registry
        self._speculative_results: dict[str, Any] = {}

    async def execute_with_speculation(
        self,
        current_tool: str,
        current_result: Any,
        llm_decision_coro,
    ) -> Any:
        """Run LLM decision and speculative tool call in parallel."""
        prediction = self.predictor.predict(current_tool)

        if prediction and prediction.tool_name in self.tools:
            # Run both in parallel
            llm_task = asyncio.create_task(llm_decision_coro)
            spec_task = asyncio.create_task(
                self.tools[prediction.tool_name](**prediction.predicted_args)
            )

            # Wait for the LLM to decide
            actual_decision = await llm_task

            if actual_decision["tool"] == prediction.tool_name:
                # Prediction was correct — use the speculative result
                result = await spec_task
                return result
            else:
                # Prediction was wrong — cancel speculative work
                spec_task.cancel()
                actual_tool = self.tools[actual_decision["tool"]]
                return await actual_tool(**actual_decision["args"])
        else:
            # No confident prediction — run sequentially
            actual_decision = await llm_decision_coro
            actual_tool = self.tools[actual_decision["tool"]]
            return await actual_tool(**actual_decision["args"])

When the prediction is correct, the tool result is ready instantly because it was computed while the LLM was thinking. When wrong, the overhead is minimal — just a cancelled async task.

Cache Warming

A lighter form of speculation is cache warming: instead of executing the predicted tool call, you warm the caches it will need.

import asyncio
from functools import lru_cache

class CacheWarmer:
    def __init__(self, db_pool, cache):
        self.db = db_pool
        self.cache = cache

    async def warm_for_customer_lookup(self, customer_id: str):
        """Pre-load data that is likely needed after a customer lookup."""
        # Warm in parallel
        await asyncio.gather(
            self._warm_orders(customer_id),
            self._warm_tickets(customer_id),
            self._warm_preferences(customer_id),
        )

    async def _warm_orders(self, customer_id: str):
        key = f"orders:{customer_id}"
        if not await self.cache.exists(key):
            orders = await self.db.fetch(
                "SELECT * FROM orders WHERE customer_id = $1 "
                "ORDER BY created_at DESC LIMIT 10",
                customer_id,
            )
            await self.cache.set(key, orders, ttl=300)

    async def _warm_tickets(self, customer_id: str):
        key = f"tickets:{customer_id}"
        if not await self.cache.exists(key):
            tickets = await self.db.fetch(
                "SELECT * FROM support_tickets WHERE customer_id = $1 "
                "AND status = 'open'",
                customer_id,
            )
            await self.cache.set(key, tickets, ttl=300)

    async def _warm_preferences(self, customer_id: str):
        key = f"prefs:{customer_id}"
        if not await self.cache.exists(key):
            prefs = await self.db.fetchrow(
                "SELECT * FROM customer_preferences WHERE customer_id = $1",
                customer_id,
            )
            await self.cache.set(key, prefs, ttl=600)

Cache warming is safer than speculative execution because it has no side effects. Even if the prediction is wrong, the cached data may be useful later.

Rollback Strategies for Failed Speculation

Speculative execution of tools with side effects (writing to a database, sending emails) requires careful rollback handling.

class SafeSpeculativeExecutor:
    """Only speculates on read-only tools. Write tools run after confirmation."""

    READ_ONLY_TOOLS = {"search", "lookup", "get", "list", "fetch"}

    def is_safe_to_speculate(self, tool_name: str) -> bool:
        return any(tool_name.startswith(prefix) for prefix in self.READ_ONLY_TOOLS)

    async def execute(self, prediction: ToolPrediction, tools: dict):
        if self.is_safe_to_speculate(prediction.tool_name):
            return await tools[prediction.tool_name](**prediction.predicted_args)
        else:
            # Never speculate on write operations
            return None

The golden rule: only speculate on read-only operations. Never speculatively send an email, update a database record, or call a third-party API with side effects.

Measuring Speculation Effectiveness

Track hit rates and latency savings to validate your speculation strategy.

import time
from dataclasses import dataclass, field

@dataclass
class SpeculationMetrics:
    total_predictions: int = 0
    correct_predictions: int = 0
    total_latency_saved_ms: float = 0
    total_wasted_compute_ms: float = 0

    @property
    def hit_rate(self) -> float:
        if self.total_predictions == 0:
            return 0.0
        return self.correct_predictions / self.total_predictions

    @property
    def net_savings_ms(self) -> float:
        return self.total_latency_saved_ms - self.total_wasted_compute_ms

    def report(self) -> str:
        return (
            f"Hit rate: {self.hit_rate:.1%} | "
            f"Net savings: {self.net_savings_ms:.0f}ms | "
            f"Predictions: {self.total_predictions}"
        )

A hit rate above 60% typically means speculation is net-positive for latency. Below 40%, the wasted compute may outweigh the savings.

FAQ

Is speculative execution worth the extra complexity?

It depends on two factors: how predictable your agent workflows are and how latency-sensitive your use case is. For customer service agents with well-defined flows (lookup customer, check orders, resolve issue), speculation can cut perceived latency by 30-50%. For open-ended creative agents, workflows are too unpredictable to benefit.

How do I handle speculative execution with rate-limited APIs?

Count speculative calls against your rate limit budget. If you are near the limit, disable speculation and run sequentially. A good approach is to reserve 20% of your rate limit budget for speculative calls and disable speculation when that budget is exhausted.

Can I use speculative execution with streaming responses?

Yes, but it requires careful coordination. Start streaming the speculative result to the client, but be prepared to interrupt and switch to the correct result if speculation was wrong. This is complex to implement correctly and is usually only worth it for the highest-traffic agents.

---

#SpeculativeExecution #Prediction #Caching #Latency #Python #AgenticAI #LearnAI #AIEngineering