----------|----------|
| 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.
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")
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")
data: {json}\n\n. The final line is data: [DONE]\n\nSkip 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.