Real-Time Agents: Handling Streaming Input and Low-Latency Response

Why "Fast Enough" Is Never Fast Enough

Picture a voice agent that takes four seconds to respond after a user finishes speaking. On paper, four seconds sounds trivial. In a live conversation, it feels like an eternity — long enough for the user to repeat themselves, hang up, or assume the system crashed. Real-time agents live or die by a different clock than the one we use for batch jobs or background workers. When a human is waiting on the other end — typing, speaking, or watching a cursor blink — every additional hundred milliseconds compounds into frustration.

This is the core tension in real-time agent design: language models are inherently sequential, streaming input arrives in unpredictable chunks, and network hops add latency you can't fully eliminate. Yet users expect agents to feel instantaneous — closer to a phone call than a support ticket. Building for this expectation requires rethinking the agent pipeline from the ground up, not just bolting streaming onto a request-response architecture.

This article walks through the practical architecture of real-time agents: how streaming input actually works, where latency hides, how to design for partial and incremental processing, and the patterns that separate agents that feel alive from agents that feel like they're thinking very hard. We'll write actual code, not just describe concepts, because the difference between a streaming demo and a production-grade real-time agent is almost entirely in the implementation details.

What Makes an Agent "Real-Time"

A real-time agent processes continuous or incremental input and produces output with latency low enough that the interaction feels conversational rather than transactional. That's a deliberately broad definition, because "real-time" covers several distinct scenarios:

• Voice agents processing live audio streams, where input arrives as a continuous signal that must be transcribed, interpreted, and responded to before the user gets impatient.
• Chat agents with streaming tokens, where the model's response is displayed incrementally as it's generated, rather than waiting for the full completion.
• Live data agents monitoring a feed (stock prices, sensor telemetry, log streams) and reacting to events as they occur.
• Collaborative agents watching a shared document or codebase change in real time and offering suggestions inline.

Each of these shares a common architectural demand: the agent cannot wait for a clean, complete, well-formed input before it starts working. It has to process partial information, make provisional decisions, and often revise those decisions as more data arrives. This is fundamentally different from the "receive full prompt, generate full response" pattern that dominates most agent tutorials.

The practical implication is that your agent needs at least three capabilities most simple agent scripts don't have: an incremental input buffer, a way to decide when a "turn" is actually complete, and a streaming output path that can start producing tokens before the full response is planned.

Where Latency Actually Comes From

Before optimizing anything, it helps to break down where time goes in a typical real-time agent request. Most teams assume the model call is the bottleneck. It's often not — or not the only one.

• Network round trips: every hop between client, server, and model provider adds tens to hundreds of milliseconds. Stacking multiple sequential API calls (retrieve context, call model, call a tool, call the model again) multiplies this cost.
• Time to first token (TTFT): the delay between sending a prompt and receiving the first streamed token. This is a function of model size, prompt length, and provider load, and it's usually the single biggest lever you have.
• Tool execution latency: if your agent calls a database, a search API, or another service mid-conversation, that call blocks the response unless you design around it.
• Turn-detection latency: for voice agents specifically, deciding "has the user finished speaking?" adds a deliberate buffer — too short and you cut people off, too long and the agent feels sluggish.
• Client-side rendering: how fast your UI can paint incoming tokens matters more than people expect. A backend that streams perfectly but a frontend that batches renders every 500ms will still feel slow.

The mistake most teams make is optimizing the model call and ignoring everything else. A well-architected real-time agent treats the entire pipeline — input capture, buffering, model inference, tool calls, and output rendering — as a single latency budget to be spent deliberately.

Designing the Streaming Input Pipeline

Streaming input needs a buffer and a policy for when to act on it. You rarely want to react to every single token or audio frame as it arrives — that wastes compute and produces jittery, premature responses. Instead, you need a windowing strategy.

Here's a minimal but realistic pattern for buffering streaming text input (for example, partial transcription results from a speech-to-text service) and deciding when to trigger agent processing:

import asyncio
import time
from dataclasses import dataclass, field


@dataclass
class StreamBuffer:
    text: str = ""
    last_update: float = field(default_factory=time.monotonic)
    silence_threshold: float = 0.6  # seconds of no new input
    min_chars_to_consider: int = 3

    def append(self, chunk: str) -> None:
        self.text += chunk
        self.last_update = time.monotonic()

    def is_ready(self) -> bool:
        idle_time = time.monotonic() - self.last_update
        has_content = len(self.text.strip()) >= self.min_chars_to_consider
        return has_content and idle_time >= self.silence_threshold


async def stream_consumer(input_queue: asyncio.Queue, on_turn_complete):
    buffer = StreamBuffer()
    while True:
        try:
            chunk = await asyncio.wait_for(input_queue.get(), timeout=0.1)
            buffer.append(chunk)
        except asyncio.TimeoutError:
            pass  # no new chunk this tick, fall through to check readiness

        if buffer.is_ready():
            finished_text = buffer.text
            buffer.text = ""
            await on_turn_complete(finished_text)

