File size: 3,650 Bytes
008ad55 5afb88c 008ad55 5afb88c 008ad55 5afb88c 008ad55 6050407 008ad55 5afb88c 008ad55 79237e7 008ad55 | 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 | from flask import Flask, request, jsonify
# from dotenv import load_dotenv
import requests
# import os
app = Flask(__name__)
# load_dotenv()
VOICEFLOW_URL = "https://general-runtime.voiceflow.com/state/user/userID/interact?logs=off"
# VOICEFLOW_API_KEY = os.getenv("VOICEFLOW_API_KEY")
HEADERS = {
"accept": "application/json",
"content-type": "application/json",
"Authorization": "VF.DM.665a0547ee998f43239ef780.HvLoducHeGFxik5M"
}
CONFIG = {
"tts": False,
"stripSSML": True,
"stopAll": True,
"excludeTypes": ["block", "debug", "flow"]
}
def get_nested_data(data, keys, default=None):
for key in keys:
try:
if isinstance(data, list):
data = data[key] if key < len(data) else default
else:
data = data.get(key, default)
except (AttributeError, IndexError, TypeError):
return default
if data is default:
break
return data
def interact_with_voiceflow(action_type, payload=None):
payload = {
"action": {"type": action_type, "payload": payload},
"config": CONFIG
}
response = requests.post(VOICEFLOW_URL, json=payload, headers=HEADERS)
response.raise_for_status() # Will raise an error for bad status codes
return response.json()
def process_response(response_data):
messages = set() # Use a set to avoid duplicate messages
for item in response_data:
if item['type'] == 'text':
messages.add(item['payload'].get('message', '').replace('\n', ' ').replace('**', ''))
for content_item in item['payload'].get('slate', {}).get('content', []):
for child in content_item.get('children', []):
text = child.get('text', '')
if text:
messages.add(text)
return list(messages) # Convert the set back to a list
@app.route('/')
def hello_world():
return 'Hello, World!'
@app.route('/start', methods=['POST'])
def launch():
if not request.is_json:
return jsonify({"error": "Request must be JSON"}), 400
data = request.get_json()
tool_call_id = get_nested_data(data, ["message", "toolWithToolCallList", 0, "toolCall", "id"])
user_question = get_nested_data(data, ["message", "toolWithToolCallList", 0, "toolCall", "function", "arguments", "query"])
if user_question is None:
return jsonify({"error": "Question not found in request"}), 400
print('STARTING VF...')
print('START QUESTION:', user_question)
response_data = interact_with_voiceflow("launch")
messages = process_response(response_data)
print('START VF RESPONSE:', messages)
return jsonify({"results": [{"toolCallId": tool_call_id, "result": messages}]})
@app.route('/query', methods=['POST'])
def api_call():
if not request.is_json:
return jsonify({"error": "Request must be JSON"}), 400
data = request.get_json()
tool_call_id = get_nested_data(data, ["message", "toolWithToolCallList", 0, "toolCall", "id"])
user_question = get_nested_data(data, ["message", "toolWithToolCallList", 0, "toolCall", "function", "arguments", "query"])
if user_question is None:
return jsonify({"error": "Question not found in request"}), 400
print('QUERY TO VF: ', user_question)
response_data = interact_with_voiceflow("text", user_question)
messages = process_response(response_data)
print('QUERY VF RESPONSE:', messages)
return jsonify({"results": [{"toolCallId": tool_call_id, "result": messages}]})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=7860, threaded=True)
|