FastAPI SSE streaming response endpoint બનાવો

પરિચય

જ્યારે તમે એવું ચેટ UI શિપ કરો છો જે રેન્ડર કરતાં પહેલાં સંપૂર્ણ LLM પ્રતિસાદની રાહ જુએ છે, ત્યારે વપરાશકર્તાઓને અનેક સેકન્ડના ફ્રીઝનો અનુભવ થાય છે અને તેઓ સેશન છોડી દે છે — ભલે કુલ લેટન્સી સ્ટ્રીમ થયેલા પ્રતિસાદ જેટલી જ હોય. Server-Sent Events (SSE) તમારા FastAPI બેકએન્ડને દરેક ટોકન અપસ્ટ્રીમ મોડેલમાંથી આવે તે જ ક્ષણે બ્રાઉઝર સુધી પુશ કરવા દે છે, જેથી અનુભવાતો ફ્રીઝ દૂર થાય છે અને ક્લાયન્ટ ડિસ્કનેક્ટ શોધવા માટે તમને એક સ્પષ્ટ બિંદુ મળે છે, જેથી કોઈ વાંચતું ન હોય તેવા ટોકન્સ માટે તમે ચૂકવણી કરવાનું બંધ કરી શકો. આ પાઠના અંત સુધીમાં તમે HTTP પર W3C-સુસંગત SSE સ્ટ્રીમિંગ એન્ડપોઇન્ટ અમલમાં મૂકી શકશો જે LLM ટોકન્સને ક્રમશઃ પહોંચાડે, રિવર્સ પ્રોક્સીઓને બાઇટ્સ તરત ફોરવર્ડ કરવા માટે જરૂરી હેડર્સ સેટ કરે, અને જ્યારે ક્લાયન્ટ કનેક્શન બંધ કરે ત્યારે સ્વચ્છ રીતે બહાર નીકળે.

મુખ્ય પરિભાષા

• Server-Sent Events (SSE): એક W3C સ્ટ્રીમિંગ પ્રોટોકોલ જે Content-Type: text/event-stream સાથે એક જ લાંબા સમય સુધી જીવંત રહેતા HTTP/1.1 પ્રતિસાદ પર વહન થાય છે, જેમાં સર્વર ન્યૂલાઇનથી અલગ કરેલા event: / data: ફ્રેમ્સ ક્લાયન્ટને પુશ કરે છે જ્યાં સુધી કોઈ એક બાજુ કનેક્શન બંધ ન કરે.
• SSE ફ્રેમ: સ્ટ્રીમનું એક એકમ, જે એક અથવા વધુ ફીલ્ડ લાઇનો (દા.ત. event: token, data: {...}) થી બનેલું હોય છે અને ખાલી લાઇન (\n\n) થી સમાપ્ત થાય છે; પાછળની ખાલી લાઇન છોડી દેવાથી ક્લાયન્ટ્સ અનિશ્ચિત સમય માટે બફર કરે છે.
• StreamingResponse: FastAPI નો પ્રતિસાદ class જે async જનરેટરનો ઉપયોગ કરે છે અને દરેક yield થયેલા ચંકને સીધો સોકેટમાં લખે છે; અહીં પ્રોક્સી બફરિંગને હરાવવા માટે media_type="text/event-stream" અને X-Accel-Buffering: no સાથે વપરાય છે.
• async જનરેટર: async def અને yield સાથે જાહેર કરેલું કોરુટિન જે મૂલ્યોને આળસુ રીતે (lazily) ઉત્પન્ન કરે છે; આ પાઠમાં તે અપસ્ટ્રીમ LLM ટોકન ચંક્સ પર પુનરાવર્તન કરે છે અને દર ટોકન માટે એક SSE ફ્રેમ yield કરે છે.
• Request.is_disconnected(): FastAPI / Starlette મેથડ જે અંતર્ગત ASGI ટ્રાન્સપોર્ટ જણાવે કે ક્લાયન્ટે કનેક્શન બંધ કરી દીધું છે ત્યારે True પરત કરે છે; જ્યારે વપરાશકર્તા દૂર નેવિગેટ કરે ત્યારે ટોકન જનરેશનને શોર્ટ-સર્કિટ કરવા માટે વપરાય છે.

કન્સેપ્ટ્સ

HTTP પર LLM પ્રતિસાદ સ્ટ્રીમ કરવું એ ચાર વિચારો પર આધારિત છે જે સાથે મળીને કામ કરે છે:

