Reto: construye un MEMM en Python

Siiiiiiii!
Aqui esta mi reto cumplido
-> https://colab.research.google.com/drive/1-9BhEEyBldc03h1BQhkSz3Ncwgea1qHZ?usp=sharing
.

también hice el reto, lo pueden ver aqui

Reto terminado. Fue un desafio, la verdad es muy diferente ver como lo hace Francisco a hacerlo uno mismo.

https://github.com/JDGALLEGOS/MEMM_Viterbi/blob/main/MEMM_Viterbi.ipynb

from conllu import parse_incr

uniqueFeatureDict = {}
contextDict = {}

tagtype = 'upos'
data_file = open("UD_Spanish-AnCora/es_ancora-ud-train.conllu", "r", encoding="utf-8")

#Calculando conteos (pre-probabilidades)
for tokenlist in parse_incr(data_file):
  prevtag = "None"
  for token in tokenlist:
    tag = token[tagtype]
    word = token['form'].lower()
    #c(tag|word,prevtag)
    largeKey = tag+'|'+word+','+prevtag
    if largeKey in uniqueFeatureDict.keys():
      uniqueFeatureDict[largeKey]+=1
    else:
      uniqueFeatureDict[largeKey]=1
    key = word+','+prevtag
    if key in contextDict.keys():
      contextDict[key]+=1
    else:
      contextDict[key]=1
    #print(largeKey, key, '\n')
    prevtag=tag

Hola, aquí deje mi solución al reto, al final traté de compactar todo en una clase para el modelo y una función extra para plotear la matriz de Viterbi. El código fue el siguiente:

Importación de librerías y recursos

###### COSAS IMPORTANTES, CORRER AL INICIO

# cargamos el dataset o corpus
!git clone https://github.com/UniversalDependencies/UD_Spanish-AnCora.git
# librerias
!pip3 install conllu
!pip3 install stanza
import numpy as np # para el manejo de la mtriz de Viterbi
import stanza # queremos el tokenizador en español
stanza.download('es') # descargamos el paquete de español

# cosas para plotear la matriz de viterbi
import seaborn as sn # para plotear
import matplotlib.pyplot as plt
import pandas as pd # para organizar la matriz

# funcion para leer el conllu
from conllu import parse_incr

El modelo y la función para leer el corpus

 # todas las etiquetas
tags = {'ADJ','ADP','ADV','AUX','CCONJ','DET','INTJ','NOUN','NUM',
        'PART','PRON','PROPN','PUNCT','SCONJ','SYM','VERB','_'}

# instanciamos el tokenizador que sera de importancia
nlp = stanza.Pipeline('es', processors='tokenize')

