Implement an OpenAI GPT-4o streaming adapter

Loading diagram...

• Line 1: Declares this as a Mermaid sequence diagram, used to visualize interactions between components over time.
• Lines 2-5: Define the four participants (actors) in the diagram: the Browser using EventSource for SSE, a FastAPI endpoint, an OpenAIStreamAdapter that mediates streaming, and the OpenAI API itself.
• Line 7: The Browser initiates the flow by sending a GET request to the /chat/stream endpoint with provider=openai as a query parameter.
• Line 8: FastAPI delegates to the Adapter by calling stream_chat with the conversation messages, model name, and temperature setting.
• Line 9: The Adapter calls the OpenAI API's chat.completions.create method with stream=True to request a streaming response.
• Line 11: Begins a loop block that iterates over each ChatCompletionChunk received from the OpenAI streaming response.
• Line 12: OpenAI sends back a chunk containing a delta (partial content) to the Adapter via a dashed return arrow (async response).
• Line 13: The Adapter performs a self-call to extract the delta.content field from the received chunk.
• Lines 14-16: A conditional (alt) block that checks if delta.content is not None; if true, the Adapter sends an SSE frame back to FastAPI with event type "token" and the content text as the data payload.
• Lines 17-19: A second conditional block that checks if finish_reason is not None (indicating the stream is complete); if true, the Adapter sends an SSE frame to FastAPI with event type "done" and the finish reason (e.g., "stop") as the data payload.
• Line 20: Closes the loop block, ending the per-chunk iteration.
• Line 22: FastAPI sends the entire accumulated stream back to the Browser as a StreamingResponse with content type text/event-stream, completing the SSE connection.

1# models/sse.py
2from dataclasses import dataclass, field
3from pydantic import BaseModel
4from typing import Literal
5import json
6
7class ChatMessage(BaseModel):
8    role: Literal["system", "user", "assistant"]
9    content: str
10
11@dataclass
12class SSEFrame:
13    event: str
14    data: str
15    id: str | None = None
16    retry: int | None = None
17
18    def serialize(self) -> str:
19        lines = []
20        if self.id is not None:
21            lines.append(f"id: {self.id}")
22        if self.retry is not None:
23            lines.append(f"retry: {self.retry}")
24        lines.append(f"event: {self.event}")
25        for data_line in self.data.split("\n"):
26            lines.append(f"data: {data_line}")
27        lines.append("")
28        lines.append("")
29        return "\n".join(lines)

• Lines 1-4: Import the required modules. The dataclass decorator provides lightweight structure for SSE frames, while Pydantic's BaseModel gives runtime validation for incoming chat messages. The Literal type restricts role to the three values the OpenAI API accepts.
• Lines 6-8: ChatMessage uses Pydantic validation to reject any message whose role is not "system", "user", or "assistant". This catches malformed requests before they reach the provider API, preventing cryptic upstream errors.
• Lines 10-14: The SSEFrame dataclass carries four fields. The event field maps to the SSE event: line that the browser's EventSource.addEventListener uses for dispatch. The data field holds the payload—a token string for content events, or a JSON-encoded metadata object for control events. The optional id enables Last-Event-ID reconnection, and retry overrides the browser's default reconnection interval.
• Lines 16-24: The serialize method produces the exact wire format defined by the W3C SSE specification. Multi-line data values are split across separate data: lines per the spec. The two trailing newlines form the mandatory blank-line event terminator that tells the browser one complete event has ended.

