FastAPI SSE streaming response endpoint ஒன்றை உருவாக்குங்கள்

அறிமுகம்

முழு LLM பதிலும் வரும் வரை காத்திருந்து பின்னர் ரெண்டர் செய்யும் அரட்டை UI ஐ நீங்கள் வெளியிடும்போது, பயனர்கள் பல வினாடிகள் நீடிக்கும் உறைதலை உணர்ந்து அமர்வைக் கைவிடுகிறார்கள் — மொத்த தாமதம் ஸ்ட்ரீம் செய்யப்பட்ட பதிலுக்குச் சமமாக இருந்தாலும் கூட. Server-Sent Events (SSE) உங்கள் FastAPI பின்தளத்திற்கு, மேல்நிலை மாடலிலிருந்து ஒவ்வொரு டோக்கனும் வந்த உடனேயே அதை உலாவிக்கு அனுப்ப அனுமதிக்கிறது; இது உணரப்படும் உறைதலை நீக்குவதோடு, கிளையன்ட் துண்டிப்புகளைக் கண்டறிவதற்கான ஒரு சுத்தமான புள்ளியையும் தருகிறது, இதனால் யாரும் படிக்காத டோக்கன்களுக்கு நீங்கள் பணம் செலுத்துவதை நிறுத்தலாம். இந்தப் பாடத்தின் முடிவில், LLM டோக்கன்களைப் படிப்படியாக வழங்கும், தலைகீழ் ப்ராக்ஸிகள் பைட்டுகளை உடனடியாக முன்னனுப்பத் தேவையான ஹெடர்களை அமைக்கும், மற்றும் கிளையன்ட் இணைப்பை மூடும்போது சுத்தமாக வெளியேறும் W3C-இணக்கமான SSE ஸ்ட்ரீமிங் எண்ட்பாயிண்ட்டை HTTP வழியாக நீங்கள் செயல்படுத்த முடியும்.

முக்கிய சொற்கள்

• Server-Sent Events (SSE): Content-Type: text/event-stream உடன் ஒரே நீண்ட-ஆயுள் HTTP/1.1 பதிலின் மூலம் கொண்டு செல்லப்படும் ஒரு W3C ஸ்ட்ரீமிங் நெறிமுறை; இதில் சர்வர், இரு தரப்பில் ஏதேனும் ஒன்று இணைப்பை மூடும் வரை, புதிய-வரி-பிரிக்கப்பட்ட event: / data: பிரேம்களை கிளையன்ட்டுக்கு அனுப்புகிறது.
• SSE frame: ஸ்ட்ரீமின் ஒரு அலகு; ஒன்று அல்லது அதற்கு மேற்பட்ட புல வரிகளால் (எ.கா. event: token, data: {...}) உருவாக்கப்பட்டு, ஒரு வெற்று வரியால் (\n\n) முடிக்கப்படுகிறது; இறுதி வெற்று வரியை விட்டுவிட்டால் கிளையன்ட்கள் முடிவில்லாமல் பஃபர் செய்யும்.
• StreamingResponse: ஒரு async generator ஐ நுகர்ந்து, yield செய்யப்படும் ஒவ்வொரு துண்டையும் நேரடியாக சாக்கெட்டுக்கு எழுதும் FastAPI இன் பதில் class; ப்ராக்ஸி பஃபரிங்கைத் தோற்கடிக்க இங்கே media_type="text/event-stream" மற்றும் X-Accel-Buffering: no உடன் சேர்த்துப் பயன்படுத்தப்படுகிறது.
• async generator: async def மற்றும் yield உடன் அறிவிக்கப்பட்டு, மதிப்புகளை சோம்பலாக (lazily) உருவாக்கும் ஒரு coroutine; இந்தப் பாடத்தில் இது மேல்நிலை LLM டோக்கன் துண்டுகளை மீள்செய்து, ஒவ்வொரு டோக்கனுக்கும் ஒரு SSE பிரேமை yield செய்கிறது.
• Request.is_disconnected(): அடிப்படை ASGI போக்குவரத்து, கிளையன்ட் இணைப்பை மூடிவிட்டதாகத் தெரிவித்தவுடன் True ஐத் திருப்பும் FastAPI / Starlette முறை; பயனர் வேறு பக்கத்திற்குச் செல்லும்போது டோக்கன் உருவாக்கத்தைக் குறுக்குவழியில் நிறுத்தப் பயன்படுத்தப்படுகிறது.

