forked from poe-platform/server-bot-quick-start
-
Notifications
You must be signed in to change notification settings - Fork 14
/
bot_MeguminWizardEx.py
181 lines (138 loc) Β· 5.87 KB
/
bot_MeguminWizardEx.py
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
"""
BOT_NAME="MeguminWizardEx"; modal deploy --name $BOT_NAME bot_${BOT_NAME}.py; curl -X POST https://api.poe.com/bot/fetch_settings/$BOT_NAME/$POE_ACCESS_KEY
Test message:
Cast an explosion
"""
from __future__ import annotations
import random
import re
import textwrap
from collections import defaultdict
from typing import AsyncIterable
import fastapi_poe.client
from fastapi_poe import PoeBot, make_app
from fastapi_poe.client import MetaMessage, stream_request
from fastapi_poe.types import QueryRequest, SettingsRequest, SettingsResponse
from modal import Image, Stub, asgi_app
from sse_starlette.sse import ServerSentEvent
fastapi_poe.client.MAX_EVENT_COUNT = 10000
# introduction to be added https://i.imgur.com/xbXviUO.gif
EMOJI_INVERSE_MAP = {
"https://i.imgur.com/lPHtYl9.gif": ["π₯"],
"https://i.imgur.com/aJ9Pnas.gif": ["π§ββοΈ"],
"https://i.imgur.com/hgphb9b.gif": ["π±"],
"https://i.imgur.com/3YY02tm.gif": ["π±"],
"https://i.imgur.com/nx8WjtW.gif": ["π€"],
"https://i.imgur.com/h9vDS5V.gif": ["π"],
"https://i.imgur.com/RGlKI4T.gif": ["π€©"],
"https://i.imgur.com/4KeeXni.gif": ["π", "π€΄", "π "],
"https://i.imgur.com/zJKBaIP.gif": ["π"],
"https://i.imgur.com/7PzO0Tk.gif": ["π"],
"https://i.imgur.com/WA2STCk.gif": ["π", "πͺ"],
}
EMOJI_MAP = defaultdict(list)
for image_url, emojis in EMOJI_INVERSE_MAP.items():
for emoji in emojis:
EMOJI_MAP[emoji].append(image_url)
DEFAULT_EMOJIS = set(EMOJI_MAP.keys())
CONVERSATION_SUGGESTED_EMOJIS = defaultdict(lambda: DEFAULT_EMOJIS)
ITALICS_PATTERN = r"(?<!\*)\*([^\*]+)\*(?!\*)|(?<!_)_([^_]+)_(?!_)"
EMOJI_PROMPT_TEMPLATE = """
You will summarize the character reply into one emojis
<user_statement>
{user_statement}
</user_statement>
<character_statement>
{character_statement}
</character_statement>
This are the available emojis
{available_emojis}
""".strip()
MeguminHelper_SYSTEM_PROMPT = """
You are Megumin from KonoSuba. You do not know about AI. You do not talk about AI. You will always stay in character. You do not talk about staying in character.
"""
EmojiClassifier_SYSTEM_PROMPT = """
You will follow the instruction and summarize the presented reply into just one emoji.
""".strip()
def redact_image_links(text):
pattern = r"!\[.*\]\(http.*\)"
redacted_text = re.sub(pattern, "", text)
return redacted_text
class EchoBot(PoeBot):
async def get_response(self, query: QueryRequest) -> AsyncIterable[ServerSentEvent]:
user_statement = query.query[-1].content
print("user_statement", user_statement)
for statement in query.query:
statement.content = redact_image_links(statement.content)
query.query = [
{"role": "system", "content": MeguminHelper_SYSTEM_PROMPT}
] + query.query
character_reply = ""
async for msg in stream_request(query, "ChatGPT", query.api_key):
# Note: See https://poe.com/MeguminHelper for the system prompt
if isinstance(msg, MetaMessage):
continue
elif msg.is_suggested_reply:
yield self.suggested_reply_event(msg.text)
elif msg.is_replace_response:
yield self.replace_response_event(msg.text)
else:
character_reply += msg.text
yield self.replace_response_event(character_reply)
italics_from_character = "\n".join(
"".join(element) for element in re.findall(ITALICS_PATTERN, character_reply)
)
character_statement = character_reply
if italics_from_character:
character_statement = italics_from_character
user_statement = ""
available_emojis = CONVERSATION_SUGGESTED_EMOJIS[query.conversation_id]
print("character_statement", character_statement)
query.query[-1].content = EMOJI_PROMPT_TEMPLATE.format(
user_statement=user_statement,
character_statement=character_statement,
available_emojis=available_emojis,
)
query.query = [{"role": "system", "content": EmojiClassifier_SYSTEM_PROMPT}] + [
query.query[-1]
]
emoji_classification = ""
async for msg in stream_request(query, "ChatGPT", query.api_key):
# Note: See https://poe.com/EmojiClassifier for the system prompt
if isinstance(msg, MetaMessage):
continue
elif msg.is_suggested_reply:
yield self.suggested_reply_event(msg.text)
elif msg.is_replace_response:
yield self.replace_response_event(msg.text)
else:
emoji_classification += msg.text
emoji_classification = emoji_classification.strip()
available_emojis.discard(emoji_classification)
print("emoji_classification", emoji_classification)
image_url_selection = EMOJI_MAP.get(emoji_classification)
print("image_url_selection", image_url_selection)
if image_url_selection:
image_url = random.choice(image_url_selection)
yield self.text_event(f"\n![{emoji_classification}]({image_url})")
async def get_settings(self, setting: SettingsRequest) -> SettingsResponse:
return SettingsResponse(
server_bot_dependencies={"ChatGPT": 2},
allow_attachments=False,
introduction_message=textwrap.dedent(
"""
My name is Megumin! My calling is that of an archwizard, one who controls explosion magic, the strongest of all offensive magic!
![](https://i.imgur.com/xbXviUO.gif)
"""
).strip(),
)
bot = EchoBot()
image = Image.debian_slim().pip_install(
"fastapi-poe==0.0.23", "huggingface-hub==0.16.4"
)
stub = Stub("poe-bot-quickstart")
@stub.function(image=image)
@asgi_app()
def fastapi_app():
app = make_app(bot, allow_without_key=True)
return app