Streaming API позволяет получать ответ от языковой модели по частям, по мере генерации, вместо ожидания полного ответа. Это критически важно для чат-ботов и интерактивных интерфейсов, где пользователь хочет видеть текст в реальном времени. В этой статье разберём, как работает потоковая генерация и как реализовать её через ModelSwitch.
Как работает Streaming API
При обычном (non-streaming) запросе клиент отправляет промпт и ждёт, пока модель полностью сгенерирует ответ. Для длинных ответов это может занять 10-30 секунд. При streaming сервер отправляет токены по мере генерации через протокол Server-Sent Events (SSE).
Каждый чанк (chunk) приходит в формате:
data: {"id":"chatcmpl-abc","choices":[{"delta":{"content":"Привет"},"index":0}]}
data: {"id":"chatcmpl-abc","choices":[{"delta":{"content":" мир"},"index":0}]}
data: [DONE]Поле delta содержит только новый фрагмент текста, а не весь ответ целиком. Последний чанк [DONE] сигнализирует о завершении генерации.
Streaming на Python
OpenAI SDK для Python поддерживает streaming из коробки. Достаточно передать параметр stream=True:
from openai import OpenAI
client = OpenAI(
base_url="https://api.modelswitch.ru/v1",
api_key="msk_ваш_ключ"
)
stream = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "Ты полезный ассистент."},
{"role": "user", "content": "Напиши пошаговый план изучения Python"}
],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end="", flush=True)
print() # Перенос строки в концеДля асинхронного клиента используйте AsyncOpenAI:
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url="https://api.modelswitch.ru/v1",
api_key="msk_ваш_ключ"
)
async def stream_response():
stream = await client.chat.completions.create(
model="claude-3.5-sonnet",
messages=[{"role": "user", "content": "Объясни рекурсию"}],
stream=True
)
async for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
asyncio.run(stream_response())Streaming на TypeScript
В TypeScript SDK потоковая генерация реализована через асинхронные итераторы:
import OpenAI from "openai";
const client = new OpenAI({
baseURL: "https://api.modelswitch.ru/v1",
apiKey: "msk_ваш_ключ",
});
async function streamChat() {
const stream = await client.chat.completions.create({
model: "gpt-4o",
messages: [{ role: "user", content: "Напиши план изучения TypeScript" }],
stream: true,
});
for await (const chunk of stream) {
const content = chunk.choices[0]?.delta?.content;
if (content) {
process.stdout.write(content);
}
}
console.log();
}
streamChat();Для веб-приложений на React можно передавать чанки в состояние компонента:
const [text, setText] = useState("");
async function handleSubmit(prompt: string) {
const response = await fetch("https://api.modelswitch.ru/v1/chat/completions", {
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: "Bearer msk_ваш_ключ",
},
body: JSON.stringify({
model: "gpt-4o",
messages: [{ role: "user", content: prompt }],
stream: true,
}),
});
const reader = response.body!.getReader();
const decoder = new TextDecoder();
let result = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
// Парсим SSE-формат
const lines = chunk.split("\n").filter((l) => l.startsWith("data: "));
for (const line of lines) {
const data = line.slice(6);
if (data === "[DONE]") break;
const parsed = JSON.parse(data);
const content = parsed.choices[0]?.delta?.content || "";
result += content;
setText(result);
}
}
}Советы и типичные ошибки
При работе со streaming важно учитывать несколько нюансов:
- Обработка finish_reason — последний чанк содержит
finish_reason: "stop". Проверяйте его для корректного завершения. - Таймауты — при streaming соединение остаётся открытым долго. Убедитесь, что таймаут клиента достаточно большой (60-120 секунд).
- Буферизация — некоторые прокси (Nginx, Cloudflare) могут буферизовать SSE. Добавьте заголовок
X-Accel-Buffering: no. - Подсчёт токенов — при streaming информация о токенах приходит только в последнем чанке. Не считайте токены вручную из чанков.
- Обработка ошибок — ошибка может прийти в середине потока. Оберните итерацию в try/catch.
try:
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
except Exception as e:
print(f"\nОшибка при streaming: {e}")Streaming через ModelSwitch работает со всеми поддерживаемыми моделями: GPT-4o, Claude, Gemini, Llama, Mistral и другими. Один формат запроса, один API-ключ, потоковая генерация из коробки.