Spaces:
Paused
Paused
File size: 4,306 Bytes
bb37148 63998ab 6ad10eb 63998ab 6ad10eb 82cb440 6ad10eb 4c9b469 6ad10eb 1bbf682 63998ab 6ad10eb 4c9b469 0fdaf32 4c9b469 82cb440 4c9b469 e4f31be 0fdaf32 e4f31be 4c9b469 0fdaf32 4c9b469 6ad10eb 7bfabdf 4c9b469 6ad10eb 7bfabdf 4c9b469 6ad10eb 1bbf682 82cb440 6ad10eb 82cb440 6ad10eb 4c9b469 7bfabdf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
from utils.logger import logger
from networks import OpenaiStreamOutputer
class MessageParser:
def __init__(self, outputer=OpenaiStreamOutputer()):
self.delta_content_pointer = 0
self.outputer = outputer
def parse(self, data, return_output=False):
arguments = data["arguments"][0]
if arguments.get("throttling"):
throttling = arguments.get("throttling")
# pprint.pprint(throttling)
if arguments.get("messages"):
for message in arguments.get("messages"):
message_type = message.get("messageType")
# Message: Displayed answer
if message_type is None:
content = message["adaptiveCards"][0]["body"][0]["text"]
delta_content = content[self.delta_content_pointer :]
logger.line(delta_content, end="")
self.delta_content_pointer = len(content)
# Message: Suggested Questions
if message.get("suggestedResponses"):
logger.note("\nSuggested Questions: ")
suggestion_texts = [
suggestion.get("text")
for suggestion in message.get("suggestedResponses")
]
for suggestion_text in suggestion_texts:
logger.file(f"- {suggestion_text}")
if return_output:
completions_output = self.outputer.output(
delta_content, content_type="Completions"
)
if message.get("suggestedResponses"):
suggestion_texts_str = "\nSuggested Questions:\n"
suggestion_texts_str += "\n".join(
f"- {item}" for item in suggestion_texts
)
suggestions_output = self.outputer.output(
suggestion_texts_str,
content_type="SuggestedResponses",
)
return [completions_output, suggestions_output]
else:
return completions_output
# Message: Search Query
elif message_type in ["InternalSearchQuery"]:
message_hidden_text = message["hiddenText"]
search_str = f"[Searching: [{message_hidden_text}]]"
logger.note(search_str)
if return_output:
return self.outputer.output(
search_str, content_type="InternalSearchQuery"
)
# Message: Internal Search Results
elif message_type in ["InternalSearchResult"]:
analysis_str = f"[Analyzing search results ...]"
logger.note(analysis_str)
if return_output:
return self.outputer.output(
analysis_str, content_type="InternalSearchResult"
)
# Message: Loader status, such as "Generating Answers"
elif message_type in ["InternalLoaderMessage"]:
# logger.note("[Generating answers ...]\n")
pass
# Message: Internal thoughts, such as "I will generate my response to the user message"
elif message_type in ["Internal"]:
pass
# Message: Internal Action Marker, no value
elif message_type in ["InternalActionMarker"]:
continue
# Message: Render Cards for Webpages
elif message_type in ["RenderCardRequest"]:
continue
elif message_type in ["ChatName"]:
continue
# Message: Not Implemented
else:
raise NotImplementedError(
f"Not Supported Message Type: {message_type}"
)
return self.outputer.output(content="", content_type="NotImplemented")
|