கருத்துகள்

HTTP வழியாக ஒரு LLM பதிலை ஸ்ட்ரீம் செய்வது, ஒன்றாகச் செயல்படும் நான்கு யோசனைகளில் அடங்கியுள்ளது:

1. ஒரே HTTP பதிலின் மூலம் பிரேம்-வாரியான விநியோகம். முழு நிறைவையும் பஃபர் செய்வதற்குப் பதிலாக, எண்ட்பாயிண்ட் ஒரே text/event-stream பதிலைத் திறந்து வைத்து, வழங்குநரிடமிருந்து பெறும் ஒவ்வொரு டோக்கன் துண்டுக்கும் ஒரு SSE பிரேமை எழுதுகிறது. உலாவிப் பக்கத்தின் EventSource (அல்லது fetch() + ReadableStream) ஒவ்வொரு பிரேமையும் வந்த உடனேயே நுகர்கிறது; WebSockets இல்லாமல் தட்டச்சு-பாணி UX ஐ சாத்தியமாக்குவது இதுவே.
2. வகைப்படுத்தப்பட்ட நிகழ்வு சொற்களஞ்சியம். எண்ட்பாயிண்ட் சரியாக மூன்று நிகழ்வுப் பெயர்களை வெளியிடுகிறது: உள்ளடக்க டெல்டாக்களுக்கு token, கட்டமைக்கப்பட்ட JSON ஆக வெளிப்படுத்தப்படும் மேல்நிலைத் தோல்விகளுக்கு error, மற்றும் finish_reason ஐக் கொண்டு செல்லும் இறுதி சென்டினலாக done. கிளையன்ட் ஒவ்வொரு பிரேமையும் வழிநடத்த event: ஐப் பயன்படுத்துகிறது; done சென்டினல் இல்லாவிட்டால், "முடிந்தது" மற்றும் "நின்றுவிட்டது" என்பதை கிளையன்ட்டால் வேறுபடுத்த முடியாது.
3. Generator வாழ்க்கைச் சுழற்சி = ஸ்ட்ரீம் வாழ்க்கைச் சுழற்சி. StreamingResponse க்கு அனுப்பப்படும் async generator தான் ஸ்ட்ரீம். அது return செய்யும்போது பதில் மூடப்படுகிறது; அது raise செய்யும்போது பதில் ரத்தாகிறது. இதனால் இறுதி error பிரேம்களை வெளியிடவும், வழங்குநர்-பக்க வளங்களை (HTTP/gRPC இணைப்புகள்) விடுவிக்கவும் try / except / finally தொகுதியே ஒரே சரியான இடமாகிறது.
4. துண்டிப்பு-விழிப்புடன் கூடிய ரத்து. கிளையன்ட்கள் தொடர்ந்து இணைப்புகளைத் துண்டிக்கிறார்கள் (தாவல் மூடுதல், புதுப்பித்தல், புதிய ப்ராம்ப்ட்). எண்ட்பாயிண்ட் yield களுக்கு இடையே request.is_disconnected() ஐ வாக்கெடுப்பு செய்கிறது, அல்லது ஒரு asyncio.Event ஐ அமைக்கும் பின்னணி கண்காணிப்பு coroutine ஐ இயக்குகிறது; இதனால் சாக்கெட் மறைந்த தருணமே டோக்கன் உருவாக்கம் நின்றுவிடுகிறது — இல்லையெனில் யாரும் படிக்கப்போகாத டோக்கன்களுக்கு சர்வர் தொடர்ந்து பணம் செலுத்தும்.

