Streaming LLM Outputs: как работает real-time генерация
opensourceaillmstreamingsseit
Введение: почему 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>