-
Notifications
You must be signed in to change notification settings - Fork 0
/
FlaskApi.py
203 lines (175 loc) · 8.31 KB
/
FlaskApi.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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
'''
@Author: Borja Otero Ferreira
'''
from flask import Flask, render_template, request,jsonify,json,Response
from flask_socketio import SocketIO
import os, signal
from Assistant import Assistant
import logging
class IASuiteApi:
def __init__(self):
self.app = Flask(__name__, static_url_path='/static')
self.socketio = SocketIO(self.app, async_mode='threading')
self.logging_setup()
self.default_model_path = "models/llama/llama-2-7b-chat.Q8_0.gguf"
self.default_chat_format = "llama-2"
self.assistant = None
self.setup_routes()
def logging_setup(self):
logging.basicConfig(filename='logs/flask_log.log', level=logging.ERROR)
def setup_routes(self):
self.app.before_request(self.before_first_request)
self.app.route('/')(self.index)
self.socketio.on_event('user_input', self.handle_user_input_route, namespace='/test')
self.app.route('/crear_historial', methods=['POST'])(self.crear_historial)
self.app.route('/actualizar_historial', methods=['POST'])(self.actualizar_historial)
self.app.route('/eliminar_historial', methods=['DELETE'])(self.eliminar_historial)
self.app.route('/recuperar_historial', methods=['GET'])(self.recuperar_historial)
self.app.route('/user_input', methods=['POST'])(self.handle_user_input_route)
self.app.route('/load_model', methods=['POST'])(self.load_model)
self.app.route('/unload_model', methods=['POST'])(self.unload_model)
self.app.route('/stop_response', methods=['POST'])(self.stop_response)
self.app.route('/v1/chat/completions', methods=['POST'])(self.ollama)
def ollama(self):
request_data = request.json # Obtener los datos JSON del cuerpo de la solicitud
user_input = request_data.get('content')
user_input.pop(0) # Elimina el mensaje del sistema
self.assistant.emit_ollama_response_stream(user_input,self.socketio)
print(f'\n\nInput Usuario: {user_input}\n\n')
return 'Response finished'
def before_first_request(self):
if self.assistant is None:
self.assistant = Assistant(
default_model_path=self.default_model_path,
default_chat_format=self.default_chat_format
)
def index(self):
models_list = self.get_models_list("models")
format_list = self.get_format_list()
chat_list = self.get_chat_list()
return render_template('index.html', models_list=models_list, format_list=format_list, chat_list=chat_list)
def crear_historial(self):
nombre_chat = request.form.get('nombre_chat')
historial = request.json.get('historial')
ruta_archivo = os.path.join('chats', f'{nombre_chat}.json')
with open(ruta_archivo, 'w') as f:
json.dump(historial, f, indent=4)
return jsonify({'message': f'Historial {nombre_chat} creado exitosamente.'}), 200
def actualizar_historial(self):
nombre_chat = request.json.get('nombre_chat')
nombre_chat = nombre_chat.replace('/', '-')
nombre_chat = nombre_chat.replace(':', '-')
nombre_chat = nombre_chat.replace(' ', '_')
print(nombre_chat)
historial = request.json.get('historial')
ruta_archivo = os.path.join('chats', f'{nombre_chat}.json')
# Verificar si el archivo existe
if not os.path.exists(ruta_archivo):
# Si el archivo no existe, crear uno nuevo con el historial
with open(ruta_archivo, 'w') as f:
json.dump(historial, f, indent=4)
return jsonify({'message': f'Historial {nombre_chat} creado exitosamente.'}), 201
else:
# Si el archivo ya existe, actualizar el historial
with open(ruta_archivo, 'w') as f:
json.dump(historial, f, indent=4)
return jsonify({'message': f'Historial {nombre_chat} actualizado exitosamente.'}), 200
def eliminar_historial(self):
nombre_chat = request.args.get('nombre_chat')
ruta_archivo = os.path.join('chats', f'{nombre_chat}.json')
if os.path.exists(ruta_archivo):
os.remove(ruta_archivo)
return jsonify({'message': f'Historial {nombre_chat} eliminado exitosamente.'}), 200
else:
return jsonify({'error': f'No se encontró el historial {nombre_chat}.'}), 404
def recuperar_historial(self):
nombre_chat = request.args.get('nombre_chat')
ruta_archivo = os.path.join('chats', f'{nombre_chat}.json')
if os.path.exists(ruta_archivo):
with open(ruta_archivo, 'r') as f:
historial = json.load(f)
return jsonify(historial), 200
else:
return jsonify({'error': f'No se encontró el historial {nombre_chat}.'}), 404
def handle_user_input_route(self):
request_data = request.json # Obtener los datos JSON del cuerpo de la solicitud
chat_history = request_data.get('content')
print("Usuario dijo:", chat_history)
self.assistant.add_user_input(chat_history,self.socketio)
return 'Response finished! 📩'
def load_model(self):
selected_model = request.form.get('model_path')
selected_format = request.form.get('format')
n_gpu_layers = int(request.form.get('gpu_layers')) if request.form.get('gpu_layers') != '' else -1
system_message = request.form.get('system_message')
temperature = float(request.form.get('temperature')) if request.form.get('temperature') != '' else 0.81
max_response_tokens = request.form.get('max_response_tokens')
n_ctx = int(request.form.get('context')) if request.form.get('context') != '' else 2048
self.assistant.unload_model()
self.assistant.load_model(selected_model,selected_format,temperature,n_gpu_layers,system_message,n_ctx,max_response_tokens)
return f'''
\nModel:{selected_model}
\nformat: {selected_format}
\ntemp: {temperature}
\nlayers: {n_gpu_layers}
\nSM: {system_message}
\nctx: {n_ctx}
\nmax tokens: {n_ctx}
'''
def unload_model(self):
self.assistant.unload_model()
return 'Model uninstalled 🫗!'
def stop_response(self):
self.assistant.stop_response()
return "Finishing response...\nPlease wait ⏳"
def get_models_list(self, folder_path):
models_list = []
for root, dirs, files in os.walk(folder_path):
for file in files:
if file.endswith(".gguf"):
model_path = os.path.join(root, file)
models_list.append(model_path)
return models_list
def get_format_list(self):
format_list = [ "llama-2",
"Custom-IALab",
"functionary",
"airoboros",
"guanaco",
"mistral-24",
"gemma",
"saiga",
"openchat",
"chatglm3",
"mistral-instruct",
"chatml",
"pygmalion",
"zephyr",
"mistrallite",
"open-orca",
"intel",
"phind",
"snoozy",
"redpajama-incite",
"openbuddy",
"baichuan",
"vicuna",
"qwen",
"baichuan-2",
"oasst_llama",
"rarete"]
return format_list
def get_chat_list(self):
nombres_archivos_json = []
# Obtener la lista de archivos en el directorio
archivos = os.listdir('chats')
# Filtrar solo los archivos con extensión .json
for archivo in archivos:
if archivo.endswith('.json'):
nombres_archivos_json.append(archivo.replace('.json',''))
# Ordenar la lista en orden inverso
nombres_archivos_json.sort(reverse=True)
return nombres_archivos_json
def stop_server(self):
os.kill(os.getpid(), signal.SIGINT)
return 'Server shutting down...'