குறியீடு விளக்கம்

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: கிளையன்ட் Accept: text/event-stream ஹெடருடன் /api/v1/chat/stream இல் உள்ள FastAPI எண்ட்பாயிண்ட்டுக்கு POST கோரிக்கையை அனுப்பி, Server-Sent Events (SSE) இணைப்பைத் தொடங்குவதைக் காட்டுகிறது.
• வரி 7: FastAPI கோரிக்கையை LLM Provider API க்கு stream=True உடன் ஸ்ட்ரீமிங் அழைப்பாக முன்னனுப்பி, துண்டு-வாரியான டோக்கன்-வாரியான பதில்களை செயல்படுத்துவதைக் காட்டுகிறது.
• வரிகள் 8-11: மீண்டும் மீண்டும் நடைபெறும் ஸ்ட்ரீமிங் சுழற்சியைக் குறிக்கும் ஒரு loop தொகுதியை வரையறுக்கின்றன — ஒவ்வொரு டோக்கன் துண்டுக்கும், Provider ஒரு chunk.delta.content ஐ FastAPI க்கு திருப்பி அனுப்புகிறது (async பதிலைக் குறிக்கும் கோடிட்ட அம்பு), மேலும் FastAPI அதை token வகை மற்றும் உள்ளடக்கத்தைக் கொண்ட JSON data பேலோடுடன் கூடிய SSE-வடிவமைக்கப்பட்ட நிகழ்வாக கிளையன்ட்டுக்கு முன்னனுப்புகிறது.
• வரி 12: Provider finish_reason: stop உடன் இறுதிச் செய்தியை FastAPI க்கு அனுப்பி, LLM தனது பதில் உருவாக்கத்தை முடித்துவிட்டதைச் சமிக்ஞை செய்வதைக் காட்டுகிறது.
• வரி 13: FastAPI நிறைவு சமிக்ஞையை, நிறுத்தக் காரணத்தைக் கொண்ட JSON பேலோடுடன் done வகை SSE நிகழ்வாக கிளையன்ட்டுக்கு முன்னனுப்பி, ஸ்ட்ரீம் முடிந்துவிட்டதைக் குறிப்பதைக் காட்டுகிறது.
• வரி 14: கிளையன்ட் தனக்குத் தானே ஒரு செய்தியை அனுப்புவதைக் (self-call) காட்டுகிறது; இது EventSource இணைப்பை மூடுதல் அல்லது SSE ஸ்ட்ரீமை முடிக்க AbortController ஐத் தூண்டுதல் போன்ற கிளையன்ட்-பக்க சுத்தம் செய்தலைக் குறிக்கிறது.

இந்த வரைபடம் முழு வாழ்க்கைச் சுழற்சியையும் படம்பிடிக்கிறது. கிளையன்ட் ஒரு POST கோரிக்கையைத் தொடங்குகிறது (குறிப்பு: நேட்டிவ் EventSource API GET ஐ மட்டுமே ஆதரிக்கிறது, எனவே தயாரிப்பு அரட்டை UI கள் ReadableStream ரீடருடன் fetch() ஐ அல்லது @microsoft/fetch-event-source போன்ற ஒரு polyfill ஐப் பயன்படுத்துகின்றன). FastAPI இணைப்பைத் திறந்து வைத்து, மேல்நிலை வழங்குநர் துண்டுகளை வழங்கும்போது SSE பிரேம்களை yield செய்து, ஸ்ட்ரீம் நிறைவைச் சமிக்ஞை செய்ய இறுதி done நிகழ்வை வெளியிடுகிறது.

Pydantic மாடல்கள் மற்றும் ஸ்ட்ரீமிங் எண்ட்பாயிண்ட்