1. એક HTTP પ્રતિસાદ પર ફ્રેમ-બાય-ફ્રેમ ડિલિવરી. સંપૂર્ણ કમ્પ્લીશનને બફર કરવાને બદલે, એન્ડપોઇન્ટ એક જ text/event-stream પ્રતિસાદ ખુલ્લો રાખે છે અને પ્રોવાઇડર પાસેથી મળતા દર ટોકન ચંક માટે એક SSE ફ્રેમ લખે છે. બ્રાઉઝર-બાજુનું EventSource (અથવા fetch() + ReadableStream) દરેક ફ્રેમ આવે તે જ ક્ષણે તેનો ઉપયોગ કરે છે, અને આ જ WebSockets વગર ટાઇપિંગ-શૈલીનો UX શક્ય બનાવે છે.
2. ટાઇપ કરેલી ઇવેન્ટ શબ્દાવલિ. એન્ડપોઇન્ટ બરાબર ત્રણ ઇવેન્ટ નામો ઉત્સર્જિત કરે છે: કન્ટેન્ટ ડેલ્ટા માટે token, સ્ટ્રક્ચર્ડ JSON તરીકે રજૂ થતી અપસ્ટ્રીમ નિષ્ફળતાઓ માટે error, અને finish_reason વહન કરતા ટર્મિનલ સેન્ટિનલ તરીકે done. ક્લાયન્ટ દરેક ફ્રેમને રૂટ કરવા માટે event: નો ઉપયોગ કરે છે; done સેન્ટિનલ વગર ક્લાયન્ટ "પૂર્ણ થયું" અને "અટકી ગયું" વચ્ચે તફાવત કરી શકતો નથી.
3. જનરેટર લાઇફસાઇકલ = સ્ટ્રીમ લાઇફસાઇકલ. StreamingResponse ને આપેલો async જનરેટર જ સ્ટ્રીમ છે. જ્યારે તે return થાય, પ્રતિસાદ બંધ થાય છે; જ્યારે તે raise કરે, પ્રતિસાદ રદ થાય છે. આથી ટર્મિનલ error ફ્રેમ્સ ઉત્સર્જિત કરવા અને પ્રોવાઇડર-બાજુના સંસાધનો (HTTP/gRPC કનેક્શન્સ) મુક્ત કરવા માટે try / except / finally બ્લોક જ એકમાત્ર સાચું સ્થાન બને છે.
4. ડિસ્કનેક્ટ-અવેર કેન્સલેશન. ક્લાયન્ટ્સ સતત કનેક્શન છોડે છે (ટેબ બંધ, રિફ્રેશ, નવો પ્રોમ્પ્ટ). એન્ડપોઇન્ટ yield વચ્ચે request.is_disconnected() ને પોલ કરે છે, અથવા બેકગ્રાઉન્ડ વૉચર કોરુટિન ચલાવે છે જે asyncio.Event સેટ કરે છે, જેથી સોકેટ જતું રહે તે જ ક્ષણે ટોકન જનરેશન બંધ થાય — નહીંતર સર્વર એવા ટોકન્સ માટે ચૂકવણી કરતું રહે છે જે કોઈ વાંચશે નહીં.

કોડ વૉકથ્રુ

W3C Server-Sent Events પ્રોટોકોલ

SSE સ્પેસિફિકેશન (W3C, 2015) એક ટેક્સ્ટ-આધારિત ફ્રેમિંગ પ્રોટોકોલ વ્યાખ્યાયિત કરે છે જે text/event-stream કન્ટેન્ટ ટાઇપ સાથે પ્રતિસાદ બોડી પર પ્રસારિત થાય છે. દરેક ફ્રેમ એક અથવા વધુ ફીલ્ડ લાઇનોથી બનેલી હોય છે જે ખાલી લાઇન (\n\n) થી સમાપ્ત થાય છે. LLM સ્ટ્રીમિંગમાં તમે જે ત્રણ ફીલ્ડ વાપરશો તે છે:

• event: એક વૈકલ્પિક ઇવેન્ટ ટાઇપ સ્ટ્રિંગ. જ્યારે છોડી દેવામાં આવે, ત્યારે બ્રાઉઝરનું EventSource API સામાન્ય message ઇવેન્ટ ફાયર કરે છે. ચેટ સ્ટ્રીમિંગ માટે તમે કન્ટેન્ટ ડેલ્ટા માટે event: token, અપસ્ટ્રીમ નિષ્ફળતાઓ માટે event: error, અને ટર્મિનલ સેન્ટિનલ તરીકે event: done ઉત્સર્જિત કરશો.
• data: પેલોડ લાઇન. એક જ ફ્રેમમાં અનેક data: લાઇનોને ક્લાયન્ટ ન્યૂલાઇન અક્ષરોથી જોડે છે. JSON પેલોડ માટે, સિરિયલાઇઝ કરેલા ઑબ્જેક્ટવાળી એક જ data: લાઇન પ્રમાણભૂત પ્રથા છે.
• id: એક વૈકલ્પિક ઇવેન્ટ ઓળખકર્તા જે last-event-ID પુનઃકનેક્શન સક્ષમ કરે છે. EventSource ક્લાયન્ટ્સ પુનઃકનેક્ટ પર Last-Event-ID મોકલે છે, પરંતુ LLM સ્ટ્રીમિંગ સેશન્સ ફરી શરૂ કરી શકાય તેવા નથી, તેથી તમે આ ફીલ્ડ છોડી દેશો અને તેના બદલે એપ્લિકેશન-સ્તરના રિટ્રાય લોજિક પર આધાર રાખશો.

એક નિર્ણાયક અમલીકરણ વિગત: દરેક ફીલ્ડ લાઇન એક \n થી સમાપ્ત થાય છે, અને ફ્રેમ વધારાના \n થી સમાપ્ત થાય છે, જે ડબલ-ન્યૂલાઇન ડિલિમિટર \n\n બનાવે છે. આ પાછળની ખાલી લાઇન છોડી દેવાથી ક્લાયન્ટ અનિશ્ચિત સમય માટે બફર કરે છે — એક બગ જે ફક્ત લોડ હેઠળ જ સામે આવે છે, કારણ કે લોકલ ડેવલપમેન્ટ દરમિયાન TCP Nagle કોએલેસિંગ ખૂટતા ડિલિમિટરને ઢાંકી દે છે.

• લાઇન 1: આને Mermaid સિક્વન્સ ડાયાગ્રામ તરીકે જાહેર કરે છે, જે સમય દરમિયાન કમ્પોનન્ટ્સ વચ્ચેની આંતરક્રિયાઓ દર્શાવવા માટે વપરાય છે.
• લાઇન 2-4: ડાયાગ્રામમાં ત્રણ પાર્ટિસિપન્ટ્સ (એક્ટર્સ) વ્યાખ્યાયિત કરે છે: Client (Browser/fetch() તરીકે લેબલ કરેલ), FastAPI (FastAPI Endpoint તરીકે લેબલ કરેલ), અને Provider (LLM Provider API તરીકે લેબલ કરેલ).
• લાઇન 6: ક્લાયન્ટને /api/v1/chat/stream પર FastAPI એન્ડપોઇન્ટને Accept: text/event-stream હેડર સાથે POST રિક્વેસ્ટ મોકલતો દર્શાવે છે, જે Server-Sent Events (SSE) કનેક્શન શરૂ કરે છે.
• લાઇન 7: FastAPI રિક્વેસ્ટને LLM Provider API ને stream=True સાથે સ્ટ્રીમિંગ કૉલ તરીકે ફોરવર્ડ કરતું દર્શાવે છે, જે ચંક કરેલા ટોકન-બાય-ટોકન પ્રતિસાદો સક્ષમ કરે છે.
• લાઇન 8-11: પુનરાવર્તિત સ્ટ્રીમિંગ ચક્રનું પ્રતિનિધિત્વ કરતો લૂપ બ્લોક વ્યાખ્યાયિત કરે છે — દરેક ટોકન ચંક માટે, Provider FastAPI ને chunk.delta.content પાછું મોકલે છે (ડૅશવાળું તીર async પ્રતિસાદ દર્શાવે છે), અને FastAPI તેને ક્લાયન્ટને token ટાઇપ અને કન્ટેન્ટ ધરાવતા JSON ડેટા પેલોડ સાથે SSE-ફોર્મેટેડ ઇવેન્ટ તરીકે ફોરવર્ડ કરે છે.
• લાઇન 12: Provider FastAPI ને finish_reason: stop સાથે અંતિમ સંદેશ મોકલતો દર્શાવે છે, જે સંકેત આપે છે કે LLM એ પોતાનું પ્રતિસાદ જનરેશન પૂર્ણ કર્યું છે.
• લાઇન 13: FastAPI પૂર્ણતાના સંકેતને ક્લાયન્ટને done ટાઇપની SSE ઇવેન્ટ તરીકે stop reason ધરાવતા JSON પેલોડ સાથે ફોરવર્ડ કરતું દર્શાવે છે, જે સૂચવે છે કે સ્ટ્રીમ પૂર્ણ થઈ ગઈ છે.
• લાઇન 14: ક્લાયન્ટને પોતાને સંદેશ મોકલતો (સેલ્ફ-કૉલ) દર્શાવે છે, જે EventSource કનેક્શન બંધ કરવા અથવા SSE સ્ટ્રીમ સમાપ્ત કરવા માટે AbortController ટ્રિગર કરવાના ક્લાયન્ટ-બાજુના ક્લીનઅપનું પ્રતિનિધિત્વ કરે છે.