class MEMM():

  # en el constructor que se definan los diccionarios
  def __init__(self):
    # diccionarios de conteos
    self.uniqueFeatureDict = {}
    self.contextDict = {}
    # diccionarios de las probabilidades
    self.posteriorProbDict = {}
    self.initialTagProb = {}
    # diccionario de tags con ids
    self.tagsDict = {}

  # ayudante para contar en diccionarios
  def __count_dict(self, dic, key):
    if key in dic.keys():
      # si ya esta le sumamos 1
      dic[key] += 1
    else:
      # si no la iniciamos en 1
      dic[key] = 1
    return dic

  # privado, conteo que se usara en entrenamiento
  def __counts(self, para_list): # paragraph_list*
    # uniqueFeatureDict = {} # C(tag|word,prevtag)
    # contextDict = {} # C(word|prevtag)
    # iterar los "parrafos" del dataset
    for para in para_list:
      prevtag = "None"# al inicio no hay tag previo
      # iterar las palabras del "parrafo"
      for token in para:
        # leemos la palabra y tag del token
        word, tag = token[0], token[1]

        ########## CONTEO PARA uniqueFeatureDict C(tag|word,prevtag)
        # definimos la clave para el diccionario
        ufd_key = tag + '|' + word + ',' + prevtag
        self.uniqueFeatureDict = self.__count_dict(self.uniqueFeatureDict, ufd_key)
        
        ########## CONTEO PARA contextDict C(word|prevtag)
        # definimos la clave para el diccionario
        cd_key = word + ',' + prevtag
        self.contextDict = self.__count_dict(self.contextDict, cd_key)

        # al final el tag será el prevtag
        prevtag = tag

  # probabilidades, igual privada, para el entrenamiento
  def __probs(self):
    # iteramos las claves del ciccionario
    for key in self.uniqueFeatureDict.keys():
      if len(key.split('|'))==3: # cae un caso, es algo raro por lo que vi
        # clave para el contextDict
        cd_key = '|'+key.split('|')[-1]
        # efectuamos el calculo
        self.posteriorProbDict[key] = self.uniqueFeatureDict[key]/self.contextDict[cd_key]
        # el caso que cayo fue [ SYM||,NOUN ], al parecer el dataset contuvo un
        # caso en el que el texto traia un |, y lo toma como palabra
      else:
        # clave para el contextDict
        cd_key = key.split('|')[1]
        # efectuamos el calculo
        self.posteriorProbDict[key] = self.uniqueFeatureDict[key]/self.contextDict[cd_key]
  
  def __initial_count(self, data):
    # hay que enumerarlas para que el modelo prediga segun el numero del tag
    # de modo que el tagDict traera tag: id, el id será la que use le modelo
    for i, tag in enumerate(tags):
      self.tagsDict[tag] = i

    for para in data: # iteramos los parrafos
      # buscamos el tag
      tag = para[0][1]
      # y hacemos el conteo mediante la funcion de contar
      self.__count_dict(self.initialTagProb, tag)
      # ayuda para las probabilidades
    total = len(data)
    # iteramos los conteos
    for key in self.initialTagProb.keys():
      # dividir el conteo entre el total
      self.initialTagProb[key] /= total

  # entrenamiento: conteos y probabilidades
  def fit(self, data):
    # realizamos los conteos
    self.__counts(data)
    # y el calculo de las probabilidades
    self.__probs()
    # sacar las distribuciones iniciales
    self.__initial_count(data)
  
  # plotear la matriz de viterbi
  def __plot_matrix(self, matrix, seq):
    # hay que hacer un dataframe para plotear mejor
    df = pd.DataFrame(matrix, index = self.tagsDict.keys(), columns = list(seq))
    # ploteamos
    plt.figure(figsize = (10,7))
    plt.title("Predicción")
    sn.heatmap(df, annot=True)

  # algoritmo de viterbi para las predicciones, será publico
  def Viterbi(self, string, plot=False):
    if string == '':
      raise ValueError('El mensaje no puede estar vacío')
    # primero hay que tokenizar la palabra
    tok = nlp(string)
    # como queremos solo las palabras hay que hacer esto
    seq = [word.text for word in tok.sentences[0].words]
    # matriz de ceros para comenzar
    viterbiProb = np.zeros((len(self.tagsDict), len(seq)))

    # calculo de la primera columna, la especial
    for tag in self.tagsDict.keys(): 
      # buscamos la llave para el diccionario inicial
      row = self.tagsDict[tag]
      # buscamos la otra probabilidad
      key = tag + '|' + seq[0].lower() + ',' + "None"
      # efectuamos el calculo de la probabilidad el try es por
      # que la palabra puede no corresponder a el tag de la key
      try:
        viterbiProb[row, 0] = self.initialTagProb[tag] * self.posteriorProbDict[key]
      except: # si no pos no
        pass

    # calculos para las siguientes columnas o palabras
    for col in range(1, len(seq)): # iteramos las columnas
      for tag in self.tagsDict.keys(): # iteramos los tags o filas
        tag_row = self.tagsDict[tag] # seleccionamos la fila
        possible_probs = []
        # se hace doble for y possible_probs por que tambien hay que
        # intentar todas las probabilidades previas y de esas ver la
        # mas probable es decir la maxima
        for prevtag in self.tagsDict.keys(): # buscamos el tag previo
          prevtag_row = self.tagsDict[prevtag] # buscamos el row del tag previo
          # key para la probabilidad previa de viterbi
          key = tag + '|' + seq[col].lower() + ',' + prevtag
          try:
            # tratamos de buscar la probabilidad previa
            possible_probs.append(
                viterbiProb[prevtag_row, col-1] * self.posteriorProbDict[key])
          except:
            # si no pos metemos un 0
            possible_probs.append(0)
        # de ahi metemos en la matriz aquella mas probable
        viterbiProb[tag_row, col] = max(possible_probs)

    # Buscamos el resultado de la predicción
    res = []
    for i, p in enumerate(seq): # enumaramos la cadena tokenizada
      for tag in self.tagsDict.keys(): # iteramos los tags
        # vemos si el tag es el argumento maximo de la columna i
        if self.tagsDict[tag] == np.argmax(viterbiProb[:, i]):
          # si es el maximo lo agregamos junto con la palabra
          res.append((p, tag))
    # si nos pidio plotear hacemos el plot
    if plot:
      self.__plot_matrix(viterbiProb, seq)
    return res

  # para predecir hacemos el algoritmo de viterbi
  def predict(self, string, plot=False):
    # usamos el algoritmo de Viterbi y pasamos el plot
    res = self.Viterbi(string, plot)
    return res


# función para leer el dataset
def get_word_list(path, encoding='utf-8', tagtype='upos', alert=True):
  # abrir el archivo
  data_file = open(path, 'r', encoding=encoding)
  word_list = []
  # iteramos las listas de tokens procesando el conllu
  for tokenlist in parse_incr(data_file): # con parse_incr()
      paragraph_list = [] # iteramos por parrafo
      # iteramos buscando la palabra y su tag segun tagtype
      for token in tokenlist:
          # agregamos la palabra y el tag seleccionado
          word = token['form'].lower()
          tag = token[tagtype]
          paragraph_list.append((word, tag))
      word_list.append(paragraph_list) # le agregamos el parrafo
  # si hay alerta ponemos eso
  if alert:
    print(word_list[0])
  return word_list

