:gem: [Feature] Enable OpenAI API call from fastapi with EventSourceResponse of sse_starlette

apis/chat_api.py

@@ -1,13 +1,25 @@
 import uvicorn
 
 from fastapi import FastAPI
-from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 from conversations import (
     ConversationConnector,
     ConversationCreator,
     ConversationSession,
 )
+from networks import OpenaiStreamOutputer
+from sse_starlette.sse import EventSourceResponse
+
+
+def mock_stream_chat(prompt):
+    outputer = OpenaiStreamOutputer()
+    for i in range(10):
+        output = outputer.output(content=f"MSG {i} ", content_type="Completions")
+        print(output)
+        yield output
+    output = outputer.output(content="", content_type="Finished")
+    print(output)
+    yield output
 
 
 class ChatAPIApp:
@@ -131,14 +143,14 @@ class ChatAPIApp:
             description="(int) Invocation ID",
         )
 
-    async def chat_completions(self, item: ChatCompletionsPostItem):
-        connector = ConversationConnector(
-            conversation_style=item.model,
-            sec_access_token=item.sec_access_token,
-            client_id=item.client_id,
-            conversation_id=item.conversation_id,
-            invocation_id=item.invocation_id,
-        )
+    def chat_completions(self, item: ChatCompletionsPostItem):
+        # connector = ConversationConnector(
+        #     conversation_style=item.model,
+        #     sec_access_token=item.sec_access_token,
+        #     client_id=item.client_id,
+        #     conversation_id=item.conversation_id,
+        #     invocation_id=item.invocation_id,
+        # )
 
         if item.invocation_id == 0:
             # TODO: History Messages Merger
@@ -146,9 +158,10 @@ class ChatAPIApp:
         else:
             prompt = item.messages[-1]["content"]
 
-        return StreamingResponse(
-            connector.stream_chat(prompt=prompt, yield_output=True),
-            media_type="text/plain",
+        return EventSourceResponse(
+            # connector.stream_chat(prompt=prompt, yield_output=True),
+            mock_stream_chat(prompt),
+            media_type="text/event-stream",
         )
 
     def setup_routes(self):

examples/chat_with_openai.py

@@ -22,17 +22,11 @@ response = client.chat.completions.create(
     extra_body=extra_body,
 )
 
-print(response)
-
-for chunk in response:
-    # print(chunk.choices[0].delta)
-    print("??")
-
-# print(response.choices[0].message)
 # print(response)
-
-# for chunk in stream:
-#     if chunk.choices[0].delta.content is not None:
-#         print(chunk.choices[0].delta.content)
-#     else:
-#         print(chunk)
+for chunk in response:
+    if chunk.choices[0].delta.content is not None:
+        print(chunk.choices[0].delta.content, end="", flush=True)
+    elif chunk.choices[0].finish_reason == "stop":
+        print()
+    else:
+        print(chunk)

examples/chat_with_post.py

@@ -3,7 +3,7 @@ import httpx
 import json
 import re
 
-
+# If runnning this service with proxy, you might need to unset `http(s)_proxy`.
 chat_api = "http://localhost:22222"
 api_key = "sk-xxxxx"
 requests_headers = {}

networks/message_outputer.py

@@ -7,14 +7,21 @@ class OpenaiStreamOutputer:
     * https://platform.openai.com/docs/api-reference/chat/create
     """
 
-    def data_to_string(self, data, content_type=""):
+    def data_to_string(self, data={}, content_type="", media_type=None):
         # return (json.dumps(data) + "\n").encode("utf-8")
-        data_str = f"data: {json.dumps(data)}\n"
-        if content_type == "Finished":
-            data_str += "data: [DONE]\n"
+        data_str = f"{json.dumps(data)}\n"
+
+        if media_type == "text/event-stream":
+            data_str = f"data: {data_str}"
+
         return data_str
 
-    def output(self, content=None, content_type=None) -> bytes:
+    def output(
+        self,
+        content=None,
+        content_type=None,
+        media_type=None,
+    ) -> bytes:
         data = {
             "created": 1677825464,
             "id": "chatcmpl-bing",
@@ -37,6 +44,8 @@ class OpenaiStreamOutputer:
             "InternalSearchResult",
             "SuggestedResponses",
         ]:
+            if content_type in ["InternalSearchQuery", "InternalSearchResult"]:
+                content += "\n"
             data["choices"] = [
                 {
                     "index": 0,
@@ -56,8 +65,8 @@ class OpenaiStreamOutputer:
             data["choices"] = [
                 {
                     "index": 0,
-                    "delta": {"content": ""},
+                    "delta": {},
                     "finish_reason": None,
                 }
             ]
-        return self.data_to_string(data, content_type)
+        return self.data_to_string(data, content_type, media_type)

networks/message_parser.py

@@ -50,7 +50,7 @@ class MessageParser:
                 # Message: Search Query
                 elif message_type in ["InternalSearchQuery"]:
                     message_hidden_text = message["hiddenText"]
-                    search_str = f"\n[Searching: [{message_hidden_text}]]\n"
+                    search_str = f"[Searching: [{message_hidden_text}]]"
                     logger.note(search_str)
                     if return_output:
                         return self.outputer.output(
@@ -58,7 +58,7 @@ class MessageParser:
                         )
                 # Message: Internal Search Results
                 elif message_type in ["InternalSearchResult"]:
-                    analysis_str = f"\n[Analyzing search results ...]\n"
+                    analysis_str = f"[Analyzing search results ...]"
                     logger.note(analysis_str)
                     if return_output:
                         return self.outputer.output(
@@ -85,5 +85,4 @@ class MessageParser:
                         f"Not Supported Message Type: {message_type}"
                     )
 
-        # return OpenaiStreamOutputer().output(content="", content_type="NotImplemented")
-        return b""
+        return self.outputer.output(content="", content_type="NotImplemented")