આ ડાયાગ્રામ સંપૂર્ણ લાઇફસાઇકલને આવરી લે છે. ક્લાયન્ટ POST રિક્વેસ્ટ શરૂ કરે છે (નોંધ: નેટિવ EventSource API ફક્ત GET ને સપોર્ટ કરે છે, તેથી પ્રોડક્શન ચેટ UI fetch() ને ReadableStream રીડર સાથે અથવા @microsoft/fetch-event-source જેવા પોલીફિલનો ઉપયોગ કરે છે). FastAPI કનેક્શન ખુલ્લું રાખે છે, અપસ્ટ્રીમ પ્રોવાઇડર ચંક્સ પહોંચાડે તેમ SSE ફ્રેમ્સ yield કરે છે, અને સ્ટ્રીમ પૂર્ણતાનો સંકેત આપવા માટે ટર્મિનલ done ઇવેન્ટ ઉત્સર્જિત કરે છે.

Pydantic મોડેલ્સ અને સ્ટ્રીમિંગ એન્ડપોઇન્ટ

async જનરેટરને વાયર કરતાં પહેલાં, તમને રિક્વેસ્ટ વેલિડેશન અને SSE ફ્રેમ ફોર્મેટરની જરૂર છે. નીચેનો કોડ ChatMessage અને ChatRequest Pydantic મોડેલ્સ વ્યાખ્યાયિત કરે છે જે આવતા પેલોડને માન્ય કરે છે, એક હેલ્પર ફંક્શન format_sse જે ઇવેન્ટ ટાઇપ અને ડેટા ડિક્શનરી આર્ગ્યુમેન્ટ્સમાંથી W3C-સુસંગત SSE ફ્રેમ્સ બનાવે છે, અને stream_chat FastAPI રૂટ હેન્ડલર જે text/event-stream કન્ટેન્ટ ટાઇપ સાથે StreamingResponse પરત કરે છે. stream_chat ફંક્શન async જનરેટર _sse_generator ને કામ સોંપે છે, જ્યાં વાસ્તવિક ટોકન પુનરાવર્તન અને ક્લાયન્ટ ડિસ્કનેક્ટ શોધ થાય છે. Cache-Control અને X-Accel-Buffering હેડર્સ પર વિશેષ ધ્યાન આપો — Nginx જેવી રિવર્સ પ્રોક્સીઓ ક્લાયન્ટને ફોરવર્ડ કરતાં પહેલાં સંપૂર્ણ સ્ટ્રીમને બફર ન કરે તે માટે આ આવશ્યક છે.

Code snippetpython
1import json
2import asyncio
3from typing import AsyncGenerator
4from pydantic import BaseModel, Field
5from fastapi import FastAPI, Request
6from fastapi.responses import StreamingResponse
7
8app = FastAPI()
9
10class ChatMessage(BaseModel):
11    role: str = Field(..., pattern="^(system|user|assistant)$")
12    content: str = Field(..., min_length=1, max_length=32_000)
13
14class ChatRequest(BaseModel):
15    messages: list[ChatMessage]
16    provider: str = Field(..., pattern="^(openai|gemini|anthropic|together)$")
17    model: str
18    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
19
20def format_sse(event: str, data: dict) -> str:
21    payload = json.dumps(data, ensure_ascii=False)
22    return f"event: {event}\ndata: {payload}\n\n"
23
24@app.post("/api/v1/chat/stream")
25async def stream_chat(body: ChatRequest, request: Request):
26    async def _sse_generator() -> AsyncGenerator[str, None]:
27        try:
28            async for token in dispatch_provider(body):
29                if await request.is_disconnected():
30                    break
31                yield format_sse("token", {"content": token})
32            yield format_sse("done", {"finish_reason": "stop"})
33        except Exception as exc:
34            yield format_sse("error", {"message": str(exc)})
35
36    return StreamingResponse(
37        _sse_generator(),
38        media_type="text/event-stream",
39        headers={
40            "Cache-Control": "no-cache",
41            "X-Accel-Buffering": "no",
42            "Connection": "keep-alive",
43        },
44    )