This pattern — poll on a short interval, track idle time, fire when silence exceeds a threshold — is the backbone of turn detection in most voice and streaming-text agents. The silence_threshold is your single most important tuning knob. Too low, and the agent interrupts users mid-sentence. Too high, and every pause feels like the agent stalled. Production systems often make this adaptive: shorter thresholds for short utterances, longer ones when the user is clearly still composing a longer thought (e.g., detecting trailing conjunctions or unfinished punctuation).

For voice specifically, you'll layer a proper voice activity detection (VAD) model underneath this instead of relying purely on text-arrival gaps, since audio silence and transcription lag don't move in lockstep.

Streaming the Model's Output, Not Just the Input

Getting input right is half the problem. The other half is making sure the agent's response starts appearing before it's fully generated. Nearly every major model API supports token streaming, and if your agent isn't using it, you're leaving the single biggest perceived-latency win on the table.

Here's a pattern for streaming a model response while also watching for tool-call intents mid-stream, which is the trickiest part of real-time agent output:

async def stream_agent_response(client, messages, tools, on_token, on_tool_call):
    accumulated_text = ""
    pending_tool_call = None

    async with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=messages,
        tools=tools,
    ) as stream:
        async for event in stream:
            if event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    accumulated_text += event.delta.text
                    await on_token(event.delta.text)  # push to UI immediately

            elif event.type == "content_block_start":
                if event.content_block.type == "tool_use":
                    pending_tool_call = {
                        "name": event.content_block.name,
                        "input": ""
                    }

            elif event.type == "content_block_stop":
                if pending_tool_call is not None:
                    await on_tool_call(pending_tool_call)
                    pending_tool_call = None

    return accumulated_text

The key idea is on_token fires as soon as text arrives, not after the full message resolves. If your architecture instead buffers the entire response server-side and forwards it as one payload, you've thrown away the latency advantage streaming was supposed to give you. This is a surprisingly common mistake — teams add a middleware layer, a queue, or a logging step that silently collects the full stream before forwarding it downstream, turning a streaming API into a blocking one without anyone noticing until a user complains.

Note also that tool calls interrupt the "just stream text" story. When the model decides to call a tool mid-response, you often want to show the user some indication of that ("looking that up...") rather than silence, and you need a strategy for resuming the text stream after the tool result comes back.

Handling Tool Calls Without Killing the Vibe

Tool calls are where real-time agents most often stall. A synchronous database query, a slow third-party API, or a poorly indexed search can turn a snappy conversation into a multi-second dead zone. There are three practical mitigations worth building into any real-time agent.

• Parallelize independent tool calls. If the agent needs both a weather lookup and a calendar check, fire them concurrently, not sequentially.
• Speculative or filler responses. Many production voice agents insert a short acknowledgment ("let me check that") while a tool call resolves, rather than leaving dead air. This is a UX trick, not a latency fix, but it changes perceived latency enormously.
• Timeout and degrade. Every external tool call needs an explicit timeout with a fallback path. An agent that hangs indefinitely on a flaky API is worse than one that gracefully says "I couldn't reach that service, here's what I know."

import asyncio

async def call_tool_with_timeout(tool_fn, args, timeout=2.5, fallback=None):
    try:
        return await asyncio.wait_for(tool_fn(**args), timeout=timeout)
    except asyncio.TimeoutError:
        return fallback or {"error": "timed_out", "detail": "tool did not respond in time"}


async def run_parallel_tools(tool_calls: list):
    tasks = [
        call_tool_with_timeout(call["fn"], call["args"], timeout=call.get("timeout", 2.5))
        for call in tool_calls
    ]
    return await asyncio.gather(*tasks)

The fallback parameter matters more than it looks. A hard failure with no fallback path forces your agent's response logic to handle an exception mid-conversation, which usually means either a crash or an awkward silence. A structured fallback lets the agent narrate the failure naturally ("I'm having trouble reaching the calendar right now") instead of breaking character.

Managing State Across a Live Conversation

Real-time agents typically run longer, messier conversations than single-shot completions, and they need to track state that spans multiple turns: what's been said, what tools have been called, what the user corrected or interrupted. A naive approach re-sends the entire conversation history on every turn, which works until your context grows large enough that it starts adding meaningful latency to every single model call — longer prompts mean longer time-to-first-token.

A few practical patterns help here:

1. Sliding window with summarization. Keep the last N turns verbatim and periodically summarize older turns into a compact block, so the prompt size stays roughly constant instead of growing linearly with conversation length.
2. Interrupt handling as a first-class event. If the user starts speaking while the agent is still responding (a "barge-in"), your state machine needs an explicit interrupted state — not just a cancelled request. The agent should remember it was cut off and may need to acknowledge that on the next turn.
3. Separate "fast path" and "slow path" state. Session-level state that changes every turn (current topic, last tool result) should live in memory or a low-latency store like Redis. Long-term state (user preferences, history across sessions) can live in a slower persistent store, fetched asynchronously and merged in when it's ready rather than blocking the turn on it.

