0Pricing
AI Engineering Academy · Lekcja

Streaming w FastAPI z Server-Sent Events

Zbuduj endpoint FastAPI, który przekazuje strumieniowane odpowiedzi LLM do klienta przeglądarkowego za pomocą StreamingResponse i typu zawartości text/event-stream.

Streaming w FastAPI z Server-Sent Events to bezpłatna lekcja AI Engineering Academy na CoddyKit. To lekcja 3 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Engineering Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Dlaczego Server-Sent Events do strumieniowania LLM

Server-Sent Events (SSE) to standard W3C umożliwiający serwerowi przesyłanie strumienia zdarzeń tekstowych do klienta w przeglądarce za pośrednictwem jednego długotrwałego połączenia HTTP. W przeciwieństwie do WebSocketów SSE działa jednokierunkowo, od serwera do klienta, korzysta ze standardowego HTTP/1.1, automatycznie ponawia połączenie po rozłączeniu i nie wymaga specjalnej biblioteki przeglądarkowej. Dzięki tym właściwościom jest idealnym mechanizmem transportowym do strumieniowania tokenów LLM z backendu FastAPI do frontendowej części aplikacji.

Format SSE na poziomie protokołu

SSE przesyła dane tekstowe sformatowane jako serie pól oddzielonych znakami nowego wiersza. Każde zdarzenie zawiera opcjonalne pole typu event, pole data z danymi oraz opcjonalne pole id używane przy ponownym łączeniu. Zdarzenia są oddzielone pustym wierszem. W przypadku strumieniowania LLM każdy token należy wysyłać jako wiersz data: token_text\n\n, a na końcu jako sygnał zakończenia strumienia wysłać specjalne zdarzenie data: [DONE]\n\n.

# SSE wire format example
'''
data: The\n\n
data:  capital\n\n
data:  of\n\n
data:  France\n\n
data:  is\n\n
data:  Paris\n\n
data: [DONE]\n\n
'''

# Each 'data:' line is one event.
# The double newline (\n\n) terminates each event.
# The client receives these as EventSource message events.
# The content-type must be 'text/event-stream'.

StreamingResponse w FastAPI

StreamingResponse w FastAPI przyjmuje asynchroniczny generator zwracający ciągi znaków i przesyła je strumieniowo do klienta. Ustawiając media_type na 'text/event-stream' i formatując każdy zwracany ciąg jako zdarzenie SSE, można przekształcić dowolny asynchroniczny generator w prawidłowy strumień SSE. FastAPI automatycznie obsługuje cykl życia połączenia, opróżnianie bufora i nagłówki HTTP.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio

app = FastAPI()
async_client = AsyncOpenAI()

async def token_generator(prompt: str):
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield f'data: {delta}\n\n'  # SSE format
    yield 'data: [DONE]\n\n'

@app.get('/stream')
async def stream_endpoint(prompt: str):
    return StreamingResponse(
        token_generator(prompt),
        media_type='text/event-stream',
        headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
    )

Ważne nagłówki HTTP dla SSE

Trzy nagłówki HTTP mają kluczowe znaczenie dla prawidłowego działania SSE za pośrednictwem serwerów proxy i CDN-ów. Cache-Control: no-cache zapobiega buforowaniu strumienia przez elementy pośredniczące. Connection: keep-alive utrzymuje otwarte połączenie TCP. X-Accel-Buffering: no wyłącza buforowanie odpowiedzi przez Nginx, które w przeciwnym razie grupowałoby fragmenty i niweczyło efekt strumieniowania. Bez tego ostatniego nagłówka Nginx zbuforuje całe dane wyjściowe przed przekazaniem ich do przeglądarki.

from fastapi.responses import StreamingResponse

SSE_HEADERS = {
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',   # disable nginx buffering
    'Access-Control-Allow-Origin': '*',  # CORS for cross-origin clients
}

@app.get('/chat')
async def chat_stream(prompt: str):
    return StreamingResponse(
        token_generator(prompt),
        media_type='text/event-stream',
        headers=SSE_HEADERS,
    )

Ustrukturyzowane zdarzenia SSE z danymi JSON