• લાઇન 1-5: જરૂરી મોડ્યુલ્સ import કરે છે. asyncio import ડિસ્કનેક્ટ હેન્ડલિંગમાં વપરાતી async sleep અને કેન્સલેશન પેટર્નને સપોર્ટ કરે છે. typing માંથી AsyncGenerator SSE જનરેટર ફંક્શન માટે return ટાઇપ એનોટેશન પૂરું પાડે છે.
• લાઇન 7: FastAPI એપ્લિકેશનનું ઇન્સ્ટન્સ બનાવે છે. પ્રોડક્શનમાં, આ ઇન્સ્ટન્સ મલ્ટી-પ્રોસેસ સર્વિંગ માટે --workers સાથે Uvicorn દ્વારા લોડ થયેલા મોડ્યુલમાં રહે છે.
• લાઇન 9-11: ChatMessage મોડેલ વ્યાખ્યાયિત કરે છે. role ફીલ્ડ મૂલ્યોને ત્રણ પ્રમાણભૂત ચેટ રોલ્સ સુધી મર્યાદિત કરવા માટે regex પેટર્ન કન્સ્ટ્રેઇન્ટનો ઉપયોગ કરે છે. content ફીલ્ડ ખાલી સંદેશાઓને નકારવા માટે લઘુત્તમ લંબાઈ 1 લાગુ કરે છે અને પેલોડના દુરુપયોગને રોકવા માટે 32,000 અક્ષરોની મર્યાદા રાખે છે.
• લાઇન 13-17: ChatRequest મોડેલ વ્યાખ્યાયિત કરે છે. provider ફીલ્ડ ચાર સપોર્ટેડ બેકએન્ડ્સની ગણતરી કરે છે. temperature ફીલ્ડ ડિફૉલ્ટ 0.7 રાખે છે અને રેન્જને 0.0 અને 2.0 વચ્ચે મર્યાદિત કરે છે, જે ચારેય પ્રોવાઇડર્સની માન્ય રેન્જના યુનિયન સાથે મેળ ખાય છે.
• લાઇન 19-21: format_sse હેલ્પર Python ડિક્શનરીને W3C-સુસંગત SSE ફ્રેમમાં સિરિયલાઇઝ કરે છે. ensure_ascii=False ફ્લેગ બહુભાષી ચેટ પ્રતિસાદોમાં Unicode અક્ષરોને \uXXXX સિક્વન્સમાં એસ્કેપ કર્યા વગર જાળવી રાખે છે, જે CJK કન્ટેન્ટ માટે ફ્રેમનું કદ 5x સુધી ઘટાડે છે.
• લાઇન 23-24: રૂટ ડેકોરેટર POST એન્ડપોઇન્ટ રજિસ્ટર કરે છે. POST જરૂરી છે કારણ કે ચેટ રિક્વેસ્ટ્સ મેસેજ હિસ્ટ્રી બોડી વહન કરે છે જે GET રિક્વેસ્ટ માટેની સુરક્ષિત URL લંબાઈ મર્યાદા ઓળંગી જાય છે.
• લાઇન 25-33: આંતરિક _sse_generator async જનરેટર સ્ટ્રીમિંગ પાઇપલાઇનનો મુખ્ય ભાગ છે. તે dispatch_provider (એક રાઉટર ફંક્શન જે તમે પછીના વિભાગોમાં દરેક LLM પ્રોવાઇડર માટે બનાવશો) દ્વારા yield થયેલા ટોકન્સ પર પુનરાવર્તન કરે છે. દરેક પુનરાવર્તન પર, તે ક્લાયન્ટ અબોર્ટ શોધવા માટે request.is_disconnected() તપાસે છે. જો ક્લાયન્ટે કનેક્શન બંધ કરી દીધું હોય, તો જનરેટર લૂપમાંથી બહાર નીકળે છે, જે છોડી દેવાયેલી રિક્વેસ્ટ્સ પર ઇન્ફરન્સ ટોકન્સનો વ્યય અટકાવે છે. try/except બ્લોક અપસ્ટ્રીમ પ્રોવાઇડર ભૂલોને પકડે છે અને તેમને event: error SSE ફ્રેમ્સ તરીકે ઉત્સર્જિત કરે છે જેથી ક્લાયન્ટને તૂટેલા કનેક્શનને બદલે સ્ટ્રક્ચર્ડ ભૂલ માહિતી મળે.
• લાઇન 35-42: StreamingResponse async જનરેટરને રૅપ કરે છે. media_type પેરામીટર Content-Type: text/event-stream હેડર સેટ કરે છે. ત્રણ વધારાના હેડર્સ નિર્ણાયક છે: Cache-Control: no-cache CDN અને બ્રાઉઝર કેશને સ્ટ્રીમ બફર કરતા રોકે છે, X-Accel-Buffering: no Nginx ને પ્રોક્સી બફરિંગ અક્ષમ કરવા સૂચના આપે છે (આ હેડર વગર, Nginx ડિફૉલ્ટ રૂપે સંપૂર્ણ પ્રતિસાદ બફર કરે છે, જે સ્ટ્રીમિંગના હેતુને નિષ્ફળ બનાવે છે), અને Connection: keep-alive મધ્યસ્થીઓને સંકેત આપે છે કે કનેક્શન ટકી રહેવું જોઈએ.

