The problem of SSE in Docker environment

[EvilCalf]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], 
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"], 
)

@app.post("/chat")
async def streamed_response(request: ChatRequest):
    try:
        if request.query == "":
            logger.error("Query can not be empty")
            raise HTTPException(status_code=400, detail="Query can not be empty")

        def event_generator():
            try:
                bot = init_agent_service(request.model_id)
                messages = task_history2messages(request.task_history, request.query)
                
                pre_response = ""
                response = []
                for response in bot.run(messages):
                    new_response = response[-1]["content"].replace(pre_response, "")
                    pre_response = response[-1]["content"]
                    if (
                        response[-1]["role"] != "function"
                        and new_response != ""
                        and new_response != "<|im_end|>'"
                    ):
                        # print(time())
                        yield "data:" + ModelResponse(
                            response=new_response
                        ).model_dump_json() + "\n\n"

                if "病情分析" in response[-1]["content"]:
                    yield "data:" + ModelResponse(
                        response="", status="stop"
                    ).model_dump_json() + "\n\n"
                messages.extend(response)

                save_response_to_db(messages)

            except Exception as e:
                logger.error(f"Error processing task history: {e}")
                raise HTTPException(status_code=500, detail="LLM Output Error")

        return StreamingResponse(event_generator(),media_type="text/event-stream")

    except HTTPException as http_exception:
        logger.error(
            f"HTTP Exception in streamed_response: {http_exception.status_code}"
        )
        raise http_exception
    except Exception as e:
        logger.error(f"Exception in streamed_response: {type(e).__name__}")
        raise HTTPException(status_code=500, detail="Internal server error")

Description

In the server environment I developed and the Docker environment of the developed server, this code has no issues and implements SSE.

There is no problem on the production cloud server, but SSE cannot be implemented in the Docker on the production server. It will combine all the data together and output it in segments, sometimes missing some parts.

Operating System

Linux

Operating System Details

Linux VM-8-4-ubuntu 4.15.0-213-generic #224-Ubuntu SMP Mon Jun 19 13:30:12 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux

FastAPI Version

0.111.0

Pydantic Version

2.7.1

Python Version

3.11

Additional Context

No response

[YuriiMotov]

What about FastAPI versions and other installed libraries? A they the same on your local server and in docker container?

Can you show some example output for both cases?

[EvilCalf]

They are some output on my local server and in docker container?
The out put just like openai

            chat_response = client.chat.completions.create(
                  model=model_id,
                  messages=messages,
                  temperature=0.8,
                  stream=True,
              )

              
              messages = ""
              for chunk in chat_response:
                  if chunk.choices[0].delta.content is not None:
                      messages += str(chunk.choices[0].delta.content)
                      yield ModelResponse(
                          response=chunk.choices[0].delta.content, status="success"
                      ).model_dump_json()
                      yield "test"

[YuriiMotov]

What FastAPI version do you have 'in the Docker on the production server'?
Are you sure it's the same version that you have on your development server?
Compare versions of FastAPI and other libraries you use on different servers (development server and Docker on production server). Is there any difference between them?

[EvilCalf]

yes, I used requirements.txt to pip install packages.

fastapi==0.111.0
openai==1.30.1
pydantic==2.7.1
sse_starlette==2.1.0
uvicorn==0.29.0

But docker version is different.

Cloud Server

Client:
 Version:           20.10.21
 API version:       1.41
 Go version:        go1.18.1
 Git commit:        20.10.21-0ubuntu1~18.04.3
 Built:             Thu Apr 27 05:50:21 2023
 OS/Arch:           linux/amd64
 Context:           default
 Experimental:      true

Server:
 Engine:
  Version:          20.10.21
  API version:      1.41 (minimum version 1.12)
  Go version:       go1.18.1
  Git commit:       20.10.21-0ubuntu1~18.04.3
  Built:            Thu Apr 27 05:36:22 2023
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          1.6.12-0ubuntu1~18.04.1
  GitCommit:        
 runc:
  Version:          1.1.4-0ubuntu1~18.04.2
  GitCommit:        
 docker-init:
  Version:          0.19.0
  GitCommit:   

Local Server

Client: Docker Engine - Community
 Version:           24.0.7
 API version:       1.43
 Go version:        go1.20.10
 Git commit:        afdd53b
 Built:             Thu Oct 26 09:07:41 2023
 OS/Arch:           linux/amd64
 Context:           default

Server: Docker Engine - Community
 Engine:
  Version:          24.0.7
  API version:      1.43 (minimum version 1.12)
  Go version:       go1.20.10
  Git commit:       311b9ff
  Built:            Thu Oct 26 09:07:41 2023
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          1.6.24
  GitCommit:        61f9fd88f79f081d64d6fa3bb1a0dc71ec870523
 runc:
  Version:          1.1.9
  GitCommit:        v1.1.9-0-gccaecfc
 docker-init:
  Version:          0.19.0
  GitCommit:        de40ad0

[YuriiMotov]

Try starting from scratch and create the simplest SSE-endpoint with minimal amount of dependencies.
Something like:

import time

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def streamed_response():

    def event_generator():
        for i in range(10):
            time.sleep(2)
            yield f'message {i}'


    return StreamingResponse(event_generator(),media_type="text/event-stream")

And test it with code like (run it from local machine):

import httpx

with httpx.stream("POST", "http://127.0.0.1:8000/chat") as response:
    for sse in response.iter_text():
        print(sse)

Does it work correctly?
If it works, try adding your dependencies line by line and running test after each step. Notice the line that brakes you code.