How to Build Sub-Second Real-Time AI Interaction Pipelines

The standard LLM agent loop is fundamentally broken for real-time human interaction. While an adult waiting for an enterprise search tool might tolerate a three-second pause, many users will not. In early childhood education, latency is a project killer. During early playtests of AI tutoring systems, researchers observed that a mere two-second delay was enough for a six-year-old child to ask why the system was not doing anything, declare it boring, and completely tune out.

This latency bottleneck presents a harsh engineering trade-off. You can use a smaller, faster model to keep things responsive, but these models struggle to follow complex instructions across a broad action space. For an educational app, a small model might constantly give the answer away instead of offering a helpful hint. If you use a frontier model like GPT-4, you get excellent instruction following, but you suffer a massive latency hit. Frontier models can take two to three seconds to produce their first token, decoding at roughly 30 tokens per second. Add network round-trips and audio synthesis, and the user is left waiting four seconds between turns.

To build a truly conversational agent, developers must throw out the standard synchronous tool-calling loop. The architectural patterns emerging from the front lines of AI education, pioneered by teams at Ello and Khan Academy, provide a clean blueprint for building sub-second, highly interactive voice applications.

The Death of the Tool Loop

In a traditional agent architecture, the system relies on a synchronous tool loop. The LLM outputs a tool call, the backend executes the tool, the backend appends the tool's output to the context, and the LLM decides what to do next. This sequential design is a latency disaster.

To break the sub-second barrier, we must separate generation from execution. Instead of waiting for a tool call to complete, the system can stream multiple actions in a single response. An interpreter parses and executes each action on the fly while the model is still generating the next ones.

Under this model, the user only has to wait for the first action to stream, which typically takes about 30 tokens, rather than waiting for the entire generation to finish.

[Model Stream] ---> (Action 1: Speak) ---> [Immediate Audio Playback]
               ---> (Action 2: Draw)  ---> [UI Updates in Background]
               ---> (Action 3: Listen) ---> [Open Microphone]

This approach also allows for dynamic action availability. When a specific question is active on the screen, the system can restrict the available actions to scaffolding and hinting, rather than allowing the model to output the final answer.

Implementing a Streaming Action Interpreter

To make this work, you cannot use standard JSON parsing libraries, which expect a complete payload before they can output a structured object. Instead, you need a streaming parser that can identify and yield completed actions from a chunked stream.

Here is a lightweight Python implementation of a streaming parser designed to extract and execute actions as soon as they are fully formed in the stream, using a custom delimiter format:

import json
from typing import Generator, Dict, Any

class StreamingActionParser:
    def __init__(self):
        self.buffer = ""
        
    def feed(self, chunk: str) -> Generator[Dict[str, Any], None, None]:
        self.buffer += chunk
        
        # We look for completed action blocks wrapped in custom tags
        while "<action>" in self.buffer and "</action>" in self.buffer:
            start_idx = self.buffer.find("<action>")
            end_idx = self.buffer.find("</action>")
            
            if start_idx < end_idx:
                action_content = self.buffer[start_idx + 8:end_idx].strip()
                try:
                    action_data = json.loads(action_content)
                    yield action_data
                except json.JSONDecodeError:
                    # Handle partial or malformed JSON if necessary
                    pass
                
                # Slide the buffer past the processed action
                self.buffer = self.buffer[end_idx + 9:]
            else:
                # Handle edge case where closing tag appears before opening tag
                self.buffer = self.buffer[start_idx:]
                break

This parser allows the client to execute the first action, such as playing an audio file, while subsequent actions are still being generated and parsed.

Validation also happens on the fly. If the stream produces an invalid action, the system can immediately interrupt the generation and trigger a fallback. On the happy path, execution never pauses, resulting in a system that feels incredibly fluid.

The Dual-Agent Pattern: Converser and Async Planner

To keep the user-facing agent fast, its context window and action space must remain small. A smaller action space directly correlates with better instruction following in smaller, faster models. However, a complex task like tutoring or guided troubleshooting requires long-term planning and deep reasoning.

To solve this, we can split the workload into two distinct agents:

1. The Converser: A highly responsive, low-latency agent that handles the immediate back-and-forth conversation. It operates with a minimal action space and a tight context window.
2. The Planner: An asynchronous agent that runs in the background while the user is thinking, speaking, or reading. It reviews the conversation history against the broader objectives and updates the Converser's context state.

Because these two agents run concurrently, reading and writing to a shared state without direct coordination, state management becomes a critical challenge. The cleanest solution is to store every turn, tap, and UI update as an immutable event stream.

# A simplified representation of the immutable state log
state_log = [
    {"event": "ui_render", "element": "math_problem_1", "timestamp": 1710000000},
    {"event": "user_speech", "text": "I think the answer is five.", "timestamp": 1710000005},
    {"event": "planner_update", "scaffolding_level": "high", "timestamp": 1710000006},
    {"event": "converser_response", "text": "Not quite, let's look at the blocks again.", "timestamp": 1710000007}
]

By treating state as an append-only log of immutable events, the asynchronous Planner can compute new context instructions without risking race conditions or blocking the Converser's immediate response loop.

The Developer Trade-offs

Building a custom real-time pipeline is not free. Popular agent frameworks like LangChain are largely optimized for background tasks where latency is a secondary concern. When you choose to own the loop, you must build your own observability, tracing, and debugging tools from scratch.

Furthermore, you will find yourself fighting against the current of modern model optimization. Most frontier models are heavily post-trained on standard tool-calling patterns. Forcing them to stream custom action syntaxes instead of waiting for a complete tool-use block requires meticulous prompt engineering and rigorous system instructions.

Despite these hurdles, the performance gains are undeniable. By moving from a synchronous tool loop to a streaming action interpreter, and by offloading heavy reasoning to an asynchronous planner, you can build conversational AI systems that respond in under a second. Whether you are building an AI tutor for a five-year-old or a voice assistant for a field engineer, these architectural patterns are the key to making real-time AI feel less like a slow chatbot and more like a natural conversation.