File size: 6,536 Bytes
6dc7c27 | 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 | from abc import ABC
from typing import List, Tuple, Callable
import numpy as np
import torch
from src.disambiguation_corpora import DisambiguationInstance
from src.sense_inventories import SenseInventory
from src.utils.wsd import pos_map
class DependencyFinder(ABC):
def __init__(self, max_dependencies: int = -1):
self.max_dependencies = max_dependencies
def find_dependencies(
self, disambiguation_context: List[DisambiguationInstance], instance_idx: int
) -> List[DisambiguationInstance]:
dependencies = self._find_dependencies(disambiguation_context, instance_idx)
if self.max_dependencies >= 0:
dependencies = dependencies[: self.max_dependencies]
return dependencies
def _find_dependencies(
self, disambiguation_context: List[DisambiguationInstance], instance_idx: int
) -> List[DisambiguationInstance]:
raise NotImplementedError
class EmptyDependencyFinder(DependencyFinder):
def _find_dependencies(
self, disambiguation_context: List[DisambiguationInstance], instance_idx: int
) -> List[DisambiguationInstance]:
return []
class PolysemyDependencyFinder(DependencyFinder):
def __init__(self, sense_inventory: SenseInventory, max_dependencies: int = -1):
super().__init__(max_dependencies)
self.sense_inventory = sense_inventory
def _find_dependencies(
self, disambiguation_context: List[DisambiguationInstance], instance_idx: int
) -> List[DisambiguationInstance]:
polysemy_ordered_instances = sorted(
[di for di in disambiguation_context if di.instance_id is not None],
key=lambda di: len(self.sense_inventory.get_possible_senses(di.lemma, di.pos)),
)
instances_id = [di.instance_id for di in polysemy_ordered_instances]
current_instance_id = instances_id.index(disambiguation_context[instance_idx].instance_id)
return polysemy_ordered_instances[:current_instance_id]
class PPMIPolysemyDependencyFinder(PolysemyDependencyFinder):
def __init__(
self,
sense_inventory: SenseInventory,
single_counter_path: str,
pair_counter_path: str,
energy: float,
max_dependencies: int = -1,
normalize_ppmi: bool = False,
minimum_ppmi: float = 0.0,
undirected: bool = False,
with_pos: bool = True,
):
super().__init__(sense_inventory, max_dependencies)
self.energy = energy
self.normalize_ppmi = normalize_ppmi
self.minimum_ppmi = minimum_ppmi
self.undirected = undirected
self.with_pos = with_pos
self.ppmi_func = self.setup_ppmi_func(single_counter_path, pair_counter_path)
def setup_ppmi_func(
self,
single_counter_path: str,
pair_counter_path: str,
) -> Callable[[Tuple[str, str], Tuple[str, str]], float]:
def split_lp(lp):
if self.with_pos:
l = lp[: lp.rindex(".")]
p = lp[lp.rindex(".") + 1 :]
p = pos_map.get(p, p)
else:
l = lp
p = "FAKE-POS"
return l, p
# read single counter
single_counter = {}
N = 0
with open(single_counter_path) as f:
for line in f:
lp, c = line.strip().split("\t")
l, p = split_lp(lp)
c = int(float(c))
single_counter[(l, p)] = c
N += c
# read pair counter
pair_counter = {}
N_co = 0
with open(pair_counter_path) as f:
for line in f:
lp1, lp2, c = line.strip().split("\t")
l1, p1 = split_lp(lp1)
l2, p2 = split_lp(lp2)
assert (l1, p1) in single_counter and (l2, p2) in single_counter, f"{(l1, p1)} | {(l2, p2)}"
c = int(float(c))
pair_counter[((l1, p1), (l2, p2))] = c
N_co += c
def f(k1: Tuple[str, str], k2: Tuple[str, str]) -> float:
try:
pxy = (pair_counter[k1, k2] if (k1, k2) in pair_counter else pair_counter[k2, k1]) / N_co
px = single_counter[k1] / N
py = single_counter[k2] / N
sample_pmi = np.log2(pxy / (px * py))
if self.normalize_ppmi:
sample_pmi /= -np.log2(pxy)
except KeyError:
sample_pmi = 0.0
return max(sample_pmi, 0.0)
return f
def score_dependencies(
self, disambiguation_context: List[DisambiguationInstance], instance_idx: int
) -> Tuple[List[DisambiguationInstance], torch.Tensor]:
def di2lp(di: DisambiguationInstance):
if self.with_pos:
return di.lemma, di.pos
else:
return di.text.replace(" ", "_"), "FAKE-POS"
x = disambiguation_context[instance_idx]
if self.undirected:
instance_dependencies = [
dep for i, dep in enumerate(disambiguation_context) if i != instance_idx and dep.instance_id is not None
]
else:
instance_dependencies = super()._find_dependencies(disambiguation_context, instance_idx)
# compute ppmi for each dependency
ppmis = torch.tensor([self.ppmi_func(di2lp(x), di2lp(y)) for y in instance_dependencies])
# threshold on ppmis if minimum_ppmi is set
ppmis[ppmis < self.minimum_ppmi] = 0.0
return instance_dependencies, ppmis
def _find_dependencies(
self, disambiguation_context: List[DisambiguationInstance], instance_idx: int
) -> List[DisambiguationInstance]:
instance_dependencies, ppmis = self.score_dependencies(disambiguation_context, instance_idx)
if all(score == 0.0 for score in ppmis):
return []
# convert to probability
ps = ppmis / ppmis.sum()
# take indices up to self.energy cumulative probability
indices = []
cp = 0.0
for index in ps.argsort(descending=True):
if ps[index] == 0.0:
break
cp += ps[index]
indices.append(index)
if cp > self.energy:
break
return [instance_dependencies[i] for i in indices]
|