Latency optimization for interactive applications

A backend batch job can tolerate a model call taking several seconds. A chat interface or an autocomplete feature cannot — a user staring at a blank screen for three seconds perceives the application as broken, regardless of how good the eventual answer is. This lesson covers latency-reduction techniques specific to interactive, user-facing applications, building on the phase-timing measurement introduced in Lesson 1.

Why Interactive Latency Is a Different Problem

For a background job, the metric that matters is total completion time, and it is usually fine to wait for the entire response before doing anything with it. For an interactive application, the metric that matters most is perceived latency — specifically, time to first visible feedback — which is not the same as total completion time. A response that takes four seconds to fully generate but starts displaying text after 300 milliseconds feels dramatically faster to a user than a response that takes three seconds total but appears all at once at the end.

This distinction is why the techniques in this lesson are about managing latency for a good user experience, not just minimizing total latency, though the two often overlap. Streaming, the first technique below, is a direct example: it does not necessarily make the total request faster, but it makes the perceived latency far shorter.

Streaming Responses

Rather than waiting for the entire response to be generated before returning anything to the caller, streaming delivers output tokens to the client as soon as the model produces them.

from openai import OpenAI

client = OpenAI()


def stream_response(messages: list[dict], model: str = "gpt-5.6-terra"):
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

Note: The exact shape of a streamed chunk (chunk.choices[0].delta.content) reflects the OpenAI SDK's streaming response format at the time of writing. Confirm this against your installed SDK version, since streaming response shapes have been revised across versions.

stream_response is a Python generator (using yield rather than return), which means the caller can start displaying text as each delta arrives rather than waiting for the entire function to finish. The if delta: check matters because not every chunk in a streaming response necessarily carries new text — some chunks may carry only metadata — and yielding an empty or None delta would either display nothing useful or cause an error in code that expects a string.

For a chat UI, this is what enables the familiar "typing" effect where text appears progressively rather than all at once. For non-chat interactive features (a search box with live suggestions, a live-editing assistant), streaming still helps whenever the interface can meaningfully use partial output as it arrives, even if it doesn't display token-by-token to the user.

Measuring Time to First Token

Streaming's benefit is specifically about time to first token (TTFT), so that is the metric worth measuring, separately from total generation time.

import time


def measure_streaming_latency(messages: list[dict], model: str = "gpt-5.6-terra") -> dict:
    start = time.time()
    first_token_at = None
    full_text = []

    for delta in stream_response(messages, model):
        if first_token_at is None:
            first_token_at = time.time()
        full_text.append(delta)

    end = time.time()

    return {
        "time_to_first_token_ms": (first_token_at - start) * 1000 if first_token_at else None,
        "total_time_ms": (end - start) * 1000,
        "output_length_chars": len("".join(full_text)),
    }

Recording first_token_at only once — the first time through the loop where it is still None — isolates exactly the moment the user would first see something on screen, distinct from end, which marks when generation fully completes. Tracking both numbers together lets you see the actual user-perceived improvement from streaming: time_to_first_token_ms should be dramatically lower than total_time_ms for any response of meaningful length, and that gap is the perceived-latency benefit streaming provides.

Running Independent Calls in Parallel

When a feature needs multiple model calls that do not depend on each other's output — for example, generating a title and a summary for the same document — running them sequentially wastes time waiting on each one in turn. Running them concurrently reduces total latency to roughly the slowest single call rather than the sum of all of them.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()


async def generate_title(document_text: str) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-5.6-terra",
        messages=[{"role": "user", "content": f"Write a short title for:\n{document_text}"}],
    )
    return response.choices[0].message.content


async def generate_summary(document_text: str) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-5.6-terra",
        messages=[{"role": "user", "content": f"Summarize in two sentences:\n{document_text}"}],
    )
    return response.choices[0].message.content


async def generate_title_and_summary(document_text: str) -> tuple[str, str]:
    title, summary = await asyncio.gather(
        generate_title(document_text),
        generate_summary(document_text),
    )
    return title, summary

asyncio.gather starts both generate_title and generate_summary concurrently and waits for both to complete, so the total wall-clock time is approximately max(title_latency, summary_latency) rather than title_latency + summary_latency. This only works correctly because the two calls are genuinely independent — neither needs the other's output as input. If generate_summary needed the generated title as part of its prompt, they would have to run sequentially, and parallelizing them would be incorrect, not just unhelpful.

Speculative and Optimistic UI Updates

For features where a fast, provisional response can be shown immediately while a more thorough one is still being computed, a speculative UI pattern can make the interface feel instantaneous even when the underlying work takes time.