async generator ஐ இணைப்பதற்கு முன், உங்களுக்கு கோரிக்கை சரிபார்ப்பும் SSE பிரேம் வடிவமைப்பானும் தேவை. பின்வரும் குறியீடு உள்வரும் பேலோடுகளை சரிபார்க்கும் ChatMessage மற்றும் ChatRequest Pydantic மாடல்களையும், நிகழ்வு வகை மற்றும் தரவு அகராதி வாதங்களிலிருந்து W3C-இணக்கமான SSE பிரேம்களை உருவாக்கும் format_sse உதவி செயல்பாட்டையும், text/event-stream உள்ளடக்க வகையுடன் StreamingResponse ஐத் திருப்பும் stream_chat FastAPI route handler ஐயும் வரையறுக்கிறது. stream_chat செயல்பாடு async generator ஆன _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: தேவையான தொகுதிகளை இறக்குமதி செய்கின்றன. asyncio import துண்டிப்புக் கையாளுதலில் பயன்படுத்தப்படும் async sleep மற்றும் ரத்து முறைகளை ஆதரிக்கிறது. typing இலிருந்து AsyncGenerator SSE generator செயல்பாட்டிற்கான return வகை குறிப்பை வழங்குகிறது.
• வரி 7: FastAPI பயன்பாட்டை உருவாக்குகிறது. தயாரிப்பில், இந்த நிகழ்வு பல-செயல்முறை சேவைக்காக --workers உடன் Uvicorn ஆல் ஏற்றப்படும் ஒரு தொகுதியில் இருக்கும்.
• வரிகள் 9-11: ChatMessage மாடலை வரையறுக்கின்றன. role புலம் மதிப்புகளை மூன்று நிலையான அரட்டைப் பாத்திரங்களுக்கு மட்டுப்படுத்த regex pattern கட்டுப்பாட்டைப் பயன்படுத்துகிறது. 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 உள்ளடக்கத்திற்கு பிரேம் அளவை 5 மடங்கு வரை குறைக்கிறது.
• வரிகள் 23-24: route decorator ஒரு POST எண்ட்பாயிண்ட்டைப் பதிவு செய்கிறது. அரட்டைக் கோரிக்கைகள் GET கோரிக்கைகளுக்கான பாதுகாப்பான URL நீள வரம்புகளைத் தாண்டும் செய்தி வரலாறு உடலைக் கொண்டு செல்வதால் POST அவசியம்.
• வரிகள் 25-33: உள்ளமைந்த _sse_generator async generator ஸ்ட்ரீமிங் பைப்லைனின் மையமாகும். இது dispatch_provider (ஒவ்வொரு LLM வழங்குநருக்கும் பிந்தைய பிரிவுகளில் நீங்கள் உருவாக்கப்போகும் ஒரு ரூட்டர் செயல்பாடு) yield செய்யும் டோக்கன்களை மீள்செய்கிறது. ஒவ்வொரு மீள்செயலிலும், கிளையன்ட் ரத்தைக் கண்டறிய request.is_disconnected() ஐச் சரிபார்க்கிறது. கிளையன்ட் இணைப்பை மூடியிருந்தால், generator சுழற்சியிலிருந்து வெளியேறி, கைவிடப்பட்ட கோரிக்கைகளில் அனுமான டோக்கன்கள் வீணாவதைத் தடுக்கிறது. try/except தொகுதி மேல்நிலை வழங்குநர் பிழைகளைப் பிடித்து அவற்றை event: error SSE பிரேம்களாக வெளியிடுகிறது; இதனால் கிளையன்ட் துண்டிக்கப்பட்ட இணைப்புக்குப் பதிலாக கட்டமைக்கப்பட்ட பிழைத் தகவலைப் பெறுகிறது.
• வரிகள் 35-42: StreamingResponse async generator ஐ மூடுகிறது. media_type அளவுரு Content-Type: text/event-stream ஹெடரை அமைக்கிறது. மேலும் மூன்று ஹெடர்கள் முக்கியமானவை: Cache-Control: no-cache CDN கள் மற்றும் உலாவி கேச்கள் ஸ்ட்ரீமை பஃபர் செய்வதைத் தடுக்கிறது, X-Accel-Buffering: no Nginx க்கு ப்ராக்ஸி பஃபரிங்கை முடக்க அறிவுறுத்துகிறது (இந்த ஹெடர் இல்லாவிட்டால், Nginx இயல்பாக முழு பதிலையும் பஃபர் செய்து, ஸ்ட்ரீமிங்கின் நோக்கத்தையே தோற்கடிக்கிறது), மற்றும் Connection: keep-alive இடைநிலைகளுக்கு இணைப்பு நீடிக்க வேண்டும் எனச் சமிக்ஞை செய்கிறது.

கிளையன்ட் துண்டிப்பு கண்டறிதல் மற்றும் Generator சுத்தம் செய்தல்

