Streaming LLM Outputs: как работает real-time генерация

opensourceaillmstreamingsseit
← Back to Blog

Введение: почему streaming?

Первые LLM возвращали ответ после полного генерирования. Ждать 30 секунд — плохо. Streaming решает эту проблему.

Факт: GPT-4 генерирует ~50-100 токенов в секунду. Без streaming пользователь ждет 15-30 секунд на длинный ответ.


Что такое Streaming?

Traditional vs Streaming

Traditional (non-streaming):
  User: "Напиши эссе"
    |
    |  [30 секунд ожидания...]
    v
  Full answer: "Эссе полностью готово..."
  ✓ Просто
  ✗ Долго ждать

Streaming:
  User: "Напиши эссе"
    |
    v
  "Э" -> "Эт" -> "Это" -> "Это " -> ...
  ✓ Пользователь видит прогресс
  ✓ First token в 100ms
  ✗ Сложнее реализовать

Token-by-Token Generation

LLM генерирует по одному токену:

Step 1: "Hello"       (token: 994)
Step 2: "Hello,"      (token: 11)
Step 3: "Hello, world" (token: 1917)
Step 4: "Hello, world!" (token: 995)

Каждый шаг:
  1. Encode current tokens
  2. Predict next token distribution
  3. Sample from distribution
  4. Append to sequence
  5. Repeat

Уровень 1: SSE (Server-Sent Events)

Что такое SSE?

SSE = HTTP stream с сервера

HTTP/1.1
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

Response:
  data: {"token": "Hello"}

  data: {"token": ","}

  data: {"token": " world"}

  data: {"token": "!"}

  data: [DONE]

Server-side (Next.js)

// app/api/chat/route.ts
import { NextResponse } from 'next/server';

export async function POST(req: Request) {
  const { messages } = await req.json();
  
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages,
      stream: true,
    }),
  });

  const stream = new ReadableStream({
    async start(controller) {
      const reader = response.body!.getReader();
      
      while (true) {
        const { done, value } = await reader.read();
        if (done) {
          controller.close();
          return;
        }
        
        const text = new TextDecoder().decode(value);
        const lines = text.split('\n').filter(l => l.startsWith('data: '));
        
        for (const line of lines) {
          const data = line.slice(6);
          if (data === '[DONE]') continue;
          
          const json = JSON.parse(data);
          const token = json.choices?.[0]?.delta?.content;
          
          if (token) {
            controller.enqueue(
              `data: ${JSON.stringify({ token })}\n\n`
            );
          }
        }
      }
    },
  });

  return new NextResponse(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
}

Client-side (React)

'use client';
import { useState, useEffect, useRef } from 'react';

function Chat() {
  const [messages, setMessages] = useState([]);
  const [streaming, setStreaming] = useState(false);
  const [currentText, setCurrentText] = useState('');
  
  async function sendMessage(text: string) {
    setMessages([...messages, { role: 'user', content: text }]);
    setStreaming(true);
    setCurrentText('');
    
    const response = await fetch('/api/chat', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        messages: [...messages, { role: 'user', content: text }],
      }),
    });
    
    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() || '';
      
      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = line.slice(6);
          if (data === '[DONE]') continue;
          
          const json = JSON.parse(data);
          const token = json.token;
          if (token) {
            setCurrentText(prev => prev + token);
          }
        }
      }
    }
    
    setMessages(prev => [...prev, { role: 'assistant', content: currentText }]);
    setStreaming(false);
  }
  
  return <div>{/* chat UI */}</div>;
}

Уровень 2: React Server Components + useClient

Streaming in Next.js App Router

// app/chat/page.tsx (Server Component)
import { Suspense } from 'react';
import ChatStream from './ChatStream';

export default function ChatPage() {
  return (
    <Suspense fallback={<div>Loading...</div>}>
      <ChatStream />
    </Suspense>
  );
}
// app/chat/ChatStream.tsx (Client Component)
'use client';

import { createParser } from 'eventsource-parser';

export function ChatStream({ messages }: { messages: Message[] }) {
  const [response, setResponse] = useState('');
  
  useEffect(() => {
    const parser = createParser((event) => {
      if (event.type === 'event') {
        const data = JSON.parse(event.data);
        setResponse(prev => prev + data.token);
      }
    });
    
    fetch('/api/chat', {
      method: 'POST',
      body: JSON.stringify({ messages }),
    }).then(res => {
      const reader = res.body!.getReader();
      reader.read().then(processChunk);
    });
  }, []);
  
  return <div>{response}</div>;
}

Уровень 3: Streaming с Ollama / Local LLM

Ollama Streaming