class ConversationState:
    def __init__(self, max_turns=12):
        self.turns = []
        self.max_turns = max_turns
        self.summary = ""
        self.interrupted = False

    def add_turn(self, role: str, content: str):
        self.turns.append({"role": role, "content": content})
        if len(self.turns) > self.max_turns:
            self._compact()

    def _compact(self):
        oldest = self.turns[: len(self.turns) - self.max_turns]
        self.summary += " " + " ".join(t["content"][:120] for t in oldest)
        self.turns = self.turns[len(self.turns) - self.max_turns :]

    def mark_interrupted(self):
        self.interrupted = True

    def to_messages(self):
        messages = []
        if self.summary:
            messages.append({"role": "system", "content": f"Earlier context: {self.summary.strip()}"})
        messages.extend(self.turns)
        return messages

This keeps the prompt bounded in size regardless of how long the conversation runs, which directly protects your time-to-first-token as sessions get longer — a detail that's easy to miss in a demo that only ever runs for three or four turns.

Testing and Observability for Latency-Sensitive Agents

You cannot improve what you don't measure, and latency bugs in real-time agents are notoriously invisible in normal application logs. A request that "succeeded" but took six seconds looks identical to a fast one in a status-code-only log. You need latency-specific instrumentation from day one.

• Instrument every hop separately: time-to-first-token, tool call duration, total turn duration, and client-side render latency should each be logged as distinct metrics, not folded into one "response time" number.
• Track p50 and p95, not just averages. A real-time agent with a great average latency but a bad p95 will still generate a steady stream of frustrated users — averages hide exactly the tail behavior that matters most for perceived quality.
• Simulate network jitter and slow tools in testing. It's easy to build and test an agent entirely on a fast local network against fast APIs, then discover in production that a 300ms added delay from a mobile network completely changes user behavior (more interruptions, more repeated questions).
• Log turn boundaries explicitly. When debugging a "the agent felt slow" complaint, you want to reconstruct exactly when the user stopped talking, when the agent started processing, when the first token appeared, and when the tool calls resolved — as a timeline, not a single duration number.

A simple but effective approach is wrapping every stage of the pipeline in a timing decorator that emits structured events to your logging or tracing system, so a single conversation turn produces a full waterfall you can inspect later, the same way you'd inspect a web page's network waterfall.

import time
import functools
import logging

logger = logging.getLogger("realtime_agent")

def timed_stage(stage_name):
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            start = time.monotonic()
            result = await fn(*args, **kwargs)
            elapsed_ms = (time.monotonic() - start) * 1000
            logger.info("stage=%s elapsed_ms=%.1f", stage_name, elapsed_ms)
            return result
        return wrapper
    return decorator


@timed_stage("tool_call")
async def fetch_order_status(order_id: str):
    ...

Small as this looks, structured per-stage timing is usually the single fastest way to find where a real-time agent's latency budget is actually being spent, instead of guessing.

Common Failure Patterns to Avoid

A few mistakes show up repeatedly in real-time agent implementations, often only surfacing once real users start hitting the system under real network conditions.

• Buffering an entire streamed response before forwarding it. This defeats the purpose of streaming and is surprisingly easy to introduce accidentally through a proxy, logging middleware, or a well-meaning "let me validate the full response first" step.
• No interrupt handling. If a user speaks over the agent and the system just keeps talking or silently drops the new input, the conversation feels broken rather than fast.
• Unbounded context growth. Sending the full conversation history on every turn without compaction eventually turns a snappy agent into a sluggish one as the session lengthens.
• Treating every tool call as blocking. Sequential tool calls that could run in parallel are one of the most common and easiest-to-fix sources of unnecessary latency.
• No fallback for slow or failing tools. A single flaky dependency without a timeout can stall an otherwise well-optimized agent indefinitely.
• Ignoring perceived latency. Sometimes the fix isn't making things faster — it's giving the user feedback (a typing indicator, a short acknowledgment, a partial answer) so the wait doesn't feel empty.

None of these are exotic problems. They're the same categories of mistakes that show up in any distributed system with latency requirements — the difference is that in an agent, they're directly visible to the end user in a way that's hard to hide.

Bringing It Together

Real-time agents demand a different mindset than typical request-response AI applications. You're not just calling a model and returning a result — you're managing a continuous pipeline of partial input, provisional decisions, concurrent tool calls, and incremental output, all under a latency budget that a human is actively feeling in real time. The patterns in this article — buffered turn detection, token-level streaming, parallelized and timeout-protected tool calls, bounded conversation state, and per-stage latency instrumentation — form the foundation most production voice and chat agents are actually built on, even if the demo-level tutorials rarely show them.

The good news is that none of this requires exotic infrastructure. It requires being deliberate about where time goes, treating streaming as a first-class design constraint rather than an afterthought, and instrumenting your pipeline well enough that "the agent feels slow" turns into a specific, fixable measurement instead of a vague complaint.

If you want to go deeper — building an actual production-grade agent that handles streaming input, tool orchestration, and low-latency response end to end — that's exactly what we cover hands-on in 30 Days of Hermes Agent, our project-based course at teachyou.ai. You'll build a real streaming agent from scratch, wire up turn detection and interrupts, and instrument it for latency the way production teams actually do it, rather than just reading about the theory.