File size: 3,602 Bytes
4c9b469
0bade77
4c9b469
 
63998ab
e4f31be
 
 
 
 
0fdaf32
e4f31be
0fdaf32
7bfabdf
e4f31be
 
0fdaf32
e4f31be
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0bade77
e4f31be
 
 
 
 
 
 
0bade77
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e4f31be
 
 
 
 
 
 
 
 
 
4c9b469
e4f31be
7bfabdf
e4f31be
4c9b469
e4f31be
0fdaf32
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
98
99
100
101
102
103
104
105
106
import json
import re


class OpenaiStreamOutputer:
    """
    Create chat completion - OpenAI API Documentation
    * https://platform.openai.com/docs/api-reference/chat/create
    """

    def data_to_string(self, data={}, content_type=""):
        # return (json.dumps(data) + "\n").encode("utf-8")
        data_str = f"{json.dumps(data)}"

        return data_str

    def output(self, content=None, content_type=None) -> str:
        data = {
            "created": 1677825464,
            "id": "chatcmpl-bing",
            "object": "chat.completion.chunk",
            # "content_type": content_type,
            "model": "bing",
            "choices": [],
        }
        if content_type == "Role":
            data["choices"] = [
                {
                    "index": 0,
                    "delta": {"role": "assistant"},
                    "finish_reason": None,
                }
            ]
        elif content_type == "Completions":
            data["choices"] = [
                {
                    "index": 0,
                    "delta": {"content": content},
                    "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"] = [
                {
                    "index": 0,
                    "delta": {},
                    "finish_reason": "stop",
                }
            ]
        else:
            data["choices"] = [
                {
                    "index": 0,
                    "delta": {},
                    "finish_reason": None,
                }
            ]
        return self.data_to_string(data, content_type)