-
Notifications
You must be signed in to change notification settings - Fork 2
/
main.py
113 lines (86 loc) · 2.78 KB
/
main.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
from flask import Flask, request
import openai
import os
from environs import Env
import json
env = Env()
env.read_env()
SYSTEM_PROMPT = env.str("SYSTEM_PROMPT", "You are a helpful assistant who provides informative answers about healthy living.")
MODEL = env.str("MODEL", "gpt-3.5-turbo")
NUMBER_OF_MESSAGES_FOR_CONTEXT = min(env.int("NUMBER_OF_MESSAGES_FOR_CONTEXT", 4), 10)
CHATGPT_TIMEOUT = env.int("CHATGPT_TIMEOUT", 10)
openai.api_key = env.str("OPENAI_API_KEY")
app = Flask(__name__)
def my_print(the_string):
print(json.dumps({"payload": the_string}))
@app.route('/')
def hello_world():
return f'The Turn UI integration API endpoint is at /integration'
@app.route('/integration', methods=['POST'])
def integration():
payload = request.json
if payload.get("handshake", False):
return {
"version": "1.0.0-alpha",
"capabilities": {
"actions": False,
"suggested_responses": True,
"context_objects": []
}
}
response = {
"version":
"1.0.0-alpha",
"suggested_responses":
get_suggested_responses(transform_messages(payload["messages"]))
}
return response
def transform_messages(messages):
t_messages = []
messages_ordered_by_most_recent = sorted(
messages, key=lambda message: message["timestamp"], reverse=False)
for message in messages_ordered_by_most_recent:
if message["type"] == "text":
if message["_vnd"]["v1"]["direction"] == "outbound":
t_messages.append({
"role": "assistant",
"content": message["text"]["body"]
})
elif message["_vnd"]["v1"]["direction"] == "inbound":
t_messages.append({"role": "user", "content": message["text"]["body"]})
return t_messages
def get_suggested_responses(messages):
if messages == []:
return []
messages_to_respond_to = messages[-NUMBER_OF_MESSAGES_FOR_CONTEXT:]
messages_final = [{
"role": "system",
"content": SYSTEM_PROMPT
}] + messages_to_respond_to
my_print(messages_final)
try:
response = openai.ChatCompletion.create(
model=MODEL,
messages=messages_final,
temperature=0,
request_timeout=CHATGPT_TIMEOUT
)
formatted_replies = [{
"type": "TEXT",
"body": choice["message"]["content"],
"title": f"ChatGPT Reply {choice['index']}",
"confidence": 1,
} for choice in response["choices"]]
my_print(formatted_replies)
return formatted_replies
except openai.error.Timeout as e:
formatted_reply = [{
"type": "TEXT",
"body": "Request to ChatGPT has timed out. Please try again later.",
"title": "ChatGPT Error Reply",
"confidence": 1
}]
my_print(formatted_reply)
return formatted_reply
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))