forked from poe-platform/server-bot-quick-start
-
Notifications
You must be signed in to change notification settings - Fork 14
/
wrapper_bot.py
95 lines (75 loc) · 2.66 KB
/
wrapper_bot.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
"""
Sample bot that wraps OpenAI async API.
"""
from __future__ import annotations
import os
from typing import AsyncIterable
import fastapi_poe as fp
from modal import App, Image, asgi_app, exit
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def stream_chat_completion(request: fp.QueryRequest):
messages = []
# this is a demo bot, the messages and message length will be truncated
for query in request.query[-5:]:
if query.role == "system":
messages.append({"role": "system", "content": query.content[:50]})
elif query.role == "bot":
messages.append({"role": "assistant", "content": query.content[:50]})
elif query.role == "user":
messages.append({"role": "user", "content": query.content[:50]})
else:
raise
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=1.0,
stream=True,
max_tokens=50,
)
async for chunk in response:
if chunk.choices[0].delta.content:
yield fp.PartialResponse(text=chunk.choices[0].delta.content)
class WrapperBot(fp.PoeBot):
async def get_response(
self, request: fp.QueryRequest
) -> AsyncIterable[fp.PartialResponse]:
async for msg in stream_chat_completion(request):
yield msg
REQUIREMENTS = ["fastapi-poe==0.0.47", "openai"]
image = (
Image.debian_slim()
.pip_install(*REQUIREMENTS)
.env(
{
"POE_ACCESS_KEY": os.environ["POE_ACCESS_KEY"],
"OPENAI_API_KEY": os.environ["OPENAI_API_KEY"],
}
)
)
app = App(name="wrapper-bot-poe", image=image)
@app.cls()
class Model:
# See https://creator.poe.com/docs/quick-start#integrating-with-poe to find these values.
access_key: str | None = os.environ["POE_ACCESS_KEY"]
bot_name: str | None = None # REPLACE WITH YOUR BOT NAME
@exit()
def sync_settings(self):
"""Syncs bot settings on server shutdown."""
if self.bot_name and self.access_key:
try:
fp.sync_bot_settings(self.bot_name, self.access_key)
except Exception:
print("\n*********** Warning ***********")
print(
"Bot settings sync failed. For more information, see: https://creator.poe.com/docs/server-bots-functional-guides#updating-bot-settings"
)
print("\n*********** Warning ***********")
@asgi_app()
def fastapi_app(self):
bot = WrapperBot()
app = fp.make_app(bot, access_key=self.access_key)
return app
@app.local_entrypoint()
def main():
Model().run.remote()