W przypadku bardziej rozbudowanych interfejsów API do strumieniowania dane każdego zdarzenia należy kodować jako JSON, a nie jako surowy tekst. Pozwala to dołączyć do tokenu metadane, na przykład jego typ, czyli treść lub wywołanie narzędzia, identyfikator wiadomości albo znacznik czasu opóźnienia. Klient przeglądarkowy analizuje JSON każdego zdarzenia i kieruje różne typy zdarzeń do odpowiednich komponentów interfejsu.

import json
import time

async def json_token_generator(prompt: str, session_id: str):
    t_start = time.perf_counter()
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            payload = json.dumps({
                'type': 'token',
                'content': delta,
                'session_id': session_id,
                't_ms': round((time.perf_counter() - t_start) * 1000),
            })
            yield f'data: {payload}\n\n'
    # Send completion event
    yield f'data: {json.dumps({"type": "done", "session_id": session_id})}\n\n'

Odbieranie SSE w przeglądarce (JavaScript)

Przeglądarkowe API EventSource łączy się z endpointem SSE i wywołuje zdarzenia w miarę ich nadejścia. W przypadku strumieniowania tokenów należy nasłuchiwać domyślnego zdarzenia message, analizować dane jako JSON lub traktować je jako surowy ciąg znaków, a następnie dołączać każdy token do DOM. Po odebraniu znacznika [DONE] należy zamknąć połączenie EventSource.

// Browser-side JavaScript
const prompt = 'Explain hybrid search in one paragraph.';
const url = '/stream?prompt=' + encodeURIComponent(prompt);

const source = new EventSource(url);
const output = document.getElementById('output');

source.onmessage = (event) => {
  if (event.data === '[DONE]') {
    source.close();  // stop listening
    return;
  }
  output.textContent += event.data;  // append each token
};

source.onerror = (err) => {
  console.error('SSE error:', err);
  source.close();
};

Żądania POST z użyciem fetch do strumieniowania

EventSource obsługuje wyłącznie żądania GET, co ogranicza możliwość korzystania ze złożonych promptów. W przypadku żądań POST, które przesyłają treść JSON z historią konwersacji, należy użyć przeglądarkowego API fetch wraz z Streams API, aby stopniowo odczytywać treść odpowiedzi. Ten wzorzec jest używany przez interfejs WWW ChatGPT i większość produkcyjnych interfejsów czatowych LLM.

// Browser-side: POST with fetch and ReadableStream
async function streamPost(messages) {
  const response = await fetch('/chat', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({messages}),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  const output = document.getElementById('output');

  while (true) {
    const {done, value} = await reader.read();
    if (done) break;
    const text = decoder.decode(value, {stream: true});
    // Parse SSE lines
    for (const line of text.split('\n')) {
      if (line.startsWith('data: ') && line !== 'data: [DONE]') {
        output.textContent += line.slice(6);
      }
    }
  }
}

Endpoint POST FastAPI do strumieniowania czatu

W przypadku strumieniowania czatu opartego na POST należy zdefiniować model Pydantic dla treści żądania, przyjąć listę wiadomości i strumieniować odpowiedź LLM. Umożliwia to przekazywanie pełnej historii konwersacji przy każdym żądaniu, co pozwala obsługiwać aplikacje czatowe z wieloma turami rozmowy. Wzorzec jest identyczny jak w przypadku strumieniowania GET, z wyjątkiem tego, że prompt jest pobierany z treści żądania.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = 'gpt-4o-mini'

@app.post('/chat')
async def chat_post(request: ChatRequest):
    async def generate():
        stream = await async_client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f'data: {delta}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(
        generate(),
        media_type='text/event-stream',
        headers=SSE_HEADERS,
    )

Obsługa rozłączeń klienta

Gdy użytkownik przeglądarki przejdzie na inną stronę lub zamknie kartę, połączenie HTTP zostaje zamknięte, a FastAPI zgłasza asyncio.CancelledError w generatorze strumieniującym. Należy zawsze to obsługiwać, aby nie pozostawiać otwartych żądań strumieniowania LLM i nie ponosić niepotrzebnych kosztów API. Generator należy opakować w try/except dla CancelledError i po jego wykryciu anulować strumień OpenAI.

from fastapi import Request

