:zap: [Enhance] Prettify format of searching queries and results

networks/message_outputer.py

@@ -1,4 +1,5 @@
 import json
+import re
 
 
 class OpenaiStreamOutputer:
@@ -30,14 +31,7 @@ class OpenaiStreamOutputer:
                     "finish_reason": None,
                 }
             ]
-        elif content_type in [
-            "Completions",
-            "InternalSearchQuery",
-            "InternalSearchResult",
-            "SuggestedResponses",
-        ]:
-            if content_type in ["InternalSearchQuery", "InternalSearchResult"]:
-                content += "\n"
+        elif content_type == "Completions":
             data["choices"] = [
                 {
                     "index": 0,
@@ -45,6 +39,53 @@ class OpenaiStreamOutputer:
                     "finish_reason": None,
                 }
             ]
+        elif content_type == "InternalSearchQuery":
+            search_str = f"Searching: [**{content.strip()}**]\n"
+            data["choices"] = [
+                {
+                    "index": 0,
+                    "delta": {"content": search_str},
+                    "finish_reason": None,
+                }
+            ]
+        elif content_type == "InternalSearchResult":
+            invocation = content["invocation"]
+            web_search_results = content["web_search_results"]
+            matches = re.search('\(query="(.*)"\)', invocation)
+            if matches:
+                search_query = matches.group(1)
+            else:
+                search_query = invocation
+            search_str = f"Searching: [**{search_query.strip()}**]"
+            search_results_str_list = []
+            for idx, search_result in enumerate(web_search_results):
+                search_results_str_list.append(
+                    f"{idx+1}. [{search_result['title']}]({search_result['url']})"
+                )
+            search_results_str = "\n".join(search_results_str_list)
+            search_results_str = (
+                f"<details>\n"
+                f"<summary>\n{search_str}\n</summary>\n"
+                f"{search_results_str}\n"
+                f"</details>\n"
+            )
+            data["choices"] = [
+                {
+                    "index": 0,
+                    "delta": {"content": search_results_str},
+                    "finish_reason": None,
+                }
+            ]
+        elif content_type == "SuggestedResponses":
+            suggestion_texts_str = "\n\n---\n\n**Suggested Questions:**\n"
+            suggestion_texts_str += "\n".join(f"- {item}" for item in content)
+            data["choices"] = [
+                {
+                    "index": 0,
+                    "delta": {"content": suggestion_texts_str},
+                    "finish_reason": None,
+                }
+            ]
         elif content_type == "Finished":
             data["choices"] = [
                 {

networks/message_parser.py

@@ -1,3 +1,4 @@
+import json
 from utils.logger import logger
 from networks import OpenaiStreamOutputer
 
@@ -17,7 +18,8 @@ class MessageParser:
                 message_type = message.get("messageType")
                 # Message: Displayed answer
                 if message_type is None:
-                    content = message["adaptiveCards"][0]["body"][0]["text"]
+                    # 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)
@@ -35,37 +37,41 @@ class MessageParser:
                             delta_content, content_type="Completions"
                         )
                         if message.get("suggestedResponses"):
-                            suggestion_texts_str = (
-                                "\n\n---\n\n**Suggested Questions:**\n"
-                            )
-                            suggestion_texts_str += "\n".join(
-                                f"- {item}" for item in suggestion_texts
-                            )
                             suggestions_output = self.outputer.output(
-                                suggestion_texts_str,
+                                suggestion_texts,
                                 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)
+                    search_query_str = message.get("hiddenText")
                     if return_output:
-                        return self.outputer.output(
-                            search_str, content_type="InternalSearchQuery"
-                        )
+                        # 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"]:
-                    analysis_str = f"[Analyzing search results ...]"
-                    logger.note(analysis_str)
-                    # if return_output:
-                    #     return self.outputer.output(
-                    #         analysis_str, content_type="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")