Streaming, Speculative Execution, & Latency Engineering
Agents are slow by nature. A single turn involves at least one model call, possibly several tool invocations, and then another model call to synthesize the results. Even with fast models, that chain of work can take five, ten, or thirty seconds — and the user sees none of it until it is done. For a chatbot, that is annoying. For a coding agent, a research assistant, or anything embedded in a real-time workflow, it is a deal-breaker.
Latency engineering is the discipline of making agents feel fast even when the underlying work takes time. Some of it is about raw speed: reducing the number of serial steps, caching results, picking faster models for simple sub-tasks. But much of it is about perception: streaming output progressively, starting work before the user finishes asking, and surfacing intermediate results so the user never stares at a blank screen.
Agent Latency #
Before you can optimize anything, you need to know where the time goes. The latency profile of an agent call is nothing like a simple API request.
Timeline of a multi-step agent turn:
t=0ms ──── User sends request
t=50ms ──── Context assembled (memory retrieval, tool list injected)
t=100ms ──── LLM call #1 begins (reasoning / plan)
t=800ms ──── LLM first token arrives (Time to First Token, TTFT)
t=2,100ms ─── LLM call #1 complete (final tool call emitted)
t=2,150ms ─── Tool call #1 begins (external API)
t=3,900ms ─── Tool call #1 returns
t=4,000ms ─── LLM call #2 begins (synthesis)
t=4,700ms ─── LLM first token arrives (TTFT for second call)
t=6,400ms ─── Final response complete
Total wall-clock time: 6.4 seconds
Time user waited for any visible output: 4.7 seconds
A few things stand out. First, serial LLM calls are brutal. Each one adds both a time-to-first-token delay and a generation time. An agent that makes three LLM calls in sequence will have triple the TTFT overhead before the user sees anything from the final answer. Second, tool calls dominate middle latency. External APIs are often the longest single delay — and unlike LLM calls, they produce no visible output during execution. Third, perception latency is worse than total latency. The user does not feel the 6.4 seconds evenly; they feel 4.7 seconds of complete silence, then words start appearing.
The levers you have are:
- TTFT reduction — getting visible tokens to the user faster
- Tool call overlap — running tool calls in parallel instead of sequentially
- Speculative prefetch — starting tool calls before the model explicitly requests them
- Streaming intermediate output — showing partial results during long operations
- Cancellation — killing stale work when the user changes direction
Streaming - The Baseline Fix #
The single highest-leverage thing you can do is stream the model's output token by token rather than waiting for the full generation. Most model APIs support this natively via server-sent events or chunked HTTP responses. The result is that the user sees the first word within the TTFT window (~700ms in the example above) rather than waiting for the entire generation to finish.
import asyncio
from collections.abc import AsyncIterator
async def stream_agent_response(
messages: list[dict],
model_client,
) -> AsyncIterator[str]:
"""
Stream tokens from a model call as they arrive,
yielding each text chunk to the caller.
"""
async with model_client.stream(messages=messages) as stream:
async for chunk in stream:
if chunk.type == "content_block_delta":
yield chunk.delta.text
elif chunk.type == "tool_use":
# Tool calls are not text — signal the caller separately
yield f"\n[calling {chunk.name}...]\n"
The real engineering challenge starts when the agent makes tool calls mid-generation. The model streams some reasoning text, then emits a tool call. At that point you have a choice: wait silently while the tool runs, or show the user that work is happening. The better approach is to emit a structured progress signal.
Streaming Tool Call Status #
Modern frontends can render a "thinking" or "working" state distinct from response text. Your streaming layer should emit typed events rather than raw text so the UI can handle each appropriately.
from dataclasses import dataclass
from enum import Enum
import json
from typing import Any
class EventType(Enum):
TEXT_DELTA = "text_delta"
TOOL_START = "tool_start"
TOOL_RESULT = "tool_result"
TOOL_ERROR = "tool_error"
DONE = "done"
@dataclass
class StreamEvent:
type: EventType
payload: Any
async def agent_event_stream(
messages: list[dict],
model_client,
tool_executor,
) -> AsyncIterator[StreamEvent]:
"""
Stream a complete agent turn as typed events:
text deltas, tool call lifecycle, and completion.
"""
accumulated_text = ""
async with model_client.stream(messages=messages) as stream:
current_tool_call = None
async for chunk in stream:
if chunk.type == "content_block_delta":
accumulated_text += chunk.delta.text
yield StreamEvent(type=EventType.TEXT_DELTA, payload=chunk.delta.text)
elif chunk.type == "tool_use_start":
current_tool_call = {"name": chunk.name, "input": ""}
yield StreamEvent(
type=EventType.TOOL_START,
payload={"tool": chunk.name, "id": chunk.id},
)
elif chunk.type == "tool_use_delta":
# Tool input arguments are streamed as JSON fragments
if current_tool_call:
current_tool_call["input"] += chunk.partial_json
elif chunk.type == "tool_use_end":
if current_tool_call:
# Execute the tool and stream the result back
try:
tool_input = json.loads(current_tool_call["input"] or "{}")
result = await tool_executor.run(
name=current_tool_call["name"],
input=tool_input,
)
yield StreamEvent(type=EventType.TOOL_RESULT, payload=result)
except Exception as error:
yield StreamEvent(
type=EventType.TOOL_ERROR,
payload={
"tool": current_tool_call["name"],
"error": str(error),
},
)
current_tool_call = None
yield StreamEvent(type=EventType.DONE, payload=None)
This gives the frontend enough information to render something like: "Searching for flights... found 3 results. Fetching baggage policies..." — instead of a spinner and silence.
Speculative Execution #
Streaming makes existing work visible sooner. Speculative execution makes work finish sooner by starting it before you are certain it is needed.
The core insight: in many agent workflows, the next tool call is highly predictable from the current context, even before the model has finished generating its current reasoning step. If you can identify that with high confidence, you can start the tool call in parallel with the remaining generation — and by the time the model formally emits the tool call, the result is already waiting.
Without speculation:
LLM generates reasoning... ── 1,500ms
LLM emits tool call ── 100ms
Tool call executes ── 2,000ms
LLM synthesizes result ── 800ms
─────────────────────────────────── 4,400ms total
With speculative prefetch (tool predicted at 800ms into generation):
LLM generates reasoning... ── 1,500ms
│ ▲ prefetch starts at 800ms
└─ Tool call executes in parallel ── 2,000ms (starts at 800ms)
LLM synthesizes result ── 800ms (starts at 1,500ms, tool
already done at 2,800ms)
─────────────────────────────────── 2,300ms total (~48% faster)
Predictive Tool Call Triggering #
The simplest form of speculative execution is pattern-based: if the agent almost always calls tool B after tool A, you can start tool B as soon as A returns, in parallel with the model reasoning about A's result.
import asyncio
class SpeculativeExecutor:
"""
Runs tool calls speculatively based on learned call-graph patterns.
Tool results are cached until formally requested by the model.
"""
def __init__(self, tool_executor, call_graph: dict[str, list[str]]):
# call_graph: { "tool_a": ["tool_b", "tool_c"] }
# means: when tool_a completes, speculatively start tool_b and tool_c
self.tool_executor = tool_executor
self.call_graph = call_graph
self._speculative_results: dict[str, asyncio.Task] = {}
async def execute(self, tool_name: str, tool_input: dict) -> dict:
"""
Execute a tool. If a speculative result is waiting, return it.
Otherwise run the tool normally.
"""
cache_key = self._cache_key(tool_name, tool_input)
if cache_key in self._speculative_results:
task = self._speculative_results.pop(cache_key)
result = await task # Result may already be ready
self._prefetch_successors(tool_name, result)
return result
result = await self.tool_executor.run(tool_name, tool_input)
self._prefetch_successors(tool_name, result)
return result
def _prefetch_successors(self, completed_tool: str, result: dict) -> None:
"""
Start speculative calls for tools likely to follow this one.
"""
successors = self.call_graph.get(completed_tool, [])
for successor_name in successors:
predicted_input = self._predict_input(successor_name, result)
if predicted_input is not None:
key = self._cache_key(successor_name, predicted_input)
if key not in self._speculative_results:
task = asyncio.create_task(
self.tool_executor.run(successor_name, predicted_input)
)
self._speculative_results[key] = task
def _predict_input(self, tool_name: str, previous_result: dict) -> dict | None:
"""
Derive the predicted input for a speculative call from the
previous tool's output. Returns None if prediction is not possible.
Override per domain.
"""
return None # Subclass or inject domain-specific logic
def _cache_key(self, tool_name: str, tool_input: dict) -> str:
import json
return f"{tool_name}:{json.dumps(tool_input, sort_keys=True)}"
The hard part is _predict_input. For some domains it is trivial — if you just fetched a flight list, you know the follow-up will fetch baggage policies for each airline in the result. For others it is impossible without waiting for the model's reasoning. The trick is to apply speculation only where the prediction confidence is high, and accept the waste of a cancelled speculative call when you are wrong.
Speculative Branching #
A more aggressive form of speculation is to run multiple paths of the agent simultaneously, then keep the winner and discard the rest. This is useful when:
- You are uncertain which of two planning strategies will succeed
- A multi-step task has an early branch point and both paths are cheap to start
- You want to race a fast approximate answer against a slow accurate one
import asyncio
from dataclasses import dataclass
from typing import Callable, Awaitable
@dataclass
class Branch:
name: str
task: Callable[[], Awaitable[dict]]
priority: int = 0 # Higher = prefer if both succeed
async def race_branches(
branches: list[Branch],
cancel_on_first_success: bool = True,
) -> tuple[str, dict]:
"""
Run multiple agent branches concurrently.
With cancel_on_first_success, return the first successful result and
use priority to break ties. Otherwise, return the highest-priority
successful branch after all branches finish.
"""
tasks = {
asyncio.create_task(branch.task()): branch
for branch in branches
}
results = {}
try:
pending = set(tasks.keys())
while pending:
done, pending = await asyncio.wait(
pending, return_when=asyncio.FIRST_COMPLETED
)
for task in done:
branch = tasks[task]
try:
results[branch.name] = (branch.priority, task.result())
if cancel_on_first_success:
for p in pending:
p.cancel()
pending.clear()
except Exception:
pass # Branch failed; continue waiting for others
finally:
for task in tasks:
if not task.done():
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
if not results:
raise RuntimeError("All branches failed")
best_name = max(results, key=lambda k: results[k][0])
return best_name, results[best_name][1]
The economics of speculative branching depend on the cost of the wasted work. If each branch costs $0.001 and the winner saves the user five seconds, the tradeoff is obvious. If each branch costs $0.50 and they are equally likely to succeed, you have just doubled your cost with no expected improvement. Branches should be cheap to start, fast to cancel, and used only when the expected latency gain outweighs the wasted compute.
Prefetching Context #
Speculation applies to context retrieval as well. Memory retrieval, RAG lookups, and database queries all have latency. If you can predict what the agent will need before the model explicitly requests it, you can start retrieving it while the user's message is still being parsed.
class ContextPrefetcher:
"""
Starts likely context retrievals as soon as a user message arrives,
before the model begins generating.
"""
def __init__(self, retriever, intent_classifier):
self.retriever = retriever
self.intent_classifier = intent_classifier
self._prefetch_tasks: dict[str, asyncio.Task] = {}
async def prefetch(self, user_message: str) -> None:
"""
Classify intent and speculatively fetch context
for the most likely retrieval queries.
"""
predicted_queries = await self.intent_classifier.predict_queries(user_message)
for query in predicted_queries:
if query not in self._prefetch_tasks:
self._prefetch_tasks[query] = asyncio.create_task(
self.retriever.retrieve(query)
)
async def get(self, query: str) -> list[dict]:
"""
Return prefetched results if available, otherwise fetch now.
"""
if query in self._prefetch_tasks:
return await self._prefetch_tasks.pop(query)
return await self.retriever.retrieve(query)
This is most effective when the system prompt, user history, or routing metadata gives you a strong signal about what the agent will do — a support agent that receives a message mentioning "billing" should start fetching the user's account and recent invoice history before the model reads the message.
Cancellation and Interruption #
Speculative work that turns out to be wrong is wasted compute. Parallel branches that lose the race need to be shut down cleanly. Users who change their mind mid-generation need to be able to abort without orphaned background tasks. Cancellation is not an afterthought — it is a first-class design requirement for any system that does predictive or parallel work.
Cancellation Propagation #
In a deep agent call stack, cancelling the top-level request needs to propagate through every in-flight sub-task, tool call, and sub-agent invocation. The standard pattern is a cancellation token — a shared signal that every layer checks before doing expensive work.
import asyncio
from dataclasses import dataclass, field
@dataclass
class CancellationToken:
_cancelled: bool = field(default=False, init=False)
_event: asyncio.Event = field(default_factory=asyncio.Event, init=False)
def cancel(self) -> None:
self._cancelled = True
self._event.set()
@property
def is_cancelled(self) -> bool:
return self._cancelled
def raise_if_cancelled(self) -> None:
if self._cancelled:
raise asyncio.CancelledError("Request was cancelled")
async def wait(self) -> None:
await self._event.wait()
async def cancellable_tool_call(
tool_name: str,
tool_input: dict,
tool_executor,
token: CancellationToken,
) -> dict:
"""
Run a tool call that respects a cancellation token.
Uses asyncio.wait to race the tool against the cancel signal.
"""
token.raise_if_cancelled()
tool_task = asyncio.create_task(tool_executor.run(tool_name, tool_input))
cancel_task = asyncio.create_task(token.wait())
try:
done, _ = await asyncio.wait(
[tool_task, cancel_task],
return_when=asyncio.FIRST_COMPLETED,
)
if cancel_task in done:
raise asyncio.CancelledError(
f"Tool call to {tool_name} was cancelled"
)
return await tool_task
finally:
for task in (tool_task, cancel_task):
if not task.done():
task.cancel()
await asyncio.gather(tool_task, cancel_task, return_exceptions=True)
Pass the same CancellationToken instance down through every layer of the agent call stack. When the user hits "stop" or the orchestrator decides to abandon a branch, a single token.cancel() propagates the signal everywhere.
Partial Result Delivery #
When a long generation is cancelled mid-stream, you have a choice: discard everything or deliver whatever was produced. For research agents or document synthesis tasks, a partial answer is often better than no answer. The streaming event model described earlier makes this easy — the frontend simply stops rendering new events and surfaces what it has.
async def interruptible_stream(
messages: list[dict],
model_client,
tool_executor,
token: CancellationToken,
) -> AsyncIterator[StreamEvent]:
"""
Stream agent output with cooperative cancellation.
Yields a DONE event with partial=True if interrupted.
"""
try:
async for event in agent_event_stream(
messages, model_client, tool_executor
):
token.raise_if_cancelled()
yield event
except asyncio.CancelledError:
yield StreamEvent(
type=EventType.DONE,
payload={"partial": True, "reason": "cancelled"},
)
Latency Budgets and Adaptive Degradation #
Not all latency is avoidable. Sometimes the work genuinely takes time. The engineering challenge is to allocate that time deliberately rather than letting it accumulate accidentally.
A latency budget is an explicit time limit per logical stage of an agent turn. When a stage overruns its budget, you have predefined fallback behavior rather than an unbounded wait.
import asyncio
from typing import Awaitable, TypeVar
T = TypeVar("T")
async def with_latency_budget(
coro: Awaitable[T],
budget_ms: float,
fallback: T,
) -> T:
"""
Run a coroutine within a latency budget.
Returns the fallback value if the budget is exceeded.
"""
try:
return await asyncio.wait_for(coro, timeout=budget_ms / 1000)
except asyncio.TimeoutError:
return fallback
async def retrieve_with_latency_budget(retriever, user_query: str) -> list[dict]:
"""Gracefully degrade RAG retrieval if it is too slow."""
return await with_latency_budget(
coro=retriever.retrieve(user_query),
budget_ms=300,
fallback=[], # Proceed with no retrieved context rather than wait
)
Applied across the agent lifecycle, latency budgets give you adaptive degradation: context retrieval that is too slow is skipped, a slow speculative tool call is cancelled and re-run formally, a sub-agent that takes too long gets a timeout and the orchestrator falls back to a direct model call.
Latency-budgeted agent turn:
┌──────────────────────────────────────────────────────┐
│ Stage │ Budget │ Fallback if exceeded │
├──────────────────────────────────────────────────────┤
│ Memory retrieval │ 200ms │ Skip long-term memory │
│ RAG retrieval │ 400ms │ Proceed without docs │
│ Tool call #1 │ 3,000ms │ Return cached result │
│ Tool call #2 │ 3,000ms │ Omit from synthesis │
│ Final synthesis │ 5,000ms │ Return partial stream │
└──────────────────────────────────────────────────────┘
The fallback values are a product decision. A customer-facing agent might skip context retrieval rather than show a loading spinner. A research agent might return a "still working" message and continue in the background. Design the fallbacks before you need them.
Optimization Measurement #
Latency engineering without measurement is guesswork. The metrics to track are:
Time to First Token (TTFT). The latency from request receipt to the first visible character. This is the metric most correlated with perceived responsiveness. Even if total generation takes 8 seconds, a 400ms TTFT feels responsive.
Time to First Tool Result. Relevant when the agent's first action is a tool call. This captures how long the user waits before the agent has any information to work with.
End-to-End Latency (E2E). Total wall-clock time from request to complete response. Important for automated pipelines and background tasks.
Speculative Hit Rate. The fraction of speculative tool calls or context prefetches that were actually used. A hit rate below ~60% suggests you are wasting more compute than you are saving.
Cancellation Rate. If users are cancelling frequently, either the agent is too slow or the early output is not useful enough to keep them engaged.
from dataclasses import dataclass, field
import time
@dataclass
class LatencyTrace:
request_id: str
start_time: float = field(default_factory=time.monotonic)
first_token_time: float | None = None
first_tool_result_time: float | None = None
end_time: float | None = None
speculative_calls: int = 0
speculative_hits: int = 0
cancelled: bool = False
def record_first_token(self) -> None:
if self.first_token_time is None:
self.first_token_time = time.monotonic()
def record_first_tool_result(self) -> None:
if self.first_tool_result_time is None:
self.first_tool_result_time = time.monotonic()
def complete(self) -> None:
self.end_time = time.monotonic()
@property
def ttft_ms(self) -> float | None:
if self.first_token_time is None:
return None
return (self.first_token_time - self.start_time) * 1000
@property
def e2e_ms(self) -> float | None:
if self.end_time is None:
return None
return (self.end_time - self.start_time) * 1000
@property
def speculative_hit_rate(self) -> float | None:
if self.speculative_calls == 0:
return None
return self.speculative_hits / self.speculative_calls
Instrument every agent turn with a LatencyTrace instance and emit it as a structured log event at completion. Over time, you will see which stages are reliable, which are inconsistent, and where speculative work is paying off.
Conclusion #
Latency in agents is a layered problem. At the surface, streaming output token-by-token eliminates the worst user experience — staring at a blank screen while the model thinks. Parallel tool execution removes the serial penalty of multi-step workflows. Speculative prefetching starts work before it is formally requested, turning prediction latency into hidden background work. Latency budgets and adaptive fallbacks keep the user experience bounded even when individual stages are slow.
The key trade-offs to carry forward:
- Streaming is almost always worth it. The implementation cost is low and the perceived improvement is large.
- Speculation pays off when call patterns are predictable and individual calls are cheap. It adds waste when predictions are wrong and adds complexity regardless.
- Speculative branching is best reserved for high-value, high-uncertainty decision points.
- Cancellation is not optional in any system that does parallel or predictive work. Design it in from the start.
- Latency budgets enforce discipline. Unbounded waits compound; bounded stages degrade gracefully.
An agent that feels fast is not necessarily faster — it is one where the user always has something happening, something to read, and some control over how long they wait.