Speculative Execution for High‑Reliability AI Agents

Predictive (Speculative) Execution Pattern

Predictive execution launches a tool call that is likely to be needed before the LLM finishes its reasoning, allowing the tool latency to overlap with LLM latency. If the LLM later selects the same tool, the result is already available; otherwise the speculative call is discarded.

Simulated slow tool

from langchain_core.tools import tool
import time, json

DATABASE_LATENCY_SECONDS = 3

@tool
def get_order_history(user_id: str) -> str:
    """A simulated slow tool that fetches the order history for a given user from a database."""
    print(f"--- [DATABASE] Starting query for user_id: {user_id}. This will take {DATABASE_LATENCY_SECONDS} seconds. ---")
    time.sleep(DATABASE_LATENCY_SECONDS)
    mock_db = {
        "user123": [
            {"order_id": "A123", "item": "QuantumLeap AI Processor", "status": "Shipped"},
            {"order_id": "B456", "item": "Smart Coffee Mug", "status": "Delivered"}
        ]
    }
    result = mock_db.get(user_id, [])
    print(f"--- [DATABASE] Query finished for user_id: {user_id}. ---")
    return json.dumps(result)

The tool introduces a predictable 3‑second latency that can be hidden by speculative execution.

State definition

from typing import TypedDict, Annotated, List, Optional
from langchain_core.messages import BaseMessage
import operator
from concurrent.futures import Future

class GraphState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    user_id: str
    # Holds the Future for the speculative pre‑fetch
    prefetched_data: Optional[Future]
    agent_decision: Optional[BaseMessage]
    performance_log: Annotated[List[str], operator.add]

The prefetched_data field stores a Future representing the background tool call.

Entry‑point node (orchestrator)

from concurrent.futures import ThreadPoolExecutor
import time

thread_pool = ThreadPoolExecutor(max_workers=5)

def entry_point(state: GraphState):
    """Starts the speculative pre‑fetch and the main LLM reasoning in parallel."""
    print("--- [ORCHESTRATOR] Entry point started. --- ")
    start_time = time.time()
    print("--- [ORCHESTRATOR] Starting speculative pre‑fetch of order history... ---")
    prefetched_data_future = thread_pool.submit(
        get_order_history.invoke, {"user_id": state['user_id']}
    )
    print("--- [ORCHESTRATOR] Starting main agent LLM call... ---")
    agent_response = llm_with_tools.invoke(state['messages'])
    execution_time = time.time() - start_time
    log_entry = f"[Orchestrator] LLM reasoning completed in {execution_time:.2f}s."
    print(log_entry)
    return {
        "prefetched_data": prefetched_data_future,
        "agent_decision": agent_response,
        "performance_log": [log_entry]
    }

The thread_pool.submit() call returns immediately with a Future, allowing the LLM call to proceed without waiting for the 3‑second database query.

Tool executor node

from langchain_core.messages import ToolMessage
import time

def tool_executor_node(state: GraphState):
    """Executes the chosen tool, using the pre‑fetched result when possible."""
    print("--- [TOOL EXECUTOR] Node started. --- ")
    start_time = time.time()
    agent_decision = state['agent_decision']
    tool_call = agent_decision.tool_calls[0]
    if tool_call['name'] == "get_order_history":
        print("--- [TOOL EXECUTOR] Agent wants order history. Checking pre‑fetch... ---")
        prefetched_future = state['prefetched_data']
        tool_result = prefetched_future.result()  # Returns instantly if background task finished
        print("--- [TOOL EXECUTOR] Pre‑fetch successful! Using cached data instantly. ---")
    else:
        print(f"--- [TOOL EXECUTOR] Speculation failed. Agent wants {tool_call['name']}. Executing normally. ---")
        tool_result = "Tool not implemented for this demo."
    tool_message = ToolMessage(content=tool_result, tool_call_id=tool_call['id'])
    execution_time = time.time() - start_time
    log_entry = f"[ToolExecutor] Resolved tool call in {execution_time:.2f}s."
    print(log_entry)
    return {"messages": [agent_decision, tool_message], "performance_log": [log_entry]}

If the speculative tool matches the LLM’s decision, prefetched_future.result() returns the cached data in ~0.01 s; otherwise a normal tool execution path would be taken.

Workflow assembly

from langgraph.graph import StateGraph, END

def should_call_tool(state: GraphState) -> str:
    if state['agent_decision'].tool_calls:
        return "execute_tool"
    return END

workflow = StateGraph(GraphState)
workflow.add_node("entry_point", entry_point)
workflow.add_node("execute_tool", tool_executor_node)
workflow.add_node("final_answer", final_answer_node)  # assumed defined elsewhere
workflow.set_entry_point("entry_point")
workflow.add_conditional_edges("entry_point", should_call_tool)
workflow.add_edge("execute_tool", "final_answer")
workflow.add_edge("final_answer", END)
app = workflow.compile()

Performance measurement

The workflow is run with a sample user query ( user_id = "user123"). Timing logs are extracted from the state after execution.

# Extract timings (seconds) from performance_log entries
llm_time_1 = float(final_state['performance_log'][0].split(' ')[-2])
resolution_time = float(final_state['performance_log'][1].split(' ')[-2])
llm_time_2 = float(final_state['performance_log'][2].split(' ')[-2])

db_time = DATABASE_LATENCY_SECONDS

speculative_total = llm_time_1 + resolution_time + llm_time_2
sequential_total = llm_time_1 + db_time + llm_time_2

time_saved = sequential_total - speculative_total
reduction_percent = (time_saved / sequential_total) * 100

Observed values:

LLM call 1: 4.21 s

Database query (speculative): 3.00 s (hidden)

Tool‑result resolution: 0.01 s (cache hit)

LLM call 2: 3.55 s

Resulting totals:

Speculative workflow total: 7.77 s

Sequential (simulated) total: 10.76 s

Time saved: 2.99 s

Perceived latency reduction: ≈28 %

Conclusion

By overlapping a predictable 3‑second tool call with the first LLM reasoning step, predictive execution hides the tool latency entirely. The subsequent tool‑executor node resolves the result in ~0.01 s, yielding a 28 % reduction in end‑to‑end latency for this demonstration.

Source code and full experiment are available in the GitHub repository:

https://github.com/FareedKhan-dev/agentic-parallelism