app.py

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)