fix(stream): streaming enable to work with proxy

[jianshen92]

What does this PR address?

Relates to #4310

1. streaming now just stream raw text instead of pickled data container
2. provides to_sse and from_sse to be used at boundaries (api server and runner) if user wants to stream text in SSE format

Fixes #(issue)

Before submitting:

• Does the Pull Request follow Conventional Commits specification naming? Here are GitHub's
  guide on how to create a pull request.
• Does the code follow BentoML's code style, pre-commit run -a script has passed (instructions)?
• Did you read through contribution guidelines and follow development guidelines?
• Did your changes require updates to the documentation? Have you updated
  those accordingly? Here are documentation guidelines and tips on writting docs.
• Did you write tests to cover your changes?

[aarnphm]

@jianshen92 I think might be worth merging this to 1.2

[jianshen92]

@jianshen92 I think might be worth merging this to 1.2

@bojiang should I be merging to 1.2 instead?

[bojiang]

‌‌‌@aarnphm This is a bug fix. It's affecting the deployment of openllm under certain circumstances. There's still some time for 1.2 to release. Let's merge to main first?

[bojiang]

‌‌‌‌By the way, I think what Aaron meant might not be that we don't offer these utils, but rather that we provide utils which then return SSE objects. @jianshen92
I have the same thought, and I wrote some unit test accordingly. Let me push them here.
You can run them by

pip install pytest pytest-asyncio
pytest tests/unit/test_utils.py

[aarnphm]

@bojiang @jianshen92 what is the status of this PR?

[bojiang]

@aarnphm Jianshen is still working on the utils.

[jianshen92]

Usage:

runner

class Runner(bentoml.Runnable):
    ...
    @bentoml.Runnable.method()
    async def raw_text_streaming():
        yield "this is valid"


    @bentoml.Runnable.method()
    async def sse_streaming():
        yield SSE(data="my data", event="my event", id="a123cbx", retry=123)

api-server

# if we expect raw text, nothing changes. we dont care if they get chunked 
# if we expect a SSE text stream, we can use the `from_chunks` method to reconstruct the chunked data
async for event in SSE.from_chunks(runner_iterator):
    yield event.to_str()
    yield event.data