ક્લાયન્ટ ડિસ્કનેક્ટ શોધ અને જનરેટર ક્લીનઅપ

પ્રોડક્શન સ્ટ્રીમિંગમાં ક્લાયન્ટ ડિસ્કનેક્ટ સૌથી સામાન્ય નિષ્ફળતા મોડ છે. વપરાશકર્તા દૂર નેવિગેટ કરે છે, ટેબ બંધ કરે છે, અથવા અગાઉની રિક્વેસ્ટ પૂર્ણ થાય તે પહેલાં નવી રિક્વેસ્ટ ટ્રિગર કરે છે. સ્પષ્ટ હેન્ડલિંગ વગર, સર્વર LLM પ્રોવાઇડર પાસેથી ટોકન્સનો ઉપયોગ કરતું રહે છે — ખર્ચ બાળે છે અને કનેક્શન સ્લોટ રોકી રાખે છે. FastAPI ની Request.is_disconnected મેથડ અંતર્ગત ASGI ટ્રાન્સપોર્ટ પર નોન-બ્લોકિંગ તપાસ કરે છે. જોકે, આ મેથડની એક સૂક્ષ્મ મર્યાદા છે: તે ફક્ત ત્યારે જ ડિસ્કનેક્ટ શોધે છે જ્યારે ઇવેન્ટ લૂપ નિયંત્રણ yield કરે. જો તમારો async જનરેટર yield વચ્ચે CPU-બાઉન્ડ સિરિયલાઇઝેશન સ્ટેપ કરે, તો ડિસ્કનેક્ટ તપાસ એક અથવા વધુ ટોકન્સથી પાછળ રહી શકે છે.

નીચેનો કોડ પ્રોડક્શન-હાર્ડન્ડ ડિસ્કનેક્ટ શોધ પેટર્ન દર્શાવે છે જે is_disconnected પોલિંગ અભિગમને જનરેટર ક્લીનઅપ માટેના asyncio.shield ગાર્ડ સાથે જોડે છે. guarded_sse_stream ફંક્શન કાચા પ્રોવાઇડર ટોકન ઇટરેટરને રૅપ કરે છે અને ખાતરી કરે છે કે ક્લાયન્ટ સ્ટ્રીમની મધ્યમાં ડિસ્કનેક્ટ થાય તો પણ, ખુલ્લા HTTP કનેક્શન્સ અથવા gRPC સ્ટ્રીમ્સ જેવા પ્રોવાઇડર-બાજુના સંસાધનો મુક્ત કરવા માટે અંતિમ ક્લીનઅપ કોરુટિન ચાલે. _check_disconnect કોરુટિન બેકગ્રાઉન્ડ ટાસ્ક તરીકે ચાલે છે જે ક્લાયન્ટ છૂટી જાય ત્યારે cancel_event સેટ કરે છે, જેથી જનરેટર આગામી પ્રોવાઇડર ચંક આવવાની રાહ જોયા વગર તરત બહાર નીકળી શકે.