# Ollama поддерживает streaming по умолчанию
curl http://localhost:11434/api/generate \
  -d '{
    "model": "llama3",
    "prompt": "What is streaming?",
    "stream": true
  }'
{"model":"llama3","created_at":"2026-05-04T10:00:00Z","response":"Stream","done":false}
{"model":"llama3","created_at":"2026-05-04T10:00:00Z","response":"ing","done":false}
{"model":"llama3","created_at":"2026-05-04T10:00:01Z","response":" is","done":false}
{"model":"llama3","created_at":"2026-05-04T10:00:01Z","response":" great","done":false}
{"model":"llama3","created_at":"2026-05-04T10:00:01Z","response":"!","done":true,"done_reason":"stop"}

Python (ollama library)

import ollama

stream = ollama.chat(
    model='llama3',
    messages=[{'role': 'user', 'content': 'Hello!'}],
    stream=True  # ← streaming
)

for chunk in stream:
    print(chunk['message']['content'], end='', flush=True)
# Output: HHelloello!! (token by token)

Python (llama.cpp)

from llama_cpp import Llama

llama = Llama(model_path="./llama3.gguf", n_ctx=2048)

output = llama.create_completion(
    prompt="Hello",
    max_tokens=100,
    stream=True  # ← generator
)

for chunk in output:
    token = chunk['choices'][0]['text']
    print(token, end='', flush=True)

Уровень 4: Parsing SSE Streams

Manual Parser

function parseSSE(raw: string) {
  const lines = raw.split('\n');
  const events = [];
  
  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const data = line.slice(6);
      if (data === '[DONE]') {
        events.push({ type: 'done' });
        continue;
      }
      try {
        events.push({ type: 'event', data: JSON.parse(data) });
      } catch (e) {
        // Skip malformed JSON
      }
    }
  }
  
  return events;
}

Using eventsource-parser

npm install eventsource-parser
import { createParser } from 'eventsource-parser';

const parser = createParser((event) => {
  if (event.type === 'event') {
    console.log(event.data);  // JSON string
  }
  if (event.type === 'done') {
    console.log('Stream complete');
  }
});

// Feed chunks to parser
parser.feed('data: {"token": "Hello"}\n\n');
parser.feed('data: {"token": "!"}\n\ndata: [DONE]\n\n');

Уровень 5: Streaming Different Response Types

Text Streaming

Simple text:
  data: {"token": "Hello"}
  data: {"token": " world"}
  data: [DONE]

JSON Streaming (structured)

Structured response:
  data: {"type": "thought", "content": "Let me think..."}
  data: {"type": "code", "language": "python", "content": "def hello():"}
  data: {"type": "code", "content": "    print('Hello')"}
  data: {"type": "answer", "content": "That's how you print!"}
  data: [DONE]

Markdown Streaming

Markdown rendering:
  data: {"token": "# Hello"}
  data: {"token": "\n\n**World**"}
  data: {"token": "!\n\n```python\n"}
  data: {"token": "print('hi')\n```"}
  data: [DONE]

Client renders markdown incrementally

Уровень 6: UX Patterns для Streaming

Typewriter Effect

function TypewriterText({ text }: { text: string }) {
  const [displayed, setDisplayed] = useState('');
  
  useEffect(() => {
    let i = 0;
    const interval = setInterval(() => {
      if (i < text.length) {
        setDisplayed(text.slice(0, i + 1));
        i++;
      } else {
        clearInterval(interval);
      }
    }, 16); // ~60fps
    
    return () => clearInterval(interval);
  }, [text]);
  
  return <span>{displayed}<span className="blink">|</span></span>;
}

Markdown Streaming

import ReactMarkdown from 'react-markdown';

function StreamingMarkdown({ rawText }: { rawText: string }) {
  return (
    <div className="prose">
      <ReactMarkdown>{rawText}</ReactMarkdown>
    </div>
  );
}

// Renders markdown as it streams in
// Code blocks get syntax highlighting

Loading States

function StreamingResponse({ tokens }: { tokens: string[] }) {
  if (tokens.length === 0) {
    return <ThinkingIndicator />;
  }
  
  return (
    <div>
      <Markdown text={tokens.join('')} />
      <CursorBlinker isActive={true} />
    </div>
  );
}

function ThinkingIndicator() {
  return (
    <div className="flex gap-1">
      <span className="dot animate-bounce" style={{ animationDelay: '0ms' }}>.</span>
      <span className="dot animate-bounce" style={{ animationDelay: '150ms' }}>.</span>
      <span className="dot animate-bounce" style={{ animationDelay: '300ms' }}>.</span>
    </div>
  );
}

Уровень 7: Performance Optimization

Token Prediction Overlap

Time →

TTFT (Time To First Token):
  Prompt Processing: [||||||||]
  First Token:       [●]
  