தயாரிப்பு ஸ்ட்ரீமிங்கில் கிளையன்ட் துண்டிப்புகளே மிகவும் பொதுவான தோல்வி முறை. பயனர் வேறு பக்கத்திற்குச் செல்கிறார், தாவலை மூடுகிறார், அல்லது முந்தையது முடிவதற்கு முன் புதிய கோரிக்கையைத் தூண்டுகிறார். வெளிப்படையான கையாளுதல் இல்லாமல், சர்வர் LLM வழங்குநரிடமிருந்து தொடர்ந்து டோக்கன்களை நுகர்கிறது — செலவை எரித்து, இணைப்பு இடத்தைப் பிடித்து வைக்கிறது. FastAPI இன் Request.is_disconnected முறை அடிப்படை ASGI போக்குவரத்தில் தடுக்காத (non-blocking) சோதனையைச் செய்கிறது. எனினும், இந்த முறைக்கு ஒரு நுட்பமான வரம்பு உள்ளது: event loop கட்டுப்பாட்டை விட்டுக்கொடுக்கும்போது மட்டுமே இது துண்டிப்புகளைக் கண்டறிகிறது. உங்கள் async generator yield களுக்கு இடையே CPU-சார்ந்த வரிசைப்படுத்தல் படியைச் செய்தால், துண்டிப்புச் சோதனை ஒன்று அல்லது அதற்கு மேற்பட்ட டோக்கன்களால் பின்தங்கலாம்.

பின்வரும் குறியீடு, is_disconnected வாக்கெடுப்பு அணுகுமுறையை generator சுத்தம் செய்தலுக்கான asyncio.shield பாதுகாப்புடன் இணைக்கும், தயாரிப்பு-கடினப்படுத்தப்பட்ட துண்டிப்பு கண்டறிதல் முறையை விளக்குகிறது. guarded_sse_stream செயல்பாடு மூல வழங்குநர் டோக்கன் iterator ஐ மூடி, கிளையன்ட் ஸ்ட்ரீமின் நடுவில் துண்டித்தாலும், திறந்த HTTP இணைப்புகள் அல்லது gRPC ஸ்ட்ரீம்கள் போன்ற வழங்குநர்-பக்க வளங்களை விடுவிக்க இறுதி சுத்தம் செய்யும் coroutine இயங்குவதை உறுதி செய்கிறது. _check_disconnect coroutine கிளையன்ட் துண்டிக்கும்போது cancel_event ஐ அமைக்கும் பின்னணிப் பணியாக இயங்குகிறது; இதனால் generator அடுத்த வழங்குநர் துண்டு வரும் வரை காத்திராமல் உடனடியாக வெளியேற முடிகிறது.

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 generator ஐ அறிவிக்கிறது. இது சரிபார்க்கப்பட்ட ChatRequest உடலையும், துண்டிப்பு ஆய்வுக்கான மூல Request பொருளையும் ஏற்றுக்கொள்கிறது.
• வரி 4: asyncio.Event நிகழ்வு, துண்டிப்பு கண்காணிப்பாளருக்கும் முதன்மை generator சுழற்சிக்கும் இடையே பகிரப்படும் thread-safe கொடியாகச் செயல்படுகிறது. boolean க்குப் பதிலாக event ஐப் பயன்படுத்துவது இரு coroutine களுக்கு இடையிலான race condition களைத் தவிர்க்கிறது.
• வரிகள் 6-11: _watch_disconnect coroutine ஒவ்வொரு 250 மில்லிவினாடிகளுக்கும் request.is_disconnected() ஐ வாக்கெடுப்பு செய்கிறது. 0.25-வினாடி இடைவெளி பதிலளிக்கும் தன்மைக்கும் CPU மேல்சுமைக்கும் இடையே சமநிலையைப் பேணுகிறது — 100ms ஐ விட அடிக்கடி வாக்கெடுப்பு செய்வது நடைமுறைப் பயனைத் தராது, ஏனெனில் load balancer கள் வழியாக TCP FIN பரவலுக்கு பொதுவாக 50-200ms ஆகும். துண்டிப்பு கண்டறியப்பட்டதும், event உடனடியாக அமைக்கப்பட்டு, generator ஐ yield செய்வதை நிறுத்தச் சமிக்ஞை செய்கிறது.
• வரி 13: கண்காணிப்பாளர் coroutine பின்னணி asyncio.Task ஆகத் தொடங்கப்படுகிறது. இது generator இன் டோக்கன் மீள்செயல் சுழற்சியைத் தடுக்காமல் அதனுடன் ஒரே நேரத்தில் இயங்குவதை உறுதி செய்கிறது.
• வரிகள் 14-20: முதன்மை உருவாக்கச் சுழற்சி ஒவ்வொரு பிரேமையும் yield செய்வதற்கு முன் cancel_event.is_set() ஐச் சரிபார்க்கிறது. இந்தச் சோதனை கிட்டத்தட்ட உடனடியானது (இது ஒரு உள் boolean ஐப் படிக்கிறது) மற்றும் கண்காணிப்பாளர் துண்டிப்பைக் கண்டறிந்ததும் மில்லிவினாடிக்கும் குறைவான ரத்து தாமதத்தை வழங்குகிறது. ரத்து இல்லாமல் ஸ்ட்ரீம் இயல்பாக முடிந்தால் மட்டுமே done நிகழ்வு வெளியிடப்படுகிறது.
• வரிகள் 21-22: asyncio.CancelledError கையாளி, Uvicorn இன் ASGI சர்வர் பதில் coroutine ஐ நேரடியாக ரத்து செய்யும் நிலையைப் பிடிக்கிறது (செயலில் உள்ள ஸ்ட்ரீமின்போது சர்வர் நிறுத்தப்படும்போது இது நிகழ்கிறது). பிழை பிரேம் கிளையன்ட்டுக்கு மூல இணைப்புத் துண்டிப்புக்குப் பதிலாக கட்டமைக்கப்பட்ட ரத்து சமிக்ஞையை வழங்குகிறது.
• வரிகள் 23-29: finally தொகுதி generator எப்படி வெளியேறினாலும் சுத்தம் செய்தலை உத்தரவாதம் செய்கிறது. இது cancel event ஐ அமைக்கிறது (ஏற்கனவே அமைக்கப்பட்டிருந்தால் idempotent), கண்காணிப்பாளர் பணியை ரத்து செய்கிறது, மற்றும் அதன் நிறைவுக்காக await செய்கிறது. await watcher ஐச் சுற்றியுள்ள உள் try/except, ஒரு பணி அதன் அடுத்த await புள்ளிக்கு முன் ரத்து செய்யப்படும்போது பரவும் CancelledError ஐ அமைதிப்படுத்துகிறது.

