Streaming API for Real-Time AI

Streaming is no longer optional for AI applications. Users expect to see responses appear word-by-word, not wait seconds for a complete block of text. Server-Sent Events (SSE) is the industry-standard protocol for streaming LLM responses. This guide shows you how to implement streaming with Chinese AI models through TokenEase.

Why Streaming Matters

Metric	Blocking API	Streaming API
Time to First Token	2-5 seconds	200-500ms
Perceived Latency	High (wait for all)	Low (immediate feedback)
User Engagement	Users abandon	Users stay and read
Timeout Risk	High for long outputs	Low (stream bypasses timeout)

How SSE Streaming Works

Server-Sent Events uses a persistent HTTP connection where the server pushes data as it becomes available:

1. Client opens HTTP connection with Accept: text/event-stream
2. Server sends data chunks as data: {...}\n\n
3. Client reads each chunk via EventSource or fetch reader
4. Connection closes when generation completes

Implementing Streaming with TokenEase

TokenEase supports SSE streaming across all 6 model providers with a single parameter change: set stream: true.

Python: Streaming with requests

import requests
import json

API_KEY = "your-tokenease-api-key"
BASE_URL = "https://tokenease.io/v1"

def stream_chat(model: str, messages: list):
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": model,
            "messages": messages,
            "stream": True,  # Enable streaming
            "max_tokens": 500
        },
        stream=True  # requests library streaming
    )
    
    # Parse SSE events
    for line in response.iter_lines():
        if line:
            line = line.decode('utf-8')
            if line.startswith('data: '):
                data = line[6:]  # Remove "data: " prefix
                
                if data == '[DONE]':
                    break
                
                try:
                    chunk = json.loads(data)
                    delta = chunk["choices"][0]["delta"]
                    
                    if "content" in delta:
                        print(delta["content"], end='', flush=True)
                    
                    if "reasoning_content" in delta:
                        print(f"[Thinking: {delta['reasoning_content']}]", end='')
                        
                except (json.JSONDecodeError, KeyError):
                    continue

# Usage
messages = [
    {"role": "user", "content": "Explain quantum computing in simple terms"}
]
stream_chat("deepseek", messages)

JavaScript: Browser Implementation

async function streamChat(model, messages, onToken) {
    const response = await fetch('https://tokenease.io/v1/chat/completions', {
        method: 'POST',
        headers: {
            'Authorization': `Bearer ${API_KEY}`,
            'Content-Type': 'application/json'
        },
        body: JSON.stringify({
            model: model,
            messages: messages,
            stream: true,
            max_tokens: 500
        })
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop(); // Keep incomplete line

        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = line.slice(6);
                if (data === '[DONE]') continue;
                
                try {
                    const chunk = JSON.parse(data);
                    const content = chunk.choices[0]?.delta?.content;
                    if (content) onToken(content);
                } catch (e) {
                    // Skip malformed chunks
                }
            }
        }
    }
}

// Usage with React-style state
let fullResponse = '';
streamChat('deepseek', messages, (token) => {
    fullResponse += token;
    updateUI(fullResponse);
});

Node.js: Server-Side Streaming

const express = require('express');
const fetch = require('node-fetch');
const app = express();

app.post('/api/chat', async (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');

    const response = await fetch('https://tokenease.io/v1/chat/completions', {
        method: 'POST',
        headers: {
            'Authorization': `Bearer ${process.env.TOKENEASE_KEY}`,
            'Content-Type': 'application/json'
        },
        body: JSON.stringify({
            model: req.body.model || 'deepseek',
            messages: req.body.messages,
            stream: true
        })
    });

    const reader = response.body.getReader();
    
    while (true) {
        const { done, value } = await reader.read();
        if (done) {
            res.write('data: [DONE]\n\n');
            res.end();
            break;
        }
        res.write(value);
    }
});

app.listen(3000);

Model-Specific Streaming Behavior

Model	First Token	Streaming Notes
DeepSeek-V4	Fast	Includes reasoning_content stream for thinking steps
GLM-5.1	Very Fast	Smooth token-by-token output
Qwen-Plus	Very Fast	Lowest latency, best for real-time apps
Kimi-K3	Medium	Streams reasoning process before final answer
Doubao-Pro	Fast	Consistent streaming rate

Handling Reasoning Content

Some models (DeepSeek, Kimi) stream their internal reasoning before the final answer:

// DeepSeek reasoning_content handling
for line in response.iter_lines():
    if line:
        line = line.decode('utf-8')
        if line.startswith('data: '):
            data = line[6:]
            if data == '[DONE]':
                break
            
            chunk = json.loads(data)
            delta = chunk["choices"][0]["delta"]
            
            # Reasoning steps (thinking process)
            if "reasoning_content" in delta:
                reasoning += delta["reasoning_content"]
                # Optionally show "thinking..." indicator
            
            # Final answer tokens
            if "content" in delta:
                answer += delta["content"]
                print(delta["content"], end='', flush=True)

# Final state: reasoning contains the model's thought process
# answer contains the response shown to the user

UX Patterns for Streaming

1. Typing Indicator

Show a blinking cursor while waiting for the first token, then switch to streaming text:

// React example
const [isTyping, setIsTyping] = useState(false);
const [response, setResponse] = useState('');

const handleSend = async (message) => {
    setIsTyping(true);
    setResponse('');
    
    let firstToken = true;
    await streamChat('deepseek', [{role: 'user', content: message}], (token) => {
        if (firstToken) {
            setIsTyping(false);
            firstToken = false;
        }
        setResponse(prev => prev + token);
    });
};

// Render: {isTyping ?  : }

2. Token Counter

Show real-time token usage as the stream progresses:

let tokenCount = 0;
streamChat('deepseek', messages, (token) => {
    tokenCount++;
    updateUI(token);
    updateTokenCounter(tokenCount);
});

3. Cancel / Stop Generation

Allow users to abort long-running generations:

const controller = new AbortController();

const response = await fetch(API_URL, {
    method: 'POST',
    headers: { ... },
    body: JSON.stringify({ ... }),
    signal: controller.signal  // For cancellation
});

// User clicks "Stop"
function stopGeneration() {
    controller.abort();
}

// Python equivalent
import requests
response = requests.post(..., stream=True)
# To cancel: close the response
response.close()

Error Handling in Streams

Streaming connections can fail mid-generation. Handle these cases:

try:
    for line in response.iter_lines():
        if line:
            # ... process line ...
            pass
except requests.exceptions.ChunkedEncodingError:
    # Connection dropped mid-stream
    print("Connection lost. Retrying...")
    # Retry with partial context
    retry_with_context(full_response_so_far)
    
except requests.exceptions.Timeout:
    # Server took too long to start streaming
    print("Timeout. The model may be overloaded.")
    
except Exception as e:
    print(f"Stream error: {e}")

Performance Optimization

• Connection pooling: Reuse HTTP connections across requests to reduce handshake overhead
• Buffer size: Tune your SSE parser buffer. Too small = frequent parsing overhead. Too large = delayed display.
• Model selection: Qwen-Plus has the lowest time-to-first-token. Use it for latency-critical applications.
• max_tokens tuning: Set a reasonable max_tokens to prevent excessively long generations that feel sluggish

Conclusion

Streaming transforms the user experience of AI applications from frustrating waits to engaging, real-time interactions. With TokenEase, implementing SSE streaming is a single parameter change that works identically across DeepSeek, GLM, Qwen, Kimi, and Doubao.

Start with the Python requests example for prototyping, then move to the JavaScript fetch implementation for production web apps. Handle reasoning_content for models that support it, and always implement cancel/stop for user control.