Introducción y Configuración del Entorno
El análisis de sentimiento en idioma chino presenta desafíos únicos debido a la falta de espacios entre palabras y la complejidad semántica. Para abordar este problema, se puede utilizar una arquitectura de Red Neuronal Recurrente, específicamente una Red de Memoria a Corto y Largo Plazo (LSTM), implementada con TensorFlow. A continuación, se detalla el flujo de trabajo completo, desde la preparación de los embeddings hasta el entrenamiento y la inferencia del modelo.
Las bibliotecas fundamentales requeridas para este pipeline son:
tensorflowykeraspara la construcción y entrenamiento de la red neuronal.gensimpara la gestión de vectores de palabras preentrenados.jiebapara la segmentación de texto en chino.numpy,pandasymatplotlibpara la manipulación de datos y visualización.scikit-learnpara la división de conjuntos de datos y métricas.
- Importación de Dependencias
import os
import re
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import jieba
import tensorflow as tf
from gensim.models import KeyedVectors
from sklearn.model_selection import train_test_split
from tensorflow.keras.preprocessing.sequence import pad_sequences
- Carga de Vectores de Palabras Preentrenados
Para capturar el significado semántico, utilizaremos vectores de palabras preentrenados. Se recomienda descargar un modelo de código abierto (como los proporcionados por laboratorios de investigación en procesamiento de lenguaje natural chino) y almacenarlo localmente. En este caso, cargaremos un modelo binario o de texto utilizando la API actualizada de Gensim.
# Ruta al archivo de vectores preentrenados descargado localmente
VECTOR_PATH = 'resources/sgns.zhihu.bigram'
# Carga del modelo utilizando la API moderna de Gensim (versión 4.x+)
word_embeddings = KeyedVectors.load_word2vec_format(
VECTOR_PATH,
binary=False,
unicode_errors='ignore'
)
print(f"Vocabulario cargado: {len(word_embeddings.key_to_index)} palabras.")
- Ingesta y Etiquetado del Corpus
El conjunto de datos consta de reseñas de hoteles divididas en categorías positivas y negativas. Se encapsulará la lectura de archivos en una función para mantener el código limpio y reutilizable. Las etiquetas se generarán como 1 para positivo y 0 para negativo, followed by a randomized shuffle.
def load_corpus(data_dir):
texts = []
labels = []
# Procesar reseñas positivas
pos_dir = os.path.join(data_dir, 'positive')
for filename in os.listdir(pos_dir):
with open(os.path.join(pos_dir, filename), 'r', encoding='utf-8') as f:
texts.append(f.read().strip())
labels.append(1)
# Procesar reseñas negativas
neg_dir = os.path.join(data_dir, 'negative')
for filename in os.listdir(neg_dir):
with open(os.path.join(neg_dir, filename), 'r', encoding='utf-8') as f:
texts.append(f.read().strip())
labels.append(0)
return np.array(texts), np.array(labels)
# Cargar datos desde el directorio local
corpus_texts, sentiment_labels = load_corpus('dataset/hotel_reviews')
# Mezclar los datos de forma determinista
np.random.seed(42)
indices = np.arange(len(corpus_texts))
np.random.shuffle(indices)
corpus_texts = corpus_texts[indices]
sentiment_labels = sentiment_labels[indices]
- Tokenización y Conversión a Índices
El texto en chino debe segmentarse utilizando jieba. Cada palabra resultante se mapea a su índice correspondiente en el modelo de embeddings. Las palabras fuera del vocabulario (OOV) se asignarán al índice 0.
def tokenize_and_encode(texts, embeddings_model):
encoded_sequences = []
for text in texts:
# Segmentación de palabras
words = list(jieba.cut(text))
# Mapeo a índices usando la API key_to_index
encoded = [embeddings_model.key_to_index.get(w, 0) for w in words]
encoded_sequences.append(encoded)
return encoded_sequences
tokenized_corpus = tokenize_and_encode(corpus_texts, word_embeddings)
- Análisis de Longitud de Secuencias y Padding
Las redes LSTM requieren entradas de longitud fija. En lugar de usar la longitud máxima absoluta (lo que desperdiciaría recursos computacionales), calcularemos el percentil 95 de las longitudes de las secuencias para cubrir la gran mayoría de los datos sin outliers extremos.
# Calcular longitudes
sequence_lengths = [len(seq) for seq in tokenized_corpus]
# Visualización de la distribución
plt.figure(figsize=(10, 5))
plt.hist(sequence_lengths, bins=50, color='skyblue', edgecolor='black')
plt.title('Distribución de Longitudes de Secuencias')
plt.xlabel('Número de Tokens')
plt.ylabel('Frecuencia')
plt.show()
# Determinar longitud máxima basada en el percentil 95
MAX_SEQ_LEN = int(np.percentile(sequence_lengths, 95))
print(f"Longitud máxima de secuencia seleccionada (95%): {MAX_SEQ_LEN}")
# Aplicar padding y truncamiento (pre-padding es preferible para LSTMs)
padded_sequences = pad_sequences(
tokenized_corpus,
maxlen=MAX_SEQ_LEN,
padding='pre',
truncating='pre',
value=0
)
- Construcción de la Matriz de Embeddings
Para optimizar la memoria y el tiempo de entrenamiento, limitaremos el vocabulario a las 50,000 palabras más frecuentes. Extraeremos los vectores correspondientes para inicializar la capa de Embedding de Keras.
VOCAB_SIZE = 50000
EMBEDDING_DIM = 300
# Inicializar matriz con ceros
embedding_matrix = np.zeros((VOCAB_SIZE, EMBEDDING_DIM))
# Llenar la matriz con los vectores preentrenados
# index_to_key es la API moderna para index2word
for i in range(min(VOCAB_SIZE, len(word_embeddings.index_to_key))):
word = word_embeddings.index_to_key[i]
embedding_matrix[i] = word_embeddings[word]
# Asegurar que los índices fuera del vocabulario (0) y los que exceden VOCAB_SIZE se manejen
padded_sequences = np.where(padded_sequences >= VOCAB_SIZE, 0, padded_sequences)
- División del Conjunto de Datos
# Dividir en entrenamiento (90%) y prueba (10%)
X_train, X_test, y_train, y_test = train_test_split(
padded_sequences,
sentiment_labels,
test_size=0.10,
random_state=42,
stratify=sentiment_labels
)
- Arquitectura del Modelo LSTM
Se diseñará una red secuencial que comienza con una capa de Embedding preentrenada (congelada para preservar la semántica original), seguida de capas LSTM bidireccionales para capturar contextos pasados y futuros, y capas densas para la clasificación final. Se añaden capas de Dropout para regularizar y prevenir el sobreajuste.
def build_sentiment_model(vocab_size, embedding_dim, max_len, embedding_weights):
model = tf.keras.Sequential([
# Capa de Embedding con pesos preentrenados
tf.keras.layers.Embedding(
input_dim=vocab_size,
output_dim=embedding_dim,
input_length=max_len,
weights=[embedding_weights],
trainable=False, # Congelar embeddings
name='embedding_layer'
),
# Primera capa LSTM Bidireccional
tf.keras.layers.Bidirectional(
tf.keras.layers.LSTM(64, return_sequences=True, dropout=0.2),
name='bi_lstm_1'
),
# Segunda capa LSTM Bidireccional
tf.keras.layers.Bidirectional(
tf.keras.layers.LSTM(32, dropout=0.2),
name='bi_lstm_2'
),
# Capas densas para clasificación
tf.keras.layers.Dense(64, activation='relu', kernel_regularizer=tf.keras.regularizers.l2(0.01)),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Dense(2, activation='softmax', name='output_layer')
])
model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
loss='sparse_categorical_crossentropy',
metrics=['sparse_categorical_accuracy']
)
return model
model = build_sentiment_model(VOCAB_SIZE, EMBEDDING_DIM, MAX_SEQ_LEN, embedding_matrix)
model.summary()
- Entrenamiento y Evaluación
# Configurar EarlyStopping para evitar sobreajuste
early_stop = tf.keras.callbacks.EarlyStopping(
monitor='val_loss',
patience=3,
restore_best_weights=True
)
# Entrenar el modelo
history = model.fit(
X_train, y_train,
epochs=15,
batch_size=64,
validation_split=0.1,
callbacks=[early_stop],
verbose=1
)
# Evaluar en el conjunto de prueba
test_loss, test_accuracy = model.evaluate(X_test, y_test, verbose=0)
print(f"Precisión en el conjunto de prueba: {test_accuracy:.4f}")
- Visualización de Métricas
def plot_training_history(hist):
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))
# Pérdida
ax1.plot(hist.history['loss'], label='Pérdida de Entrenamiento')
ax1.plot(hist.history['val_loss'], label='Pérdida de Validación')
ax1.set_title('Evolución de la Pérdida')
ax1.set_xlabel('Épocas')
ax1.set_ylabel('Loss')
ax1.legend()
# Precisión
ax2.plot(hist.history['sparse_categorical_accuracy'], label='Precisión de Entrenamiento')
ax2.plot(hist.history['val_sparse_categorical_accuracy'], label='Precisión de Validación')
ax2.set_title('Evolución de la Precisión')
ax2.set_xlabel('Épocas')
ax2.set_ylabel('Accuracy')
ax2.legend()
plt.tight_layout()
plt.show()
plot_training_history(history)
- Función de Inferencia para Nuevos Textos
Finalmente, se implementa una función para predecir el sentimiento de texto no visto, replicando exactamente los pasos de preprocesamiento aplicados al conjunto de entrenamiento.
def decode_sequence(indices, embeddings_model):
# Función auxiliar para depuración
return "".join([embeddings_model.index_to_key[i] for i in indices if i != 0])
def predict_new_sentiment(text, model, embeddings_model, max_len, vocab_size):
print(f"Texto de entrada: {text}")
# 1. Segmentación
words = list(jieba.cut(text))
# 2. Codificación
encoded = [embeddings_model.key_to_index.get(w, 0) for w in words]
# 3. Padding
padded = pad_sequences([encoded], maxlen=max_len, padding='pre', truncating='pre', value=0)
# 4. Recorte de vocabulario
padded = np.where(padded >= vocab_size, 0, padded)
# 5. Predicción
prediction = model.predict(padded, verbose=0)[0]
# 6. Interpretación
sentiment = "Positivo" if prediction[1] > prediction[0] else "Negativo"
confidence = max(prediction) * 100
print(f"Sentimiento predicho: {sentiment} (Confianza: {confidence:.2f}%)\n")
return sentiment
# Pruebas con textos de ejemplo
test_reviews = [
"La habitación estaba muy limpia y el personal fue extremadamente amable.",
"El aire acondicionado estaba roto y hacía mucho calor, una experiencia terrible.",
"La cama era muy cómoda, pero el ruido de la calle no me dejó dormir.",
"Excelente relación calidad-precio, definitivamente volveré a hospedarme aquí."
]
for review in test_reviews:
predict_new_sentiment(review, model, word_embeddings, MAX_SEQ_LEN, VOCAB_SIZE)