துறை பயன்பாடு

செய்ய வேண்டியவை மற்றும் செய்யக்கூடாதவை

செய்ய வேண்டியவை

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 வெளிப்படுத்தும் ஒரே வழிமுறை இதுதான்; நேர்மறைச் சோதனையில் generator இலிருந்து வெளியேறுவது மேல்நிலை dispatch_provider அழைப்பை நிறுத்தி, எந்தக் கிளையன்ட்டும் படிக்கப்போகாத டோக்கன்களுக்கான கட்டணத்தைத் தவிர்க்கிறது.

செய்யக்கூடாதவை

1. நேட்டிவ் உலாவி EventSource API ஐப் பயன்படுத்தி இந்த எண்ட்பாயிண்ட்டுடன் இணைக்காதீர்கள் — EventSource GET-மட்டுமே மற்றும் ChatRequest POST உடலைக் கொண்டு செல்ல முடியாது; ReadableStream ரீடருடன் fetch() ஐ அல்லது @microsoft/fetch-event-source polyfill ஐப் பயன்படுத்துங்கள், இவை இரண்டும் SSE பதிலின் மீது POST சொற்பொருளை ஆதரிக்கின்றன.
2. format_sse வழியாக அனுப்பாமல் SSE பிரேம்களை இன்லைன் f-string களாகக் கையால் எழுதாதீர்கள் — event:, data:, மற்றும் \n சரங்களைக் கையால் இணைக்கும்போது இரட்டை-புதிய-வரி முடிப்பானே விடுபட அதிக வாய்ப்புள்ள உறுப்பு; அதனால் ஏற்படும் அமைதியான பஃபரிங் தோல்வி, Nagle ஒன்றிணைப்பு அதை மறைப்பதை நிறுத்தும் தயாரிப்பு சுமையின் கீழ் மட்டுமே தோன்றும்.
3. டோக்கன்களை ஒரு பட்டியலில் பஃபர் செய்து StreamingResponse க்குப் பதிலாக JSONResponse ஐத் திருப்பாதீர்கள் — சேர்த்து வைப்பது, AsyncGenerator[str, None] உடன் இணைந்த StreamingResponse நீக்குவதற்காகவே இருக்கும் பல-வினாடி முதல்-டோக்கன்-வரை-நேர உறைதலை மீண்டும் அறிமுகப்படுத்துகிறது, மேலும் கட்டுப்பாடற்ற மேல்நிலை டோக்கன் நுகர்வைத் தடுக்கும் is_disconnected() சோதனைப் புள்ளியை நீக்குகிறது.