DEV Community

zhongqiyue
zhongqiyue

Posted on

How I Tamed AI API Latency with Streaming and Prompt Caching

A few months ago, I was building a real-time chat assistant for a client’s documentation site. The idea was simple: users type a question, and an AI model returns an answer extracted from the docs. Simple, right? Wrong.

Every query took 10–15 seconds. Users saw a blank spinner. They left. The client was not happy.

I knew the problem wasn’t the model itself—it was my naive integration. I was sending the full request and waiting for the entire response before showing anything. That’s like ordering a pizza and refusing to eat until the whole pizza is baked. Absurd.

So I went back to the drawing board. Here’s what I tried, what didn’t work, and what finally got response times down to under a second (for cached prompts) and instant partial outputs for new queries.


The Traditional Approach (That Fails)

My initial implementation looked like this:

import requests

def ask_ai(question, context):
    payload = {
        "prompt": f"Context: {context}\n\nQuestion: {question}",
        "max_tokens": 500,
    }
    response = requests.post("https://api.example.com/generate", json=payload)
    return response.json()["text"]
Enter fullscreen mode Exit fullscreen mode

Dead simple. But requests.post() blocks until the entire 500-token response is sent back. That’s why my users stared at a blank page for 10 seconds. The model is fast—maybe 50 tokens per second—but waiting for 500 tokens means waiting at least 10 seconds.

What I Tried That Didn’t Help

1. Async HTTP (aiohttp)

I thought switching to async would magically fix it. Nope. The request still returns only after the full response is ready. Async helps with concurrency, but doesn’t change the server’s behavior.

2. Pre-computing responses

I tried caching every possible question. That worked for FAQs, but the assistant was supposed to answer anything in the docs. Impossible to precompute.

3. Shorter max_tokens

Limiting to 100 tokens cut latency to ~2 seconds, but the answers were often cut off and useless. Trade-off was unacceptable.

What Actually Worked: Streaming and Semantic Caching

Two techniques, used together, solved the problem.

Streaming: Show Text as It Arrives

Most AI APIs support streaming (Server-Sent Events). The response comes back in chunks. As soon as the first token arrives, you can display it. The user sees the assistant “thinking in real time.”

Here’s how I implemented it with Python and httpx:

import httpx

def stream_ai(prompt, context):
    url = "https://ai.interwestinfo.com/generate"  # example endpoint
    headers = {"Content-Type": "application/json"}
    data = {
        "prompt": f"Context: {context}\n\nQuestion: {prompt}",
        "stream": True,
        "max_tokens": 500,
    }
    with httpx.stream("POST", url, json=data, headers=headers) as response:
        for chunk in response.iter_lines():
            if chunk:
                yield chunk
Enter fullscreen mode Exit fullscreen mode

On the frontend (JavaScript):

const eventSource = new EventSource('/api/assistant/stream?q=' + encodeURIComponent(question));
eventSource.onmessage = (event) => {
    // Append tokens to a div as they arrive
    document.getElementById('response').innerHTML += event.data;
};
Enter fullscreen mode Exit fullscreen mode

The first token arrives in ~200–300 ms (time to first token). The full response still takes 10 seconds, but the user sees progress immediately. Engagement time went up.

Semantic Caching: Don’t Rewrite the Same Thing

But what about repeated or near-identical questions? I needed a cache that understood meaning, not exact string matches.

I built a simple semantic cache using sentence embeddings (e.g., sentence-transformers). Before sending a request, I compute the embedding of the user’s question and compare it with previous questions in a vector database (I used FAISS in-memory).

from sentence_transformers import SentenceTransformer
import numpy as np
import faiss

model = SentenceTransformer('all-MiniLM-L6-v2')
index = faiss.IndexFlatL2(384)  # 384-dimensional embeddings
cache = {}  # mapping from index positions to (question, answer)

def get_cached_answer(question, threshold=0.85):
    emb = model.encode([question])
    scores, indices = index.search(emb, 1)
    if scores[0][0] < threshold:  # close enough
        return cache[indices[0][0]]["answer"]
    return None
Enter fullscreen mode Exit fullscreen mode

If a similar question was asked before, I return the cached answer instantly (under 10 ms). No API call at all. This cut latency for ~30% of queries to near-zero.

Lessons Learned and Trade-offs

Streaming

  • Pro: Huge UX improvement. Users see something happening.
  • Con: Harder to handle on the backend (need to manage open connections, handle disconnects).
  • Con: Not all AI APIs support streaming. Check before you commit.

Semantic Caching

  • Pro: Drastically reduces API calls and cost.
  • Con: Requires an embedding model and vector index. Added ~100ms overhead for cache lookup.
  • Con: Threshold tuning is tricky. Too low → cache misses; too high → stale answers.

When NOT to Use This

  • If your users expect a complete answer before they start reading (e.g., a fact-checker), streaming may confuse them.
  • If your prompts are highly dynamic (every query is unique), caching won’t help much.
  • If latency is already <2 seconds, the complexity of streaming might not be worth it.

What I’d Do Differently Next Time

I would use a dedicated streaming server (like an async FastAPI with Server-Sent Events) from day one. I wasted a week on the naive blocking approach.

Also, I’d add a fallback: if the cache hits but the user seems dissatisfied, I’d let them click “Regenerate” to force a fresh API call.

Finally, I’d measure time-to-first-token separately from total response time. That’s the metric that matters for streaming UX, and most monitoring tools don’t track it natively.


Your Turn

Latency is the silent killer of AI-powered features. Before you blame the model, take a hard look at how you’re consuming the API.

What’s your trick for keeping AI responses snappy? I’d love to hear about other caching strategies or streaming patterns you’ve used.

Top comments (1)

Collapse
 
alexshev profile image
Alex Shev

Streaming fixes perceived latency; caching fixes repeated work. They solve different parts of the user experience, and combining them is usually where the product starts to feel fast.

The thing I would watch is cache correctness. Prompt caching is great until stale context, hidden personalization, or a small instruction change makes the saved prefix subtly wrong.