from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import numpy as np import nltk from nltk.stem.lancaster import LancasterStemmer import datetime import time import json nltk.download('punkt_tab') app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], # Permite todos los orígenes (cambiar en producción) allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Inicializar el stemmer stemmer = LancasterStemmer() # Modelo de datos para la solicitud de entrenamiento class TrainingData(BaseModel): sentences: list # Lista de diccionarios con "sentence" y "class" # Variables globales para almacenar el modelo entrenado synapse_0 = None synapse_1 = None words = [] classes = [] # Preprocesamiento de datos def preprocess_data(training_data): global words, classes words = [] classes = [] documents = [] ignore_words = ['?'] for pattern in training_data: w = nltk.word_tokenize(pattern['sentence']) words.extend(w) documents.append((w, pattern['class'])) if pattern['class'] not in classes: classes.append(pattern['class']) words = [stemmer.stem(w.lower()) for w in words if w not in ignore_words] words = list(set(words)) classes = list(set(classes)) # Crear datos de entrenamiento training = [] output = [] output_empty = [0] * len(classes) for doc in documents: bag = [] pattern_words = doc[0] pattern_words = [stemmer.stem(word.lower()) for word in pattern_words] for w in words: bag.append(1) if w in pattern_words else bag.append(0) training.append(bag) output_row = list(output_empty) output_row[classes.index(doc[1])] = 1 output.append(output_row) return np.array(training), np.array(output) # Funciones de la red neuronal def sigmoid(x): return 1 / (1 + np.exp(-x)) def sigmoid_output_to_derivative(output): return output * (1 - output) def train(X, y, hidden_neurons=10, alpha=1, epochs=50000, dropout=False, dropout_percent=0.5): global synapse_0, synapse_1 print("Training with %s neurons, alpha:%s, dropout:%s %s" % (hidden_neurons, str(alpha), dropout, dropout_percent if dropout else '')) print("Input matrix: %sx%s Output matrix: %sx%s" % (len(X), len(X[0]), 1, len(classes))) np.random.seed(1) last_mean_error = 1 synapse_0 = 2 * np.random.random((len(X[0]), hidden_neurons)) - 1 synapse_1 = 2 * np.random.random((hidden_neurons, len(classes))) - 1 prev_synapse_0_weight_update = np.zeros_like(synapse_0) prev_synapse_1_weight_update = np.zeros_like(synapse_1) for j in range(epochs + 1): layer_0 = X layer_1 = sigmoid(np.dot(layer_0, synapse_0)) layer_2 = sigmoid(np.dot(layer_1, synapse_1)) layer_2_error = y - layer_2 if (j % 10000) == 0 and j > 5000: if np.mean(np.abs(layer_2_error)) < last_mean_error: print("delta after " + str(j) + " iterations:" + str(np.mean(np.abs(layer_2_error)))) last_mean_error = np.mean(np.abs(layer_2_error)) else: print("break:", np.mean(np.abs(layer_2_error)), ">", last_mean_error) break layer_2_delta = layer_2_error * sigmoid_output_to_derivative(layer_2) layer_1_error = layer_2_delta.dot(synapse_1.T) layer_1_delta = layer_1_error * sigmoid_output_to_derivative(layer_1) synapse_1_weight_update = (layer_1.T.dot(layer_2_delta)) synapse_0_weight_update = (layer_0.T.dot(layer_1_delta)) synapse_1 += alpha * synapse_1_weight_update synapse_0 += alpha * synapse_0_weight_update prev_synapse_0_weight_update = synapse_0_weight_update prev_synapse_1_weight_update = synapse_1_weight_update # Guardar el modelo entrenado now = datetime.datetime.now() synapse = {'synapse0': synapse_0.tolist(), 'synapse1': synapse_1.tolist(), 'datetime': now.strftime("%Y-%m-%d %H:%M"), 'words': words, 'classes': classes} with open("intent_class.json", "w") as outfile: json.dump(synapse, outfile, indent=4, sort_keys=True) print("Model saved to intent_class.json") # Endpoint para entrenar el modelo @app.post("/train") async def train_model(data: TrainingData): global synapse_0, synapse_1, words, classes training_data = data.sentences # Preprocesar los datos X, y = preprocess_data(training_data) # Entrenar el modelo start_time = time.time() train(X, y, hidden_neurons=20, alpha=0.1, epochs=100000, dropout=False, dropout_percent=0.2) elapsed_time = time.time() - start_time print("Training completed in:", elapsed_time, "seconds") return {"message": "Model trained successfully", "elapsed_time": elapsed_time} # Endpoint para clasificar una frase @app.post("/classify") async def classify_sentence(request: Request): global synapse_0, synapse_1, words, classes data = await request.json() sentence = data.get("sentence") if not sentence: raise HTTPException(status_code=400, detail="No sentence provided") # Clasificar la frase ERROR_THRESHOLD = 0.2 x = bow(sentence.lower(), words) l0 = x l1 = sigmoid(np.dot(l0, synapse_0)) l2 = sigmoid(np.dot(l1, synapse_1)) results = [[i, r] for i, r in enumerate(l2) if r > ERROR_THRESHOLD] results.sort(key=lambda x: x[1], reverse=True) return_results = [[classes[r[0]], r[1]] for r in results] return {"sentence": sentence, "classification": return_results} # Función para crear el bag of words def bow(sentence, words, show_details=False): sentence_words = clean_up_sentence(sentence) bag = [0] * len(words) for s in sentence_words: for i, w in enumerate(words): if w == s: bag[i] = 1 if show_details: print("found in bag: %s" % w) return np.array(bag) # Función para limpiar y tokenizar una frase def clean_up_sentence(sentence): sentence_words = nltk.word_tokenize(sentence) sentence_words = [stemmer.stem(word.lower()) for word in sentence_words] return sentence_words # Cargar el modelo si existe try: with open("intent_class.json", "r") as file: synapse = json.load(file) synapse_0 = np.asarray(synapse['synapse0']) synapse_1 = np.asarray(synapse['synapse1']) words = synapse['words'] classes = synapse['classes'] print("Model loaded from intent_class.json") except FileNotFoundError: print("No pre-trained model found. Please train the model first.") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8500)