Code snippetpython
1async def guarded_sse_stream(
2    body: ChatRequest, request: Request
3) -> AsyncGenerator[str, None]:
4    cancel_event = asyncio.Event()
5
6    async def _watch_disconnect():
7        while not cancel_event.is_set():
8            if await request.is_disconnected():
9                cancel_event.set()
10                return
11            await asyncio.sleep(0.25)
12
13    watcher = asyncio.create_task(_watch_disconnect())
14    try:
15        async for token in dispatch_provider(body):
16            if cancel_event.is_set():
17                break
18            yield format_sse("token", {"content": token})
19        if not cancel_event.is_set():
20            yield format_sse("done", {"finish_reason": "stop"})
21    except asyncio.CancelledError:
22        yield format_sse("error", {"message": "stream_cancelled"})
23    finally:
24        cancel_event.set()
25        watcher.cancel()
26        try:
27            await watcher
28        except asyncio.CancelledError:
29            pass

• લાઇન 1-3: ફંક્શન સિગ્નેચર સ્ટ્રિંગ્સ પરત કરતો async જનરેટર જાહેર કરે છે. તે માન્ય કરેલી ChatRequest બોડી અને ડિસ્કનેક્ટ નિરીક્ષણ માટે કાચો Request ઑબ્જેક્ટ સ્વીકારે છે.
• લાઇન 4: asyncio.Event ઇન્સ્ટન્સ ડિસ્કનેક્ટ વૉચર અને મુખ્ય જનરેટર લૂપ વચ્ચે શેર કરેલા થ્રેડ-સેફ ફ્લેગ તરીકે કામ કરે છે. બુલિયનને બદલે ઇવેન્ટનો ઉપયોગ કરવાથી બે કોરુટિન વચ્ચેની રેસ કન્ડિશન ટળે છે.
• લાઇન 6-11: _watch_disconnect કોરુટિન દર 250 મિલિસેકન્ડે request.is_disconnected() પોલ કરે છે. 0.25-સેકન્ડનો અંતરાલ પ્રતિભાવક્ષમતા અને CPU ઓવરહેડ વચ્ચે સંતુલન રાખે છે — 100ms થી વધુ વારંવાર પોલિંગ કોઈ વ્યવહારુ લાભ આપતું નથી કારણ કે લોડ બેલેન્સર્સ દ્વારા TCP FIN પ્રસરણ સામાન્ય રીતે 50-200ms લે છે. જ્યારે ડિસ્કનેક્ટ શોધાય છે, ત્યારે ઇવેન્ટ તરત સેટ થાય છે, જે જનરેટરને yield કરવાનું બંધ કરવા સંકેત આપે છે.
• લાઇન 13: વૉચર કોરુટિન બેકગ્રાઉન્ડ asyncio.Task તરીકે લૉન્ચ થાય છે. આ ખાતરી કરે છે કે તે જનરેટરના ટોકન પુનરાવર્તન લૂપને બ્લોક કર્યા વગર તેની સાથે સમાંતર ચાલે.
• લાઇન 14-20: મુખ્ય જનરેશન લૂપ દરેક ફ્રેમ yield કરતાં પહેલાં cancel_event.is_set() તપાસે છે. આ તપાસ લગભગ તાત્કાલિક છે (તે આંતરિક બુલિયન વાંચે છે) અને વૉચર ડિસ્કનેક્ટ શોધે પછી સબ-મિલિસેકન્ડ અબોર્ટ લેટન્સી પૂરી પાડે છે. done ઇવેન્ટ ફક્ત ત્યારે જ ઉત્સર્જિત થાય છે જ્યારે સ્ટ્રીમ કેન્સલેશન વગર કુદરતી રીતે પૂર્ણ થઈ હોય.
• લાઇન 21-22: asyncio.CancelledError હેન્ડલર એવા કિસ્સાને પકડે છે જ્યાં Uvicorn નું ASGI સર્વર પ્રતિસાદ કોરુટિનને સીધું કેન્સલ કરે છે (આ ત્યારે થાય છે જ્યારે સક્રિય સ્ટ્રીમ દરમિયાન સર્વર શટ ડાઉન થાય). ભૂલ ફ્રેમ ક્લાયન્ટને કાચા કનેક્શન ડ્રોપને બદલે સ્ટ્રક્ચર્ડ કેન્સલેશન સંકેત પૂરો પાડે છે.
• લાઇન 23-29: finally બ્લોક જનરેટર કેવી રીતે બહાર નીકળે તેને ધ્યાનમાં લીધા વગર ક્લીનઅપની ખાતરી આપે છે. તે કેન્સલ ઇવેન્ટ સેટ કરે છે (પહેલેથી સેટ હોય તો idempotent), વૉચર ટાસ્ક કેન્સલ કરે છે, અને તેની પૂર્ણતાની રાહ જુએ છે. await watcher ની આસપાસનો આંતરિક try/except એ CancelledError ને શાંત કરે છે જે ટાસ્ક તેના આગામી await બિંદુ પહેલાં કેન્સલ થાય ત્યારે પ્રસરે છે.