@app.get('/stream')
async def stream_with_disconnect(prompt: str, request: Request):
    async def generate_with_cancel():
        try:
            stream = await async_client.chat.completions.create(
                model='gpt-4o-mini',
                messages=[{'role': 'user', 'content': prompt}],
                stream=True,
            )
            async for chunk in stream:
                if await request.is_disconnected():
                    break  # client gone, stop generating
                delta = chunk.choices[0].delta.content
                if delta:
                    yield f'data: {delta}\n\n'
        except asyncio.CancelledError:
            pass  # client disconnected
        finally:
            yield 'data: [DONE]\n\n'

    return StreamingResponse(generate_with_cancel(), media_type='text/event-stream')

Dodawanie uwierzytelniania żądań

Produkcyjne endpointy strumieniowania muszą uwierzytelniać żądania, aby zapobiegać nieautoryzowanemu użyciu LLM. Należy użyć Depends w FastAPI wraz z kluczem API lub sprawdzaniem nagłówka JWT. Uwierzytelnianie odbywa się przed uruchomieniem generatora, dlatego narzut jest minimalny, a strumień rozpoczyna się dopiero po zweryfikowaniu użytkownika.

from fastapi import Header, HTTPException, Depends

VALID_API_KEYS = {'sk-demo-key-1', 'sk-demo-key-2'}

async def verify_api_key(x_api_key: str = Header(None)):
    if x_api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail='Invalid API key')
    return x_api_key

@app.post('/chat')
async def authenticated_chat(
    request: ChatRequest,
    api_key: str = Depends(verify_api_key),
):
    async def generate():
        stream = await async_client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f'data: {delta}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(generate(), media_type='text/event-stream', headers=SSE_HEADERS)

Testowanie endpointów SSE

Endpointy strumieniowania należy testować za pomocą TestClient z FastAPI w trybie strumieniowania. Użyj with client.stream('GET', '/stream', params={...}) as r i iteruj po r.iter_lines(), aby odbierać zdarzenia SSE. Pozwala to sprawdzić, czy tokeny są prawidłowo formatowane, znacznik DONE jest wysyłany, a przypadki błędów generują odpowiednie zdarzenia błędów SSE.

from fastapi.testclient import TestClient

def test_sse_endpoint():
    with TestClient(app) as client:
        with client.stream('GET', '/stream', params={'prompt': 'Say hi'}) as r:
            assert r.status_code == 200
            assert 'text/event-stream' in r.headers['content-type']

            events = []
            for line in r.iter_lines():
                if line.startswith('data: '):
                    events.append(line[6:])

            assert events[-1] == '[DONE]'
            full_text = ''.join(e for e in events if e != '[DONE]')
            assert len(full_text) > 0

Szybki test

Sprawdź swoją wiedzę na temat strumieniowania FastAPI z użyciem SSE omówionego w tej lekcji.

Podsumowanie lekcji

W tej lekcji nauczyli się Państwo, że: Server-Sent Events to standardowy mechanizm transportowy HTTP do strumieniowania tokenów LLM do klientów przeglądarkowych; StreamingResponse z text/event-stream przekształca dowolny asynchroniczny generator w strumień SSE w FastAPI; a kluczowe nagłówki, w tym X-Accel-Buffering i Cache-Control, są niezbędne do prawidłowego działania za serwerami proxy. Należy obsługiwać rozłączenia klientów, aby unikać osieroconych wywołań API LLM. Następnie zajmiemy się strumieniowanymi odpowiedziami zawierającymi wywołania narzędzi.

Często zadawane pytania

Czy lekcja „Streaming w FastAPI z Server-Sent Events” jest bezpłatna?

Tak — pełny tekst „Streaming w FastAPI z Server-Sent Events” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Engineering Academy, przejdź na CoddyKit PRO. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Co nauczysz się w „Streaming w FastAPI z Server-Sent Events”?

Zbuduj endpoint FastAPI, który przekazuje strumieniowane odpowiedzi LLM do klienta przeglądarkowego za pomocą StreamingResponse i typu zawartości text/event-stream. Ćwiczysz AI Engineering Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Engineering Academy?

Nie wymagamy żadnego doświadczenia. AI Engineering Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 3 z 4.

Ile czasu zajmuje lekcja „Streaming w FastAPI z Server-Sent Events”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Engineering Academy?

Tak. Każda lekcja AI Engineering Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Zrozumienie strumieniowania tokenów
  2. Obsługa strumieni za pomocą Python SDK
  3. Streaming w FastAPI z Server-Sent Events
  4. Obsługa wywołań narzędzi w strumieniowanych odpowiedziach
← Powrót do AI Engineering Academy