⚡ FastAPI & SSE Streaming

For a highly responsive user experience, the Digital Twin streams answers token-by-token to create a dynamic typing effect. This is achieved using Server-Sent Events (SSE) in an asynchronous FastAPI backend, utilizing Groq's ultra-low latency inference with the Llama-3.3-70b-Versatile model.

---

📡 Why Server-Sent Events (SSE)?

Unlike WebSockets, which are full-duplex and require complex handshake handling, SSE offers a unidirectional stream from server to client over standard HTTP.

Advantages of SSE:

• Simplicity: Runs over standard HTTP/1.1 or HTTP/2 without protocol upgrades.
• Auto-reconnection: Built-in browser support for automatic reconnection.
• Low Overhead: Perfect for read-only streaming workloads like LLM response streaming.

---

💻 FastAPI Stream Implementation

We use the StreamingResponse class from FastAPI to handle HTTP-based server streaming.

import os
import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from groq import AsyncGroq

app = FastAPI()

# Enable CORS for Next.js frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

client = AsyncGroq(api_key=os.environ.get("GROQ_API_KEY"))

async def token_generator(prompt: str):
    """Asynchronously calls Groq and yields tokens in SSE format."""
    try:
        # Call Groq API with streaming enabled
        chat_completion = await client.chat.completions.create(
            messages=[
                {"role": "system", "content": "You are the digital twin of Ahmed BARGADY."},
                {"role": "user", "content": prompt}
            ],
            model="llama-3.3-70b-versatile",
            stream=True
        )
        
        async for chunk in chat_completion:
            token = chunk.choices[0].delta.content
            if token is not None:
                # SSE lines must begin with "data: " and end with double newlines
                yield f"data: {token}\n\n"
                await asyncio.sleep(0.01)  # Brief yield to release the loop
                
        # Send end marker
        yield "data: [DONE]\n\n"
        
    except Exception as e:
        yield f"data: Error: {str(e)}\n\n"

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

---

🛠️ Handling Connection Closes

In asynchronous streaming, it is crucial to release resources if the user closes their browser window or navigates away.

@app.get("/api/chat/stream-safe")
async def chat_stream_safe(prompt: str, request: Request):
    async def event_generator():
        generator = token_generator(prompt)
        try:
            async for token in generator:
                # Check if the client disconnected
                if await request.is_disconnected():
                    print("Client disconnected. Cleaning up...")
                    break
                yield token
        except asyncio.CancelledError:
            print("Stream cancelled task.")
            
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Groq Inference Speed

By combining Groq's LPUs (Language Processing Units) with Llama-3.3-70B, the twin achieves a Time-to-First-Token (TTFT) of under 150ms and pushes over 200 tokens per second to the client.