Repository: xorbitsai/inference
Pull request: BUG: Fix stream not compatible with openai
OpenAI stream response always split by line and each line is a valid json with `data:` prefix.
Fixes: https://github.com/xorbitsai/inference/issues/523

Problem statement:
BUG: Stream response does not compatible with openai
### Describe the bug
A clear and concise description of what the bug is.

```python
openai.error.APIError: HTTP code 200 from API ({"id": "chatcmpl-346fc892-932c-464a-8051-9df38827d566", "model": "my_code_llama", "created": 1696932753, "object": "chat.completion.chunk", "choices": [{"index": 0, "delta": {"role": ....

```

```python
Traceback (most recent call last):

  File "openai/api_requestor.py", line 673, in _interpret_response_line

  File "json/__init__.py", line 346, in loads

  File "json/decoder.py", line 337, in decode

  File "json/decoder.py", line 355, in raw_decode

json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)


The above exception was the direct cause of the following exception:


Traceback (most recent call last):

  File "server/continuedev/core/autopilot.py", line 437, in _run_singular_step

  File "server/continuedev/core/main.py", line 376, in __call__

  File "server/continuedev/plugins/steps/chat.py", line 106, in run

  File "/var/folders/r6/h3hc6kj91s9czcyds6fh0yqh0000gn/T/_MEIxFzRFC/continuedev/libs/llm/base.py", line 406, in stream_chat
    async for chunk in self._stream_chat(messages=messages, options=options):

  File "/var/folders/r6/h3hc6kj91s9czcyds6fh0yqh0000gn/T/_MEIxFzRFC/continuedev/libs/llm/openai.py", line 133, in _stream_chat
    async for chunk in await openai.ChatCompletion.acreate(
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "openai/api_resources/chat_completion.py", line 45, in acreate

  File "openai/api_resources/abstract/engine_api_resource.py", line 217, in acreate

  File "openai/api_requestor.py", line 310, in arequest

  File "openai/api_requestor.py", line 646, in _interpret_async_response

  File "openai/api_requestor.py", line 675, in _interpret_response_line
```

### To Reproduce
To help us to reproduce this bug, please provide information below:

1. Your Python version.
2. The version of xinference you use.
3. Versions of crucial packages.
4. Full stack of the error.
5. Minimized code to reproduce the error.

### Expected behavior
A clear and concise description of what you expected to happen.

### Additional context
Add any other context about the problem here.

Hints from issue discussion:

Commit message: Refine ut

Patch to review:
diff --git a/.github/workflows/python.yaml b/.github/workflows/python.yaml
index a5bd69ec4d..25bb360901 100644
--- a/.github/workflows/python.yaml
+++ b/.github/workflows/python.yaml
@@ -97,6 +97,7 @@ jobs:
           pip install s3fs
           pip install modelscope
           pip install -e ".[dev]"
+          pip install openai
         working-directory: .
 
       - name: Test with pytest
diff --git a/xinference/client/common.py b/xinference/client/common.py
index e2a37076d5..b3d05450f6 100644
--- a/xinference/client/common.py
+++ b/xinference/client/common.py
@@ -13,20 +13,23 @@
 # limitations under the License.
 
 import json
-from typing import Iterator
-
-from ..types import ChatCompletionChunk, CompletionChunk
+from typing import Any, Iterator
 
 
 def streaming_response_iterator(
-    response_chunk: Iterator[bytes],
-) -> Iterator["CompletionChunk"]:
+    response_lines: Iterator[bytes],
+) -> Iterator[Any]:
     """
     Create an Iterator to handle the streaming type of generation.
 
+    Note
+    ----------
+    This method is for compatible with openai. Please refer to:
+    https://github.com/openai/openai-python/blob/v0.28.1/openai/api_requestor.py#L99
+
     Parameters
     ----------
-    response_chunk: Iterator[bytes]
+    response_lines: Iterator[bytes]
         Generated lines by the Model Generator.
 
     Returns
@@ -36,36 +39,12 @@ def streaming_response_iterator(
 
     """
 
-    for chunk in response_chunk:
-        content = json.loads(chunk.decode("utf-8"))
-        error = content.get("error", None)
-        if error is not None:
-            raise Exception(str(error))
-        yield content
-
-
-# Duplicate code due to type hint issues
-def chat_streaming_response_iterator(
-    response_chunk: Iterator[bytes],
-) -> Iterator["ChatCompletionChunk"]:
-    """
-    Create an Iterator to handle the streaming type of generation.
-
-    Parameters
-    ----------
-    response_chunk: Iterator[bytes]
-        Generated chunk by the Model Generator.
-
-    Returns
-    -------
-    Iterator["ChatCompletionChunk"]
-        Iterator of ChatCompletionChunks generated by models.
-
-    """
-
-    for chunk in response_chunk:
-        content = json.loads(chunk.decode("utf-8"))
-        error = content.get("error", None)
-        if error is not None:
-            raise Exception(str(error))
-        yield content
+    for line in response_lines:
+        line = line.strip()
+        if line.startswith(b"data:"):
+            data = line[len(b"data:") :].strip()
+            data = json.loads(data.decode("utf-8"))
+            yield data
+        elif line.startswith(b"error:"):
+            error = line[len(b"error:") :].strip()
+            raise Exception(error.decode("utf-8"))
diff --git a/xinference/client/restful/restful_client.py b/xinference/client/restful/restful_client.py
index a6a6d16e1c..30b127c26e 100644
--- a/xinference/client/restful/restful_client.py
+++ b/xinference/client/restful/restful_client.py
@@ -17,7 +17,7 @@
 
 import requests
 
