Build and Stream a Generative AI API

Why stream an AI response?

An LLM may take several seconds to produce a complete answer. A normal JSON endpoint makes the client wait for the entire result. Streaming delivers small events as they become available, improving perceived responsiveness and allowing progress signals.

Streaming does not make generation cheaper or automatically faster. It changes how output travels to the client.

Choose an event contract

Do not stream anonymous text fragments without meaning. Define event types:

01meta      → request accepted, request ID02token     → incremental answer text03citation  → validated source metadata04done      → stream completed05error     → safe failure information

Server-Sent Events (SSE) is a practical one-way server-to-client format for web applications.

01event: token02data: {"text":"Employees may carry"}0304event: citation05data: {"source_id":"hr-2026-4.2","title":"Employee Handbook","page":18}

Define the request

01from pydantic import BaseModel, Field0203class AssistantRequest(BaseModel):04    question: str = Field(min_length=3, max_length=2000)05    conversation_id: str | None = None

The user does not provide tenant, role or document filters. The server derives them from authenticated identity.

Build the event encoder

01import json0203def sse(event: str, payload: dict) -> str:04    data = json.dumps(payload, ensure_ascii=False)05    return f"event: {event}\ndata: {data}\n\n"

Never place raw untrusted newline content directly into the SSE protocol. Serialising JSON gives the client a predictable envelope.

Stream the grounded answer

01from collections.abc import AsyncIterator02from fastapi.responses import StreamingResponse0304async def answer_events(payload, user) -> AsyncIterator[str]:05    request_id = create_request_id()06    yield sse("meta", {"request_id": request_id})0708    evidence = await retriever.search(09        question=payload.question,10        tenant_id=user.tenant_id,11        access_groups=user.groups,12    )1314    if not evidence:15        yield sse("error", {16            "code": "insufficient_evidence",17            "message": "No authorised source can answer this question.",18        })19        return2021    allowed_ids = {item.source_id for item in evidence}2223    async for text in model.stream(24        question=payload.question,25        evidence=evidence,26    ):27        yield sse("token", {"text": text})2829    for item in evidence:30        if item.source_id in allowed_ids:31            yield sse("citation", item.public_metadata())3233    yield sse("done", {"request_id": request_id})

01@app.post("/assistant/stream")02async def stream_answer(payload: AssistantRequest, user: CurrentUser):03    return StreamingResponse(04        answer_events(payload, user),05        media_type="text/event-stream",06        headers={"Cache-Control": "no-cache"},07    )

This example teaches the boundary. In production, handle disconnects, provider cancellation, trace completion and safe partial results.

Keep citations outside generated prose

The model may mention [S1], but application code should map S1 to a source actually retrieved for this request. Send citations as separate structured events so the interface can render safe titles and URLs.

01Model produces: answer text + source labels02Server verifies: label ∈ retrieved source IDs03Client receives: safe citation metadata owned by server

Stream errors deliberately

Once response headers are sent, changing the HTTP status may no longer be possible. Validate authentication, request shape and basic limits before creating the stream. For failures during streaming, send a defined error event, close cleanly and record the trace.

Real-time analogy: live sports commentary

Expose RAG and agents behind service interfaces

The route should not know which vector database or agent framework is used. It depends on capabilities:

01class EvidenceRetriever:02    async def search(self, question: str, tenant_id: str, access_groups: list[str]): ...0304class AnswerGenerator:05    async def stream(self, question: str, evidence: list): ...

This lets you replace LangChain, provider SDKs or storage without breaking the public endpoint.

Common mistakes

Interview bit

Key takeaways

Frequently asked questions

Does streaming make the model generate faster? Not necessarily. It improves perceived responsiveness by returning useful events before the full answer is complete.

How should a streamed API report an error after output begins? Define explicit event types so the client can distinguish content, citation, completion and error events without parsing free-form text.

When should citations be sent? Send only citations backed by retrieved, authorised evidence and keep enough identifiers to trace the source.

Interactive Knowledge Check

Primary sources

• FastAPI: Stream data
• Starlette responses: StreamingResponse
• FastAPI: Response headers