Problem with Stream response

[danechitoaie]

I have the following controller

    @post()
    async def get_chat(
        self,
        data: ChatRequest,
        prompts: Prompts,
        httpx_client: AsyncClient,
    ) -> Stream:
        ...
        return Stream(iterator=httpx_stream(httpx_client, httpx_json))

and my httpx_stream function is:

async def httpx_stream(httpx_client: AsyncClient, httpx_json: dict) -> AsyncGenerator[bytes, None]:
    async with aconnect_sse(
        httpx_client,
        "POST",
        "https://api.openai.com/v1/chat/completions",
        json=httpx_json,
    ) as event_source:
        async for event in event_source.aiter_sse():
            if event.data == "[DONE]":
                break
            data = decode_json(event.data)
            for choice in data["choices"]:
                delta = choice["delta"]
                if "content" in delta:
                    yield encode_json(
                        {
                            "status": "OK",
                            "text": delta["content"],
                        }
                    )

And the problem is that this works the right way about 90% of the time.
So most of the requests are coming like:

{"status":"OK","text":"Hello"}
{"status":"OK","text":"!"}
{"status":"OK","text":" How"}
{"status":"OK","text":" can"}
{"status":"OK","text":" I"}

so one JSON object per HTTP chunk response (per yield).

But randomly it also comes like this:

{"status":"OK","text":"Hello"}
{"status":"OK","text":"!"}
{"status":"OK","text":" How"}{"status":"OK","text":" can"}
{"status":"OK","text":" I"}

So it seems sometimes it puts 2 yields on same HTTP chunk.
Any idea what's going on? Why it doesn't always respect one yield per chunk?

[provinzkraut]

Can you provide an MCVE of this without the httpx and SSE stuff?

[provinzkraut]

I'd have to look at some uvicorn source code to be sure, but the way I understand this

Protocol servers must flush any data passed to them into the send buffer before returning from a send call

Is that each await send(event) must flush data into the server's buffer, but does not require the server to flush its buffer or actually perform a send of the data therein.

Reading the uvicorn documentation, this section seems to confirm this:

If the write buffer passes a high water mark, then Uvicorn ensures the ASGI send messages will only return once the write buffer has been drained below the low water mark.

send calls first write to the buffer, but there are no guarantees as to how and when uvicorn will actually send the contents of that buffer during the lifetime of a request.

[danechitoaie]

Then I'll need to find another way to encode and send the data in chunks then in a way that can be correctly parsed on the client side.

[provinzkraut]

I think sending JSON in chunks like that isn't a great idea anyway, since it doesn't lend itself that well to being chunked.

Maybe you can use WebSockets or SSE?

[danechitoaie]

SSE was my first option but Litestar does not support it.

[provinzkraut]

We have an issue to track SSE support: #1578.

PRs are welcome (=