Spaces:
Paused
Paused
File size: 4,671 Bytes
0bade77 bb37148 63998ab 6ad10eb 63998ab 6ad10eb 82cb440 6ad10eb 4c9b469 6ad10eb 0bade77 6ad10eb 1bbf682 63998ab 6ad10eb 4c9b469 0fdaf32 4c9b469 82cb440 4c9b469 0fdaf32 0bade77 4c9b469 0fdaf32 6ad10eb 0bade77 4c9b469 0bade77 6ad10eb 0bade77 6ad10eb 1bbf682 82cb440 6ad10eb 82cb440 6ad10eb 4c9b469 b286bf5 |
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 90 91 92 93 94 95 96 97 |
import json
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"]
content = message["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"):
suggestions_output = self.outputer.output(
suggestion_texts,
content_type="SuggestedResponses",
)
return [completions_output, suggestions_output]
else:
return completions_output
# Message: Search Query
elif message_type in ["InternalSearchQuery"]:
search_query_str = message.get("hiddenText")
if return_output:
# output_str = self.outputer.output(
# search_query_str, content_type="InternalSearchQuery"
# )
# logger.note(output_str)
# return output_str
return None
# Message: Internal Search Results
elif message_type in ["InternalSearchResult"]:
if message.get("groundingInfo"):
web_search_results = message.get("groundingInfo").get(
"web_search_results"
)
invocation = message.get("invocation")
if return_output:
search_results_str = self.outputer.output(
{
"invocation": invocation,
"web_search_results": web_search_results,
},
content_type="InternalSearchResult",
)
data = json.loads(search_results_str)
logger.note(data["choices"][0]["delta"]["content"])
return search_results_str
# 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 None
|