Streaming LLM Responses: SSE with Python

[chawkian]

----------|----------|
| First token | Wait for full generation | ~200ms |
| UX | Loading spinner | Text appears progressively |
| Timeout risk | Higher (long generation) | Lower |
| Complexity | One request | Connection management |

For chat applications and IDE tools, streaming is expected behavior.

Server Implementation (FastAPI)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx, json

app = FastAPI()

@app.get("/chat")
def chat_stream(prompt: str):
    async def generate():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                "https://aiwave.live/v1/chat/completions",
                json={
                    "model": "deepseek-chat",
                    "messages": [{"role": "user", "content": prompt}],
                    "stream": True
                },
                headers={"Authorization": "***"}
            ) as resp:
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        data = json.loads(line[6:])
                        if "choices" in data:
                            delta = data["choices"][0].get("delta", {})
                            content = delta.get("content", "")
                            if content:
                                yield "data: " + json.dumps({"text": content}) + "\n\n"
                yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Client Implementation

import httpx, json

def stream_chat(prompt):
    with httpx.stream(
        "POST",
        "https://aiwave.live/v1/chat/completions",
        json={
            "model": "deepseek-chat",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True
        },
        headers={"Authorization": "***"}
    ) as resp:
        for line in resp.iter_lines():
            if line.startswith("data: "):
                data = json.loads(line[6:])
                if "choices" in data:
                    content = data["choices"][0].get("delta", {}).get("content", "")
                    if content:
                        print(content, end="", flush=True)
    print()

stream_chat("Explain async/await in Python")

Key Details

• SSE format: Each line is data: {json}\n\n. The final line is data: [DONE]\n\n
• Stream=True: This parameter enables streaming on all OpenAI-compatible APIs
• Works with all AIWave models: DeepSeek, GLM, Kimi, Qwen — same interface
• Backpressure: The server sends data as fast as the model generates. If the client reads slower, the connection buffer handles it
• Timeout: Set 30-60s for long generations

When to Stream

• Chat UIs and chatbots
• IDE autocomplete and inline suggestions
• Any UI where users see output in real-time

Skip streaming for batch processing, background jobs, and API-to-API calls where latency doesn't matter.

All models on AIWave support streaming via the standard stream: true parameter. One API key, same OpenAI-compatible format. The $1 free credit covers thousands of streaming requests to test.

---

AIWave - 48 models, one key. $1 free.

---

Resources

• AIWave pricing
• AIWave developer blog
• AIWave Python SDK