ArtyomZemlyak
commented
6 months ago Code for clients:
import json
from time import time
import requests
class EventStream:
"""Accepts lines of text and decodes it into a stream of SSE events.
Refer to the following page for details:
https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
This class is supposed to be iterated with a for loop like:
>>> for event in EventStream(lines):
... do_something_with(event)
"""
def __init__(self, lines, encoding="utf-8"):
self._lines = lines
self._encoding = encoding
@property
def decoded_lines(self):
for line in self._lines:
yield line.decode(self._encoding)
def __iter__(self):
return self
def __next__(self):
return Event.parse_from_lines(self.decoded_lines)
class Event:
"""A single event in the event stream."""
def __init__(self):
self.id = None
self.event = None
self.data = ""
def append_line(self, line):
if not line:
raise ValueError(
"Not supposed to accept empty lines. Please handle this outside of the Event class."
)
if ":" not in line:
raise ValueError("Bad format: Each line must contain `:`.")
parts = line.split(":", maxsplit=1)
if len(parts) < 2:
raise ValueError(
"Bad format: Each line must could be splitted into two parts by ':'."
)
prefix = parts[0]
data = parts[1].strip()
if prefix == "id":
if self.id is not None:
raise ValueError(
"Bad event: event id cannot be specified multiple times."
)
self.event = data
if prefix == "event":
if self.event is not None:
raise ValueError(
"Bad event: event type cannot be specified multiple times."
)
self.event = data
if prefix == "data":
if not self.data:
self.data = data
else:
self.data = "\n".join((self.data, data))
# TODO: Handle other prefixes here
@staticmethod
def parse_from_lines(lines_stream):
"""Given a lines stream, parse an event from it.
It only parse the first event. The remainder are not touched.
"""
result = Event()
for line in lines_stream:
if not line:
return result
else:
result.append_line(line)
# If we reached the end of the input lines stream,
# raise StopIteration to indicate that no more events will happen
raise StopIteration()
def __str__(self):
# Defaults to "message" when event name is not defined.
event_name = self.event or "message"
return f"Event ({event_name}): {self.data}"
def get_stream(question, chat_history, project, url):
headers = {"Content-Type": "text/event-stream", "Accept": "text/event-stream"}
request_data = {"question": question, "chat_history": chat_history, "project": project}
response = requests.post(url, json=request_data, headers=headers, stream=True)
response.raise_for_status()
content_type = response.headers.get("Content-Type")
if "text/event-stream" in content_type:
event_stream = EventStream(response.iter_lines())
for event in event_stream:
data = event.data
data = json.loads(data)
yield data["answer"]
else:
raise TypeError(f"{content_type=} is not text/event-stream")
if __name__ == "__main__":
question = "tell about"
chat_history = []
url = "http://localhost:8080/score"
project = "hr"
st = time()
for line in get_stream(question, chat_history, project, url):
print(line, end="")
print()
print(time() - st)
wangchao1230
liucheng-ms
Describe the bug
Promptflow servercant handle multiple requests with streaming output.How To Reproduce the bug
promptflow serverwith stream chat outputpromptflow serverpromptflow server("Connection broken: InvalidChunkLength(got length b'', 0 bytes read)", InvalidChunkLength(got length b'', 0 bytes read)Expected behavior On 5 step first and second terminals worked in streaming format and get results from
promptflow serverLogs 1 terminal
Logs promptflow server