-from ..common import chat_streaming_response_iterator, streaming_response_iterator
+from ..common import streaming_response_iterator
 
 if TYPE_CHECKING:
     from ...types import (
@@ -128,7 +128,7 @@ def generate(
             )
 
         if stream:
-            return streaming_response_iterator(response.iter_content(chunk_size=None))
+            return streaming_response_iterator(response.iter_lines())
 
         response_data = response.json()
         return response_data
@@ -206,9 +206,7 @@ def chat(
             )
 
         if stream:
-            return chat_streaming_response_iterator(
-                response.iter_content(chunk_size=None)
-            )
+            return streaming_response_iterator(response.iter_lines())
 
         response_data = response.json()
         return response_data
@@ -272,9 +270,7 @@ def chat(
             )
 
         if stream:
-            return chat_streaming_response_iterator(
-                response.iter_content(chunk_size=None)
-            )
+            return streaming_response_iterator(response.iter_lines())
 
         response_data = response.json()
         return response_data
diff --git a/xinference/core/restful_api.py b/xinference/core/restful_api.py
index a2569de5c0..82853904f1 100644
--- a/xinference/core/restful_api.py
+++ b/xinference/core/restful_api.py
@@ -518,12 +518,15 @@ async def stream_results():
                 try:
                     iterator = await model.generate(body.prompt, kwargs)
                     async for item in iterator:
-                        yield json.dumps(item)
+                        yield f"data: {json.dumps(item)}\n"
                 except Exception as ex:
                     logger.exception("Completion stream got an error: %s", ex)
-                    yield json.dumps({"error": str(ex)})
+                    yield f"error: {ex}"
 
-            return StreamingResponse(stream_results())
+            # The Content-Type: text/event-stream header is required for openai stream.
+            return StreamingResponse(
+                stream_results(), headers={"Content-Type": "text/event-stream"}
+            )
         else:
             try:
                 return await model.generate(body.prompt, kwargs)
@@ -631,12 +634,15 @@ async def stream_results():
                             prompt, system_prompt, chat_history, kwargs
                         )
                     async for item in iterator:
-                        yield json.dumps(item)
+                        yield f"data: {json.dumps(item)}\n"
                 except Exception as ex:
                     logger.exception("Chat completion stream got an error: %s", ex)
-                    yield json.dumps({"error": str(ex)})
+                    yield f"error: {ex}"
 
-            return StreamingResponse(stream_results())
+            # The Content-Type: text/event-stream header is required for openai stream.
+            return StreamingResponse(
+                stream_results(), headers={"Content-Type": "text/event-stream"}
+            )
         else:
             try:
                 if is_chatglm_ggml:
diff --git a/xinference/core/tests/test_restful_api.py b/xinference/core/tests/test_restful_api.py
index 88ccbe9db1..f8a82931a0 100644
--- a/xinference/core/tests/test_restful_api.py
+++ b/xinference/core/tests/test_restful_api.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import openai
 import pytest
 import requests
 
@@ -344,3 +345,54 @@ def test_restful_api_for_embedding(setup):
     response = requests.get(f"{endpoint}/v1/models")
     response_data = response.json()
     assert len(response_data) == 0
+
+
+@pytest.mark.asyncio
+async def test_openai(setup):
+    endpoint, _ = setup
+    url = f"{endpoint}/v1/models"
+
+    # list
+    response = requests.get(url)
+    response_data = response.json()
+    assert len(response_data) == 0
+
+    # launch
+    payload = {
+        "model_uid": "test_restful_api",
+        "model_name": "orca",
+        "quantization": "q4_0",
+    }
+
+    response = requests.post(url, json=payload)
+    response_data = response.json()
+    model_uid_res = response_data["model_uid"]
+    assert model_uid_res == "test_restful_api"
+
+    openai.api_key = ""
+    openai.api_base = f"{endpoint}/v1"
+
+    # chat
+    messages = [
+        {"role": "system", "content": "You are a helpful assistant."},
+        {"role": "user", "content": "Hello!"},
+        {"role": "assistant", "content": "Hi what can I help you?"},
+        {"role": "user", "content": "What is the capital of France?"},
+    ]
+
+    result = []
+    async for chunk in await openai.ChatCompletion.acreate(
+        messages=messages, stream=True, model=model_uid_res
+    ):
+        if not hasattr(chunk, "choices") or len(chunk.choices) == 0:
+            continue
+        result.append(chunk)
+    assert result
+    assert type(result[0]).__name__ == "OpenAIObject"
+
+    result = await openai.ChatCompletion.acreate(
+        messages=messages, stream=False, model=model_uid_res
+    )
+
+    assert result
+    assert type(result).__name__ == "OpenAIObject"


Code review comments:
diff_hunk: @@ -97,6 +97,7 @@ jobs:
           pip install s3fs
           pip install modelscope
           pip install -e ".[dev]"
+          pip install openai
path: .github/workflows/python.yaml
text: @user1:
openai should be added to [dev] so that developers can run the unittests without any dependency problem.

@author:
Good suggestion. Thanks.