-
Notifications
You must be signed in to change notification settings - Fork 2.8k
/
tnt.py
executable file
·576 lines (442 loc) · 17.4 KB
/
tnt.py
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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
# Natural Language Toolkit: TnT Tagger
#
# Copyright (C) 2001-2023 NLTK Project
# Author: Sam Huston <sjh900@gmail.com>
#
# URL: <https://www.nltk.org/>
# For license information, see LICENSE.TXT
"""
Implementation of 'TnT - A Statisical Part of Speech Tagger'
by Thorsten Brants
https://aclanthology.org/A00-1031.pdf
"""
from math import log
from operator import itemgetter
from nltk.probability import ConditionalFreqDist, FreqDist
from nltk.tag.api import TaggerI
class TnT(TaggerI):
"""
TnT - Statistical POS tagger
IMPORTANT NOTES:
* DOES NOT AUTOMATICALLY DEAL WITH UNSEEN WORDS
- It is possible to provide an untrained POS tagger to
create tags for unknown words, see __init__ function
* SHOULD BE USED WITH SENTENCE-DELIMITED INPUT
- Due to the nature of this tagger, it works best when
trained over sentence delimited input.
- However it still produces good results if the training
data and testing data are separated on all punctuation eg: [,.?!]
- Input for training is expected to be a list of sentences
where each sentence is a list of (word, tag) tuples
- Input for tag function is a single sentence
Input for tagdata function is a list of sentences
Output is of a similar form
* Function provided to process text that is unsegmented
- Please see basic_sent_chop()
TnT uses a second order Markov model to produce tags for
a sequence of input, specifically:
argmax [Proj(P(t_i|t_i-1,t_i-2)P(w_i|t_i))] P(t_T+1 | t_T)
IE: the maximum projection of a set of probabilities
The set of possible tags for a given word is derived
from the training data. It is the set of all tags
that exact word has been assigned.
To speed up and get more precision, we can use log addition
to instead multiplication, specifically:
argmax [Sigma(log(P(t_i|t_i-1,t_i-2))+log(P(w_i|t_i)))] +
log(P(t_T+1|t_T))
The probability of a tag for a given word is the linear
interpolation of 3 markov models; a zero-order, first-order,
and a second order model.
P(t_i| t_i-1, t_i-2) = l1*P(t_i) + l2*P(t_i| t_i-1) +
l3*P(t_i| t_i-1, t_i-2)
A beam search is used to limit the memory usage of the algorithm.
The degree of the beam can be changed using N in the initialization.
N represents the maximum number of possible solutions to maintain
while tagging.
It is possible to differentiate the tags which are assigned to
capitalized words. However this does not result in a significant
gain in the accuracy of the results.
"""
def __init__(self, unk=None, Trained=False, N=1000, C=False):
"""
Construct a TnT statistical tagger. Tagger must be trained
before being used to tag input.
:param unk: instance of a POS tagger, conforms to TaggerI
:type unk: TaggerI
:param Trained: Indication that the POS tagger is trained or not
:type Trained: bool
:param N: Beam search degree (see above)
:type N: int
:param C: Capitalization flag
:type C: bool
Initializer, creates frequency distributions to be used
for tagging
_lx values represent the portion of the tri/bi/uni taggers
to be used to calculate the probability
N value is the number of possible solutions to maintain
while tagging. A good value for this is 1000
C is a boolean value which specifies to use or
not use the Capitalization of the word as additional
information for tagging.
NOTE: using capitalization may not increase the accuracy
of the tagger
"""
self._uni = FreqDist()
self._bi = ConditionalFreqDist()
self._tri = ConditionalFreqDist()
self._wd = ConditionalFreqDist()
self._eos = ConditionalFreqDist()
self._l1 = 0.0
self._l2 = 0.0
self._l3 = 0.0
self._N = N
self._C = C
self._T = Trained
self._unk = unk
# statistical tools (ignore or delete me)
self.unknown = 0
self.known = 0
def train(self, data):
"""
Uses a set of tagged data to train the tagger.
If an unknown word tagger is specified,
it is trained on the same data.
:param data: List of lists of (word, tag) tuples
:type data: tuple(str)
"""
# Ensure that local C flag is initialized before use
C = False
if self._unk is not None and self._T == False:
self._unk.train(data)
for sent in data:
history = [("BOS", False), ("BOS", False)]
for w, t in sent:
# if capitalization is requested,
# and the word begins with a capital
# set local flag C to True
if self._C and w[0].isupper():
C = True
self._wd[w][t] += 1
self._uni[(t, C)] += 1
self._bi[history[1]][(t, C)] += 1
self._tri[tuple(history)][(t, C)] += 1
history.append((t, C))
history.pop(0)
# set local flag C to false for the next word
C = False
self._eos[t]["EOS"] += 1
# compute lambda values from the trained frequency distributions
self._compute_lambda()
def _compute_lambda(self):
"""
creates lambda values based upon training data
NOTE: no need to explicitly reference C,
it is contained within the tag variable :: tag == (tag,C)
for each tag trigram (t1, t2, t3)
depending on the maximum value of
- f(t1,t2,t3)-1 / f(t1,t2)-1
- f(t2,t3)-1 / f(t2)-1
- f(t3)-1 / N-1
increment l3,l2, or l1 by f(t1,t2,t3)
ISSUES -- Resolutions:
if 2 values are equal, increment both lambda values
by (f(t1,t2,t3) / 2)
"""
# temporary lambda variables
tl1 = 0.0
tl2 = 0.0
tl3 = 0.0
# for each t1,t2 in system
for history in self._tri.conditions():
(h1, h2) = history
# for each t3 given t1,t2 in system
# (NOTE: tag actually represents (tag,C))
# However no effect within this function
for tag in self._tri[history].keys():
# if there has only been 1 occurrence of this tag in the data
# then ignore this trigram.
if self._uni[tag] == 1:
continue
# safe_div provides a safe floating point division
# it returns -1 if the denominator is 0
c3 = self._safe_div(
(self._tri[history][tag] - 1), (self._tri[history].N() - 1)
)
c2 = self._safe_div((self._bi[h2][tag] - 1), (self._bi[h2].N() - 1))
c1 = self._safe_div((self._uni[tag] - 1), (self._uni.N() - 1))
# if c1 is the maximum value:
if (c1 > c3) and (c1 > c2):
tl1 += self._tri[history][tag]
# if c2 is the maximum value
elif (c2 > c3) and (c2 > c1):
tl2 += self._tri[history][tag]
# if c3 is the maximum value
elif (c3 > c2) and (c3 > c1):
tl3 += self._tri[history][tag]
# if c3, and c2 are equal and larger than c1
elif (c3 == c2) and (c3 > c1):
tl2 += self._tri[history][tag] / 2.0
tl3 += self._tri[history][tag] / 2.0
# if c1, and c2 are equal and larger than c3
# this might be a dumb thing to do....(not sure yet)
elif (c2 == c1) and (c1 > c3):
tl1 += self._tri[history][tag] / 2.0
tl2 += self._tri[history][tag] / 2.0
# otherwise there might be a problem
# eg: all values = 0
else:
pass
# Lambda normalisation:
# ensures that l1+l2+l3 = 1
self._l1 = tl1 / (tl1 + tl2 + tl3)
self._l2 = tl2 / (tl1 + tl2 + tl3)
self._l3 = tl3 / (tl1 + tl2 + tl3)
def _safe_div(self, v1, v2):
"""
Safe floating point division function, does not allow division by 0
returns -1 if the denominator is 0
"""
if v2 == 0:
return -1
else:
return v1 / v2
def tagdata(self, data):
"""
Tags each sentence in a list of sentences
:param data:list of list of words
:type data: [[string,],]
:return: list of list of (word, tag) tuples
Invokes tag(sent) function for each sentence
compiles the results into a list of tagged sentences
each tagged sentence is a list of (word, tag) tuples
"""
res = []
for sent in data:
res1 = self.tag(sent)
res.append(res1)
return res
def tag(self, data):
"""
Tags a single sentence
:param data: list of words
:type data: [string,]
:return: [(word, tag),]
Calls recursive function '_tagword'
to produce a list of tags
Associates the sequence of returned tags
with the correct words in the input sequence
returns a list of (word, tag) tuples
"""
current_state = [(["BOS", "BOS"], 0.0)]
sent = list(data)
tags = self._tagword(sent, current_state)
res = []
for i in range(len(sent)):
# unpack and discard the C flags
(t, C) = tags[i + 2]
res.append((sent[i], t))
return res
def _tagword(self, sent, current_states):
"""
:param sent : List of words remaining in the sentence
:type sent : [word,]
:param current_states : List of possible tag combinations for
the sentence so far, and the log probability
associated with each tag combination
:type current_states : [([tag, ], logprob), ]
Tags the first word in the sentence and
recursively tags the reminder of sentence
Uses formula specified above to calculate the probability
of a particular tag
"""
# if this word marks the end of the sentence,
# return the most probable tag
if sent == []:
(h, logp) = current_states[0]
return h
# otherwise there are more words to be tagged
word = sent[0]
sent = sent[1:]
new_states = []
# if the Capitalisation is requested,
# initialise the flag for this word
C = False
if self._C and word[0].isupper():
C = True
# if word is known
# compute the set of possible tags
# and their associated log probabilities
if word in self._wd:
self.known += 1
for history, curr_sent_logprob in current_states:
logprobs = []
for t in self._wd[word].keys():
tC = (t, C)
p_uni = self._uni.freq(tC)
p_bi = self._bi[history[-1]].freq(tC)
p_tri = self._tri[tuple(history[-2:])].freq(tC)
p_wd = self._wd[word][t] / self._uni[tC]
p = self._l1 * p_uni + self._l2 * p_bi + self._l3 * p_tri
p2 = log(p, 2) + log(p_wd, 2)
# compute the result of appending each tag to this history
new_states.append((history + [tC], curr_sent_logprob + p2))
# otherwise a new word, set of possible tags is unknown
else:
self.unknown += 1
# since a set of possible tags,
# and the probability of each specific tag
# can not be returned from most classifiers:
# specify that any unknown words are tagged with certainty
p = 1
# if no unknown word tagger has been specified
# then use the tag 'Unk'
if self._unk is None:
tag = ("Unk", C)
# otherwise apply the unknown word tagger
else:
[(_w, t)] = list(self._unk.tag([word]))
tag = (t, C)
for history, logprob in current_states:
history.append(tag)
new_states = current_states
# now have computed a set of possible new_states
# sort states by log prob
# set is now ordered greatest to least log probability
new_states.sort(reverse=True, key=itemgetter(1))
# del everything after N (threshold)
# this is the beam search cut
if len(new_states) > self._N:
new_states = new_states[: self._N]
# compute the tags for the rest of the sentence
# return the best list of tags for the sentence
return self._tagword(sent, new_states)
########################################
# helper function -- basic sentence tokenizer
########################################
def basic_sent_chop(data, raw=True):
"""
Basic method for tokenizing input into sentences
for this tagger:
:param data: list of tokens (words or (word, tag) tuples)
:type data: str or tuple(str, str)
:param raw: boolean flag marking the input data
as a list of words or a list of tagged words
:type raw: bool
:return: list of sentences
sentences are a list of tokens
tokens are the same as the input
Function takes a list of tokens and separates the tokens into lists
where each list represents a sentence fragment
This function can separate both tagged and raw sequences into
basic sentences.
Sentence markers are the set of [,.!?]
This is a simple method which enhances the performance of the TnT
tagger. Better sentence tokenization will further enhance the results.
"""
new_data = []
curr_sent = []
sent_mark = [",", ".", "?", "!"]
if raw:
for word in data:
if word in sent_mark:
curr_sent.append(word)
new_data.append(curr_sent)
curr_sent = []
else:
curr_sent.append(word)
else:
for word, tag in data:
if word in sent_mark:
curr_sent.append((word, tag))
new_data.append(curr_sent)
curr_sent = []
else:
curr_sent.append((word, tag))
return new_data
def demo():
from nltk.corpus import brown
sents = list(brown.tagged_sents())
test = list(brown.sents())
tagger = TnT()
tagger.train(sents[200:1000])
tagged_data = tagger.tagdata(test[100:120])
for j in range(len(tagged_data)):
s = tagged_data[j]
t = sents[j + 100]
for i in range(len(s)):
print(s[i], "--", t[i])
print()
def demo2():
from nltk.corpus import treebank
d = list(treebank.tagged_sents())
t = TnT(N=1000, C=False)
s = TnT(N=1000, C=True)
t.train(d[(11) * 100 :])
s.train(d[(11) * 100 :])
for i in range(10):
tacc = t.accuracy(d[i * 100 : ((i + 1) * 100)])
tp_un = t.unknown / (t.known + t.unknown)
tp_kn = t.known / (t.known + t.unknown)
t.unknown = 0
t.known = 0
print("Capitalization off:")
print("Accuracy:", tacc)
print("Percentage known:", tp_kn)
print("Percentage unknown:", tp_un)
print("Accuracy over known words:", (tacc / tp_kn))
sacc = s.accuracy(d[i * 100 : ((i + 1) * 100)])
sp_un = s.unknown / (s.known + s.unknown)
sp_kn = s.known / (s.known + s.unknown)
s.unknown = 0
s.known = 0
print("Capitalization on:")
print("Accuracy:", sacc)
print("Percentage known:", sp_kn)
print("Percentage unknown:", sp_un)
print("Accuracy over known words:", (sacc / sp_kn))
def demo3():
from nltk.corpus import brown, treebank
d = list(treebank.tagged_sents())
e = list(brown.tagged_sents())
d = d[:1000]
e = e[:1000]
d10 = int(len(d) * 0.1)
e10 = int(len(e) * 0.1)
tknacc = 0
sknacc = 0
tallacc = 0
sallacc = 0
tknown = 0
sknown = 0
for i in range(10):
t = TnT(N=1000, C=False)
s = TnT(N=1000, C=False)
dtest = d[(i * d10) : ((i + 1) * d10)]
etest = e[(i * e10) : ((i + 1) * e10)]
dtrain = d[: (i * d10)] + d[((i + 1) * d10) :]
etrain = e[: (i * e10)] + e[((i + 1) * e10) :]
t.train(dtrain)
s.train(etrain)
tacc = t.accuracy(dtest)
tp_un = t.unknown / (t.known + t.unknown)
tp_kn = t.known / (t.known + t.unknown)
tknown += tp_kn
t.unknown = 0
t.known = 0
sacc = s.accuracy(etest)
sp_un = s.unknown / (s.known + s.unknown)
sp_kn = s.known / (s.known + s.unknown)
sknown += sp_kn
s.unknown = 0
s.known = 0
tknacc += tacc / tp_kn
sknacc += sacc / tp_kn
tallacc += tacc
sallacc += sacc
# print(i+1, (tacc / tp_kn), i+1, (sacc / tp_kn), i+1, tacc, i+1, sacc)
print("brown: acc over words known:", 10 * tknacc)
print(" : overall accuracy:", 10 * tallacc)
print(" : words known:", 10 * tknown)
print("treebank: acc over words known:", 10 * sknacc)
print(" : overall accuracy:", 10 * sallacc)
print(" : words known:", 10 * sknown)