Usando el modelo y la función

######### Usando a la función
path = "UD_Spanish-AnCora/es_ancora-ud-train.conllu"
data = get_word_list(path, alert=False)
# [(palabra, tag), ...]

# instanciamos el modelo
model = MEMM()
# lo entrenamos
model.fit(data)
# y nos divertimos con el :D
model.predict('El mundo es muy pequeño', plot=True)

Les paso mi código … Aunque no funciona lo estructure como una clase y cree un método train el cual toma el dataset y va llamando a las distintas funciones de manera secuencial.

class MEMM():

  def __init__(self):
    """
    """    
    
    self.posteriorProbDict = {}
    self.initTagStateProb = {}

    self.uniqueFeatureDict = {}
    self.contextDict = {}

    self.tagStateDict = {}

  def train(self, data_file):
    """
    """

    self.data_file = data_file

    self.makeAndCountTags()
    self.posterior()
    self.initial()

    self.makeStates()

  def decode(self, string):
    """
    Algoritmo de vitebi.
    """

    doc = nlp(string)

    if len(doc.sentences) > 1:
      raise ValueError(':(')

    seq = [ word.text for word in doc.sentences[0].words ]
    viterbiProb = np.zeros( (17, len(seq)) )

    # inicilizamos la primera columna.
    for tag in self.tagStateDict.keys():
      tag_row = self.tagStateDict[tag]
      key = tag + '|' + seq[0].lower() + ',' + 'None'

      try:
        viterbiProb[tag_row, 0] = self.initTagStateProb[tag] * self.posteriorProbDict[key]
      except:
        pass

    # computo las siguientes columnas.
    for col in range( 1, len(seq) ):
      for tag in self.tagStateDict.keys():
        tag_row = self.tagStateDict[tag]
        possible_probs = []

        for prevtag in self.tagStateDict.keys(): 
          prevtag_row = self.tagStateDict[prevtag]
          key = tag + '|' + seq[col].lower() + ',' + prevtag

          try:
            possible_probs.append(
                viterbiProb[prevtag_row, col-1] * self.posteriorProbDict[key]
            )
          except:
            possible_probs.append(0)

          viterbiProb[tag_row, col] = max(possible_probs)

    # construccion de secuencia de tags
    res = []
    for i, p in enumerate(seq):
      for tag in self.tagStateDict.keys():
        if self.tagStateDict[tag] == np.argmax( viterbiProb[:, i] ):
          res.append( (p, tag) )

    return res

  def makeStates(self):
    """
    """

    # identificamos las categorias gramaticales 'upos' unicas en el corpus
    stateSet = {'ADJ',
                'ADP',
                'ADV',
                'AUX',    
                'CCONJ',
                'DET',
                'INTJ',
                'NOUN',
                'NUM',
                'PART',
                'PRON',
                'PROPN',
                'PUNCT',
                'SCONJ',
                'SYM',
                'VERB',
                '_'
    }

    # enumeramos las categorias con numeros para asignar a 
    # las columnas de la matriz de Viterbi
    
    for i, state in enumerate(stateSet):
      self.tagStateDict[state] = i

  def makeAndCountTags(self):
    """
    """

    for tokenlist in parse_incr(self.data_file):
      prevtag = 'None'
      for token in tokenlist:
        tag = token[tagtype]
        word = token['form'].lower()

        largekey = tag + '|' + word + ','
        if largekey in self.uniqueFeatureDict.keys():
          self.uniqueFeatureDict[largekey] += 1
        else:
          self.uniqueFeatureDict[largekey] = 1

        key = word + ',' + prevtag
        if key in self.contextDict.keys():
          self.contextDict[key] += 1
        else:
          self.contextDict[key] = 1

        prevtag = tag

  def posterior(self):
    """
    """

    for key in self.uniqueFeatureDict.keys():
      if len( key.split('|') ) == 3:
        self.posteriorProbDict[key] = self.uniqueFeatureDict[key] / self.contextDict['|'+key.split('|')[-1]]
      else:
        self.posteriorProbDict[key] = self.uniqueFeatureDict[key] / self.contextDict[key.split('|')[1]]

  def initial(self):
    """
    """

    wordlist = []
    count = 0

    for tokenlist in parse_incr(self.data_file):
      count += 1
      tag = tokenlist[0]['upos']

      if tag in self.initTagStateProb.keys():
        self.initTagStateProb[tag] += 1
      else:
        self.initTagStateProb[tag] = 1

    for key in self.initTagStateProb.keys():
      self.initTagStateProb[key] /= count

Aquí llamamos a la clase y al entrenamiento.

model = MEMM()

model.train(data_file)

model.decode('el mundo es pequeño')

Donde el resultado es:

[('el', 'NOUN'), ('mundo', 'NOUN'), ('es', 'NOUN'), ('pequeño', 'NOUN')]

Es aquí donde me genera la falla, creo que el error está en las funciones que calculan posterior y initial.