Streaming

Overview

Chatty the Lab streams LLM responses token by token from the backend to the frontend over a persistent WebSocket connection. Each provider streams in its native format (SSE or NDJSON), which the backend normalizes into a uniform WebSocket message protocol.

WebSocket protocol

See the WebSocket API reference for the complete protocol specification.

The key message flow for streaming:

sequenceDiagram
    participant F as Frontend
    participant B as Backend
    participant P as LLM Provider

    F->>B: {"type":"query","chat_id":"...","content":"Hello"}
    B->>P: POST /chat/completions (stream=true)
    P-->>B: data: {"choices":[{"delta":{"content":"Hi"}}]}
    B-->>F: {"type":"token","chat_id":"...","content":"Hi"}
    P-->>B: data: {"choices":[{"delta":{"content":" there"}}]}
    B-->>F: {"type":"token","chat_id":"...","content":" there"}
    P-->>B: data: [DONE]
    B-->>F: {"type":"done","chat_id":"...","full_content":"Hi there"}

Backend streaming pipeline

1. Provider calls send_stream()

Each provider implementation makes an HTTP request with streaming enabled and returns a Pin<Box<dyn Stream<Item = Result<String, AppError>> + Send>>.

2. Stream parsing

Two utility functions in stream_utils.rs handle the two common streaming formats:

SSE (Server-Sent Events) — Used by OpenAI, Mistral, and Gemini:

data: {"choices":[{"delta":{"content":"Hello"}}]}

data: {"choices":[{"delta":{"content":" world"}}]}

data: [DONE]

The parse_sse_stream function:

• Buffers incoming bytes and splits on newlines
• Strips the data: prefix from each line
• Skips empty lines and data: [DONE]
• Parses the JSON and calls the provider’s extract_token closure

NDJSON (Newline-Delimited JSON) — Used by Ollama:

{"model":"llama3","message":{"content":"Hello"}}
{"model":"llama3","message":{"content":" world"}}
{"model":"llama3","done":true}

The parse_ndjson_stream function:

• Buffers incoming bytes and splits on newlines
• Parses each line as JSON
• Calls the provider’s extract_token closure

3. WebSocket forwarding

The handle_chat_query function in routes/ws.rs consumes the token stream and sends each token through an mpsc channel to the WebSocket sender task:

while let Some(result) = stream.next().await {
    match result {
        Ok(token) => {
            full_content.push_str(&token);
            tx.send(WsServerMessage::Token { chat_id, content: token }).await;
        }
        Err(e) => {
            tx.send(WsServerMessage::Error { chat_id: Some(chat_id), message: ... }).await;
            return;
        }
    }
}
tx.send(WsServerMessage::Done { chat_id, full_content }).await;

Provider implementations

OpenAI

// SSE streaming via POST https://api.openai.com/v1/chat/completions
// Token extraction: data["choices"][0]["delta"]["content"]
Ok(parse_sse_stream(resp.bytes_stream(), |data| {
    data["choices"][0]["delta"]["content"].as_str().map(|s| s.to_string())
}))

Mistral

Same SSE format as OpenAI, using the Mistral API endpoint.

Gemini

SSE streaming via Google’s generateContent endpoint. Token extraction path differs from OpenAI.

Ollama

// NDJSON streaming via POST http://ollama:11434/api/chat
// Token extraction: data["message"]["content"]
Ok(parse_ndjson_stream(resp.bytes_stream(), |data| {
    data["message"]["content"].as_str().map(|s| s.to_string())
}))

HuggingFace

HuggingFace Inference API does not support streaming. The send_stream implementation falls back to send() and emits a single token with the full response.

Frontend streaming

The frontend ws.ts module accumulates tokens in a Svelte store:

case 'token':
    tokens.update((t) => ({
        ...t,
        [msg.chat_id]: (t[msg.chat_id] || '') + msg.content
    }));
    break;

The ChatPanel component subscribes to this store and renders the accumulated text, producing a typewriter effect as tokens arrive. When a done message arrives, the component knows the response is complete.

Note

The WebSocket client automatically marks a chat as “done” when it receives an error message with a chat_id, so the UI unblocks even on failures.

Concurrent streaming

When using query_all, each chat query runs in its own tokio::spawn task. Multiple providers stream simultaneously, and their tokens are interleaved on the single WebSocket connection. The frontend demultiplexes by chat_id.