Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla
Rakenna FastAPI-päätepiste, joka välittää LLM:n suoratoistovastaukset selainasiakkaalle StreamingResponse-olion ja text/event-stream-sisältötyypin avulla.
Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla on ilmainen AI Engineering Academy-oppitunti CoddyKitissä. Tämä on oppitunti 3/4. Voit lukea koko oppitunnin alta ilmaiseksi ja harjoitella sen jälkeen käytännössä selaimessa sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla. Oppitunti kuuluu AI Engineering Academy-oppimispolkuun, ja edistymisesi synkronoituu verkon ja CoddyKit-sovelluksen välillä. AI Engineering Academy-kurssilla on yhteensä 4 oppituntia.
Miksi Server-Sent Events sopii LLM-streamaukseen
Server-Sent Events (SSE) on W3C-standardi, jonka avulla palvelin voi lähettää tekstimuotoisia tapahtumia streamina selainasiakkaalle yhden pitkäkestoisen HTTP-yhteyden kautta. Toisin kuin WebSocketit, SSE on yksisuuntainen (palvelimelta asiakkaalle), toimii tavallisen HTTP/1.1:n yli, muodostaa yhteyden automaattisesti uudelleen katkosten jälkeen eikä vaadi erityistä selainkirjastoa. Näiden ominaisuuksien ansiosta se on ihanteellinen siirtotapa LLM-tokenien streamaamiseen FastAPI-taustapalvelusta verkkokäyttöliittymään.
SSE-johtomuoto
SSE lähettää tekstidataa kenttinä, jotka erotetaan toisistaan rivinvaihdoilla. Kukin tapahtuma sisältää valinnaisen event-tyyppikentän, hyötykuorman sisältävän data-kentän ja valinnaisen uudelleenyhdistämiseen tarkoitetun id-kentän. Tapahtumat erotetaan tyhjällä rivillä. LLM-streamauksessa lähettäkää kukin token muodossa data: token_text\n\n ja lähettäkää lopussa erityinen data: [DONE]\n\n-tapahtuma ilmoittamaan streamin valmistumisesta.
# 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'.FastAPI:n StreamingResponse
FastAPI:n StreamingResponse hyväksyy merkkijonoja palauttavan asynkronisen generaattorin ja streamaa ne asiakkaalle. Asettamalla media_type-arvoksi 'text/event-stream' ja muotoilemalla jokaisen palautetun merkkijonon SSE-tapahtumaksi muutatte minkä tahansa asynkronisen generaattorin asianmukaiseksi SSE-streamiksi. FastAPI käsittelee yhteyden elinkaaren, puskurin tyhjennyksen ja HTTP-otsakkeet automaattisesti.
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'},
)SSE:n tärkeät HTTP-otsakkeet
Kolme HTTP-otsaketta ovat ratkaisevan tärkeitä, jotta SSE toimii oikein välityspalvelinten ja CDN-palvelujen kautta. Cache-Control: no-cache estää välikäsiä tallentamasta streamia välimuistiin. Connection: keep-alive pitää TCP-yhteyden avoinna. X-Accel-Buffering: no poistaa käytöstä Nginxin vastausten puskuroinnin, joka muuten niputtaisi osat yhteen ja kumoaisi streamauksen hyödyn. Ilman viimeistä otsaketta Nginx puskuroiden kaiken tulosteen ennen sen välittämistä selaimelle.
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,
)Rakenteiset SSE-tapahtumat JSON-hyötykuormilla
Monipuolisemmissa streaming-rajapinnoissa koodatkaa kunkin tapahtuman hyötykuorma JSON-muotoon raakatekstin sijaan. Näin voitte sisällyttää tokenin rinnalle metatietoja — esimerkiksi tokenin tyypin (sisältö vai työkalukutsu), viestin tunnisteen tai viiveaikaleiman. Selainasiakas jäsentää kunkin tapahtuman JSON-tiedot ja ohjaa eri tapahtumatyypit eri käyttöliittymäkomponenteille.
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'SSE:n käyttäminen selaimessa (JavaScript)
Selaimen EventSource-rajapinta muodostaa yhteyden SSE-päätepisteeseen ja laukaisee tapahtumia niiden saapuessa. Tokenien streamausta varten kuunnelkaa oletusarvoista message-tapahtumaa, jäsentäkää data JSON-muodossa tai käsitelkää sitä raakana merkkijonona ja lisätkää kukin token DOM:iin. Käsitelkää [DONE]-sentinelli sulkemalla EventSource-yhteys.
// 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();
};POST-pyynnöt streamaukseen fetchillä
EventSource tukee vain GET-pyyntöjä, mikä rajoittaa monimutkaisten promptien käyttöä. POST-pyynnöissä (kun lähetätte JSON-rungon keskusteluhistorian kanssa) lukekaa vastausrunk g selaimen fetch-rajapinnan ja Streams API:n avulla osissa. Tätä mallia käyttävät ChatGPT:n verkkokäyttöliittymä ja useimmat tuotantokäyttöön tarkoitetut LLM-chat-käyttöliittymät.
// 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);
}
}
}
}FastAPI:n POST-päätepiste chat-streamausta varten
Määrittäkää POST-pohjaista chat-streamausta varten Pydantic-malli pyynnön rungolle, vastaanottakaa viestilista ja streamatkaa LLM:n vastaus. Näin voitte välittää koko keskusteluhistorian jokaisen pyynnön mukana ja tukea monivuorovaikutteisia chat-sovelluksia. Malli on sama kuin GET-streamauksessa, paitsi että prompti poimitaan pyynnön rungosta.
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,
)Asiakkaan yhteyden katkeamisen käsittely
Kun selainkäyttäjä siirtyy pois sivulta tai sulkee välilehden, HTTP-yhteys sulkeutuu ja FastAPI nostaa asyncio.CancelledError-poikkeuksen streaming-generaattorissa. Käsitelkää tämä aina, jotta LLM-streaming-pyynnöt eivät jää avoimiksi ja aiheuta tarpeettomia API-kuluja. Kapseloikaa generaattori CancelledError-poikkeuksen käsittelevään try/except-rakenteeseen ja perukaa OpenAI-stream, kun poikkeama havaitaan.
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')Pyyntöjen autentikoinnin lisääminen
Tuotantokäyttöön tarkoitetut streaming-päätepisteet on autentikoitava luvattoman LLM-käytön estämiseksi. Käyttäkää FastAPI:n Depends-toimintoa API-avaimen tai JWT-otsakkeen tarkistukseen. Autentikointi tapahtuu ennen generaattorin käynnistymistä, joten sen aiheuttama lisäkustannus on pieni ja stream alkaa vasta käyttäjän varmentamisen jälkeen.
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)SSE-päätepisteiden testaaminen
Testatkaa streaming-päätepisteitä FastAPI:n TestClient-asiakkaalla streaming-tilassa. Käyttäkää rakennetta with client.stream('GET', '/stream', params={...}) as r ja käykää r.iter_lines()-tulokset läpi SSE-tapahtumien vastaanottamiseksi. Näin voitte varmistaa, että tokenit muotoillaan oikein, DONE-sentinelli lähetetään ja virhetilanteet tuottavat asianmukaiset SSE-virhetapahtumat.
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) > 0Pikatesti
Testatkaa tässä oppitunnissa opitun FastAPI:n SSE-streamauksen ymmärtämistä.
Oppitunnin yhteenveto
Tässä oppitunnissa opitte, että Server-Sent Events on standardoitu HTTP-siirtotapa LLM-tokenien streamaamiseen selainasiakkaille, StreamingResponse ja text/event-stream muuttavat minkä tahansa asynkronisen generaattorin SSE-streamiksi FastAPI:ssa ja kriittiset otsakkeet, kuten X-Accel-Buffering ja Cache-Control, ovat välttämättömiä toiminnan varmistamiseksi välityspalvelinten takana. Käsitelkää asiakkaan yhteyden katkeamiset, jotta orvoiksi jääviä LLM API -kutsuja ei synny. Seuraavaksi käsittelemme streaming-vastauksia, jotka sisältävät työkalukutsuja.
Opi Python tekoälytuutorin avulla — ilmaiseksi
Kirjoita ja suorita oikeaa koodia selaimessa, saa välitöntä apua tekoälytuutorilta ympäri vuorokauden ja jatka siitä, mihin jäit, verkossa tai sovelluksessa.
- Kurssit
- 30
- Oppitunnit
- 120
Usein kysytyt kysymykset
Onko oppitunti ”Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla” ilmainen?
Kyllä – oppitunnin ”Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla” koko tekstin voi lukea täällä verkossa ilmaiseksi. Jos haluat harjoitella interaktiivisesti sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla sekä avata koko AI Engineering Academy-kurssin, päivitä CoddyKit PROhon. AI Engineering Academy-kurssilla on yhteensä 4 oppituntia.
Mitä opin oppitunnilla ”Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla”?
Rakenna FastAPI-päätepiste, joka välittää LLM:n suoratoistovastaukset selainasiakkaalle StreamingResponse-olion ja text/event-stream-sisältötyypin avulla. Harjoittelet AI Engineering Academy-aihetta koodilla, jonka suoritat suoraan selaimessa. Ympäri vuorokauden käytettävissä oleva tekoälytuutori vastaa kysymyksiisi oppitunnin aikana.
Tarvitsenko kokemusta aloittaakseni AI Engineering Academy-opiskelun?
Aiempi kokemus ei ole tarpeen. CoddyKitin AI Engineering Academy-oppimispolku sopii vasta-alkajista edistyneisiin, joten voit aloittaa tästä tai alusta ja edetä omaan tahtiisi. Tämä on oppitunti 3/4.
Kuinka kauan ”Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla”-oppitunnin suorittaminen kestää?
Useimmat CoddyKitin oppitunnit kestävät noin 5–10 minuuttia. Jokainen oppitunti on lyhyt ja interaktiivinen, joten edistyt tasaisesti ja voit jatkaa siitä, mihin jäit – sekä verkossa että sovelluksessa.
Voinko kirjoittaa ja suorittaa koodia tällä AI Engineering Academy-oppitunnilla?
Kyllä. Jokainen AI Engineering Academy-oppitunti sisältää sisäänrakennetun koodieditorin, joten voit kirjoittaa ja suorittaa oikeaa koodia suoraan selaimessa ja saada välitöntä palautetta tekoälyltä – paikallista asennusta ei tarvita.
Kaikki tämän kurssin oppitunnit
- Tokenien suoratoiston ymmärtäminen
- Suoratoistojen käsittely Python SDK:lla
- Suoratoisto FastAPIssa Server-Sent Events -tapahtumilla
- Työkalukutsujen käsittely suoratoistovastauksissa