1# adapters/openai_stream.py
2import asyncio
3from openai import AsyncOpenAI
4from models.sse import ChatMessage, SSEFrame
5from typing import AsyncGenerator
6import json
7import logging
8
9logger = logging.getLogger(__name__)
10
11class OpenAIStreamAdapter:
12    def __init__(self, api_key: str):
13        self.client = AsyncOpenAI(api_key=api_key)
14
15    async def stream_chat(
16        self,
17        messages: list[ChatMessage],
18        model: str = "gpt-4o",
19        temperature: float = 0.7,
20    ) -> AsyncGenerator[SSEFrame, None]:
21        formatted = [m.model_dump() for m in messages]
22        chunk_index = 0
23        try:
24            stream = await self.client.chat.completions.create(
25                model=model,
26                messages=formatted,
27                temperature=temperature,
28                stream=True,
29                stream_options={"include_usage": True},
30            )
31            async for chunk in stream:
32                if not chunk.choices:
33                    if chunk.usage is not None:
34                        yield SSEFrame(
35                            event="usage",
36                            data=json.dumps({
37                                "prompt_tokens": chunk.usage.prompt_tokens,
38                                "completion_tokens": chunk.usage.completion_tokens,
39                            }),
40                        )
41                    continue
42
43                choice = chunk.choices[0]
44                delta = choice.delta
45
46                if delta.content is not None:
47                    yield SSEFrame(
48                        event="token",
49                        data=delta.content,
50                        id=str(chunk_index),
51                    )
52                    chunk_index += 1
53
54                if choice.finish_reason is not None:
55                    yield SSEFrame(
56                        event="done",
57                        data=json.dumps({
58                            "finish_reason": choice.finish_reason,
59                            "chunk_count": chunk_index,
60                        }),
61                    )
62
63        except Exception as exc:
64            logger.error("OpenAI stream error: %s", exc)
65            yield SSEFrame(
66                event="error",
67                data=json.dumps({"error": str(exc)}),
68            )

• Lines 1-9: Imports bring in AsyncOpenAI for non-blocking HTTP/2 streaming, the shared ChatMessage and SSEFrame models, and Python's logging module. Using AsyncOpenAI rather than the synchronous OpenAI client is critical—synchronous streaming would block the entire FastAPI event loop, preventing concurrent request handling.
• Lines 11-13: The constructor accepts a raw API key string and instantiates AsyncOpenAI. In production, this key comes from environment variables or a secret manager, never from client-submitted parameters. The client instance is reusable across requests because AsyncOpenAI manages its own internal connection pool.
• Lines 15-20: The stream_chat method signature defines the provider-agnostic interface. The return type AsyncGenerator[SSEFrame, None] tells callers they will receive a sequence of SSEFrame objects via async for. The None send-type means this generator does not accept values sent into it via .asend().
• Lines 21-22: Messages are converted from Pydantic models to plain dictionaries using model_dump(). The chunk_index counter tracks how many content-bearing chunks have been emitted, providing monotonic id values for SSE reconnection semantics.
• Lines 23-30: The create call passes stream=True to activate the streaming protocol. The stream_options parameter with include_usage set to True is an OpenAI-specific feature that appends a final chunk containing token counts. Without this option, you cannot track usage for billing or rate-limit enforcement during streaming sessions.
• Lines 31-42: The inner loop iterates over chunks. The first guard checks whether chunk.choices is empty—this happens on the final usage-reporting chunk when include_usage is enabled. When chunk.usage is not None, the adapter yields a usage event carrying prompt and completion token counts as JSON. The continue statement skips the rest of the loop body since usage chunks have no delta content.
• Lines 44-45: For normal chunks, the code extracts choices[0] and its delta. The OpenAI API guarantees at least one choice when n=1 (the default), but the empty-choices guard on line 32 handles the usage-only edge case.
• Lines 47-53: The core content extraction. When delta.content is not None, the adapter wraps the token string in an SSEFrame with event type "token". The id field receives the monotonically increasing chunk_index, enabling browsers to resume from the last received event via the Last-Event-ID header if the connection drops.
• Lines 55-61: Finish-reason detection. The finish_reason field is None on every chunk except the last content-bearing one. When it becomes "stop" (normal completion), "length" (max-token cutoff), or "tool_calls" (function-calling mode), the adapter emits a done event. The payload includes both the reason and the total chunk count, giving the frontend enough information to display completion status or trigger follow-up actions.
• Lines 63-67: The outer try/except catches any exception—network timeouts, authentication failures, rate-limit errors (HTTP 429), or malformed responses. Rather than letting the exception propagate and crash the streaming response, the adapter emits an error event. This is essential for production resilience: the frontend receives a structured error frame instead of a broken TCP connection with no explanation.