શું કરવું અને શું ન કરવું

શું કરવું

1. દરેક ફ્રેમ બનાવવા માટે format_sse (અથવા સમકક્ષ હેલ્પર) નો ઉપયોગ કરો — W3C પ્રોટોકોલ માટે દરેક ફ્રેમ ખાલી લાઇન (\n\n) થી સમાપ્ત થવી જરૂરી છે, અને તે ડિલિમિટર છોડી દેવાથી ક્લાયન્ટ અનિશ્ચિત સમય માટે બફર કરે છે; આ બગ લોકલ ડેવલપમેન્ટમાં અદૃશ્ય હોય છે કારણ કે TCP Nagle કોએલેસિંગ તેને ઢાંકી દે છે, જે તેને ફક્ત લોડ હેઠળ જ થતી અને પુનઃઉત્પન્ન કરવી મુશ્કેલ નિષ્ફળતા બનાવે છે.
2. StreamingResponse હેડર્સમાં Cache-Control: no-cache અને X-Accel-Buffering: no સેટ કરો — બંને હેડર્સ વગર, Nginx અને અન્ય રિવર્સ પ્રોક્સીઓ ફોરવર્ડ કરતાં પહેલાં સંપૂર્ણ પ્રતિસાદ બોડી બફર કરે છે, જે પ્રતિ-ટોકન _sse_generator આઉટપુટને એક જ ચંકમાં સંકોચી દે છે અને SSE નો લેટન્સી લાભ સંપૂર્ણપણે નિષ્ફળ બનાવે છે.
3. _sse_generator ની અંદર દરેક પુનરાવર્તન પર await request.is_disconnected() કૉલ કરો — બંધ થયેલા બ્રાઉઝર ટેબ અથવા સ્ટ્રીમની મધ્યમાં AbortController કેન્સલ શોધવા માટે FastAPI આ એકમાત્ર મિકેનિઝમ પૂરું પાડે છે; પોઝિટિવ તપાસ પર જનરેટરમાંથી બહાર નીકળવાથી અપસ્ટ્રીમ dispatch_provider કૉલ અટકે છે અને કોઈ ક્લાયન્ટ કદી વાંચશે નહીં તેવા ટોકન્સ માટેનું બિલિંગ ટળે છે.

શું ન કરવું

1. નેટિવ બ્રાઉઝર EventSource API નો ઉપયોગ કરીને આ એન્ડપોઇન્ટ સાથે કનેક્ટ ન કરો — EventSource ફક્ત GET માટે છે અને ChatRequest POST બોડી વહન કરી શકતું નથી; fetch() ને ReadableStream રીડર સાથે અથવા @microsoft/fetch-event-source પોલીફિલનો ઉપયોગ કરો, જે બંને SSE પ્રતિસાદ પર POST સેમેન્ટિક્સને સપોર્ટ કરે છે.
2. format_sse દ્વારા રૂટ કરવાને બદલે SSE ફ્રેમ્સને ઇનલાઇન f-strings તરીકે હાથે ન લખો — event:, data:, અને \n સ્ટ્રિંગ્સને હાથે જોડતી વખતે ડબલ-ન્યૂલાઇન ટર્મિનેટર છૂટી જવાની સૌથી વધુ શક્યતા ધરાવતું તત્ત્વ છે, અને પરિણામી શાંત બફરિંગ નિષ્ફળતા ફક્ત પ્રોડક્શન લોડ હેઠળ જ દેખાય છે જ્યારે Nagle કોએલેસિંગ તેને ઢાંકવાનું બંધ કરે છે.
3. ટોકન્સને લિસ્ટમાં બફર કરીને StreamingResponse ને બદલે JSONResponse પરત ન કરો — સંચય અનેક સેકન્ડના ટાઇમ-ટુ-ફર્સ્ટ-ટોકન ફ્રીઝને ફરી લાવે છે જેને દૂર કરવા માટે જ AsyncGenerator[str, None] સાથે જોડાયેલું StreamingResponse અસ્તિત્વમાં છે, અને તે is_disconnected() તપાસ બિંદુને દૂર કરે છે જે બેકાબૂ અપસ્ટ્રીમ ટોકન વપરાશને રોકે છે.