def get_search_suggestions(query: str, local_index: dict[str, list[str]]) -> list[str]:
    """Fast, local, non-model lookup shown immediately."""
    prefix = query.lower()
    return [entry for entry in local_index.get(prefix[:1], []) if entry.lower().startswith(prefix)][:5]


async def get_refined_suggestions(query: str) -> list[str]:
    """Slower, model-backed suggestions that replace the local ones once ready."""
    response = await async_client.chat.completions.create(
        model="gpt-5.6-terra-mini",
        messages=[{"role": "user", "content": f"Suggest 5 search completions for: {query}"}],
    )
    return response.choices[0].message.content.split("\n")

The pattern here is to display get_search_suggestions's result immediately (a local lookup takes microseconds), then replace it with get_refined_suggestions's result once the model call resolves. The user sees something useful instantly, and the interface upgrades to a better answer shortly after, rather than showing nothing at all until the model call finishes. This trades a small amount of initial answer quality for a large improvement in perceived responsiveness — appropriate for features like search-as-you-type where an imperfect instant suggestion beats a perfect suggestion that arrives after a visible delay, and inappropriate for features where a wrong provisional answer could mislead the user before being corrected.

Setting and Respecting Timeouts

An interactive feature must never let a single slow model call block the interface indefinitely. Every call needs an explicit timeout with a defined fallback behavior.

import asyncio


async def call_with_timeout(coro, timeout_seconds: float, fallback: str) -> str:
    try:
        return await asyncio.wait_for(coro, timeout=timeout_seconds)
    except asyncio.TimeoutError:
        return fallback

call_with_timeout wraps any coroutine with a hard time limit, returning a defined fallback value rather than letting the caller wait indefinitely for a model call that may be unusually slow or hung. The specific timeout and fallback should be chosen per feature: an interactive suggestion feature might time out after 800 milliseconds and fall back to no suggestion at all, while a less time-sensitive feature might allow several seconds and fall back to a generic message. The key discipline is that some timeout always exists — an interactive feature with no timeout has an unbounded worst-case latency, which is a reliability problem as much as a user-experience one.

Testing Latency-Sensitive Code

Latency optimization logic — parallelization, timeouts, fallback behavior — can and should be tested without waiting on real network calls, using fast fake coroutines that simulate both success and slowness.

async def fake_fast_call() -> str:
    await asyncio.sleep(0.01)
    return "fast result"


async def fake_slow_call() -> str:
    await asyncio.sleep(2.0)
    return "slow result"


def test_call_with_timeout_returns_result_when_fast_enough():
    async def run():
        result = await call_with_timeout(fake_fast_call(), timeout_seconds=0.5, fallback="fallback")
        assert result == "fast result"
        print("PASS: fast call completes within timeout and returns its result")

    asyncio.run(run())


def test_call_with_timeout_falls_back_when_too_slow():
    async def run():
        result = await call_with_timeout(fake_slow_call(), timeout_seconds=0.1, fallback="fallback")
        assert result == "fallback"
        print("PASS: slow call exceeds timeout and returns the fallback value")

    asyncio.run(run())


test_call_with_timeout_returns_result_when_fast_enough()
test_call_with_timeout_falls_back_when_too_slow()

Both fake coroutines use asyncio.sleep with a small, deliberately chosen duration rather than calling a real model — fake_fast_call completes well within its test's timeout and fake_slow_call deliberately exceeds its test's timeout, so each test exercises exactly one branch of call_with_timeout without needing an actual slow network call or a long-running test suite.

Common Mistakes

Optimizing total latency while ignoring perceived latency. A feature that streams output feels faster than one that does not, even at the same total completion time; measuring only total time misses this entirely.

Running independent model calls sequentially by default. Sequential calls are the easiest to write, but for calls with no dependency between them, this needlessly adds their latencies together instead of overlapping them.

Shipping an interactive feature with no request timeout. Without a timeout, a single unusually slow model response can hang the interface indefinitely, turning a rare slow call into a full outage from the user's perspective.

Best Practices

Stream by default for any user-facing generative feature of meaningful length. The perceived-latency benefit is large and the implementation cost is low once the pattern is in place.

Parallelize model calls whenever they are genuinely independent. Check dependency direction carefully before parallelizing — an incorrect assumption of independence produces wrong output, not just a missed optimization.

Always define a timeout and a fallback for every interactive model call. A defined, tested fallback behavior converts a possible indefinite hang into a bounded, predictable degradation the user can understand.