TTFT = 500ms typical (GPT-4)
TTFT = 100-300ms typical (local LLM)

Streaming:
  Token 1: [●]
  Token 2:    [●]
  Token 3:       [●]
  Token 4:          [●]
  
Tokens/sec: 20-100 depending on model

Optimizations

# 1. Speculative decoding (already covered)
# Reduces TTFT by predicting ahead

# 2. KV Cache reuse
# Reuse cache for same prompt

# 3. Batch streaming
# Stream multiple responses together

# 4. Prefetching
# Start generating while UI renders

Network Optimization

// Compress SSE messages
// Use MessagePack instead of JSON
// Reduce overhead from ~30 bytes to ~10 bytes per token

// WebSocket vs SSE:
// SSE: simple, auto-reconnect, HTTP/1.1
// WebSocket: bidirectional, HTTP/1.1 or 2, more complex

Уровень 8: Error Handling

Stream Errors

try {
  const response = await fetch('/api/chat', {
    method: 'POST',
    body: JSON.stringify({ messages }),
  });
  
  if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
  }
  
  const reader = response.body!.getReader();
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    // Handle partial data on error
    processChunk(value);
  }
} catch (error) {
  // Show error + partial response
  showError(error);
  showPartialResponse(buffer);
}

Retry Logic

async function fetchWithRetry(url: string, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await fetch(url);
    } catch (e) {
      if (i === maxRetries - 1) throw e;
      await new Promise(r => setTimeout(r, 1000 * (i + 1)));
    }
  }
}

Уровень 9: Streaming Metrics

Key Metrics

1. TTFT (Time To First Token):
   - Время от запроса до первого токена
   - Цель: < 500ms (cloud), < 300ms (local)
   
2. Tokens Per Second (TPS):
   - Скорость генерации
   - GPT-4: 20-50 TPS
   - Local LLM: 10-100 TPS (зависит от GPU)
   
3. Total Latency:
   - Общее время генерации
   - 100 tokens / 50 TPS = 2 seconds
   
4. Time To Full Response:
   - TTFT + (tokens / TPS)

Monitoring

import time

start = time.time()
first_token = None
tokens = []

for chunk in stream:
    if first_token is None:
        first_token = time.time()
        ttft = first_token - start
    
    tokens.append(chunk['token'])

total_time = time.time() - start
tps = len(tokens) / (total_time - ttft)

print(f"TTFT: {ttft*1000:.0f}ms")
print(f"TPS: {tps:.1f}")
print(f"Total: {total_time*1000:.0f}ms")

Уровень 10: Advanced Patterns

Multi-Stream

Multiple LLMs, pick best:

Stream A: [LLM A response] ──┐
                               ├──→ Compare → Best
Stream B: [LLM B response] ──┘

# Or: Parallel generation
# User sees first complete answer

Streaming Tool Calls

Streaming with function calls:

data: {"type": "tool_call", "name": "search", "args": {"q": "news"}}
data: {"type": "tool_result", "results": [...]}
data: {"type": "token", "content": "Based on the search..."}
data: [DONE]

Cancellation

const controller = new AbortController();

async function cancelStream() {
  controller.abort();
  // Server receives abort signal
  // Can stop generation early
}

fetch('/api/chat', {
  method: 'POST',
  signal: controller.signal,
  body: JSON.stringify({ messages }),
});

Summary

Streaming = Better UX + Faster feedback

10 уровней:
  1. SSE        - Server-Sent Events basics
  2. React      - streaming in React
  3. Local LLM  - Ollama, llama.cpp streaming
  4. Parsing    - SSE stream parsing
  5. Types      - text, JSON, markdown
  6. UX         - typewriter, markdown, loading
  7. Perf       - TTFT, TPS, optimization
  8. Errors     - retry, partial data
  9. Metrics    - TTFT, TPS monitoring
  10. Advanced  - multi-stream, tools, cancel

Главное правило:
  First Token < 500ms
  Smooth streaming > Fast full response

</final_file_content>
<task_progress>
- [x] Analyze existing article format and topics
- [x] Write article 1: RAG Systems (done)
- [x] Write article 2: Streaming LLM Outputs (done)
- [ ] Write article 3: Function Calling / Tools
- [ ] Write article 4: Context Window Management
- [ ] Write article 5: Prompt Engineering Techniques
- [ ] Write article 6: Local LLM Deployment
- [ ] Write article 7: LLM Evaluation Metrics
- [ ] Write article 8: Multi-Agent Systems
- [ ] Write article 9: LLM Security & Safety
- [ ] Write article 10: LLM Monitoring & Observability
- [ ] Verify all articles are created correctly
</task_progress>
</write_to_file>