mirror of https://github.com/explosion/spaCy.git
401 lines
15 KiB
Cython
401 lines
15 KiB
Cython
# cython: infer_types=True, profile=True, binding=True
|
|
from typing import List
|
|
import numpy
|
|
import srsly
|
|
from thinc.api import Model, set_dropout_rate, SequenceCategoricalCrossentropy, Config
|
|
from thinc.types import Floats2d
|
|
import warnings
|
|
from itertools import islice
|
|
|
|
from ..tokens.doc cimport Doc
|
|
from ..morphology cimport Morphology
|
|
from ..vocab cimport Vocab
|
|
|
|
from .pipe import Pipe, deserialize_config
|
|
from ..language import Language
|
|
from ..attrs import POS, ID
|
|
from ..parts_of_speech import X
|
|
from ..errors import Errors, TempErrors, Warnings
|
|
from ..scorer import Scorer
|
|
from ..training import validate_examples
|
|
from .. import util
|
|
|
|
|
|
default_model_config = """
|
|
[model]
|
|
@architectures = "spacy.Tagger.v1"
|
|
|
|
[model.tok2vec]
|
|
@architectures = "spacy.HashEmbedCNN.v1"
|
|
pretrained_vectors = null
|
|
width = 96
|
|
depth = 4
|
|
embed_size = 2000
|
|
window_size = 1
|
|
maxout_pieces = 3
|
|
subword_features = true
|
|
"""
|
|
DEFAULT_TAGGER_MODEL = Config().from_str(default_model_config)["model"]
|
|
|
|
|
|
@Language.factory(
|
|
"tagger",
|
|
assigns=["token.tag"],
|
|
default_config={"model": DEFAULT_TAGGER_MODEL},
|
|
default_score_weights={"tag_acc": 1.0},
|
|
)
|
|
def make_tagger(nlp: Language, name: str, model: Model):
|
|
"""Construct a part-of-speech tagger component.
|
|
|
|
model (Model[List[Doc], List[Floats2d]]): A model instance that predicts
|
|
the tag probabilities. The output vectors should match the number of tags
|
|
in size, and be normalized as probabilities (all scores between 0 and 1,
|
|
with the rows summing to 1).
|
|
"""
|
|
return Tagger(nlp.vocab, model, name)
|
|
|
|
|
|
class Tagger(Pipe):
|
|
"""Pipeline component for part-of-speech tagging.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger
|
|
"""
|
|
def __init__(self, vocab, model, name="tagger", *, labels=None):
|
|
"""Initialize a part-of-speech tagger.
|
|
|
|
vocab (Vocab): The shared vocabulary.
|
|
model (thinc.api.Model): The Thinc Model powering the pipeline component.
|
|
name (str): The component instance name, used to add entries to the
|
|
losses during training.
|
|
labels (List): The set of labels. Defaults to None.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#init
|
|
"""
|
|
self.vocab = vocab
|
|
self.model = model
|
|
self.name = name
|
|
self._rehearsal_model = None
|
|
cfg = {"labels": labels or []}
|
|
self.cfg = dict(sorted(cfg.items()))
|
|
|
|
@property
|
|
def labels(self):
|
|
"""The labels currently added to the component. Note that even for a
|
|
blank component, this will always include the built-in coarse-grained
|
|
part-of-speech tags by default.
|
|
|
|
RETURNS (Tuple[str]): The labels.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#labels
|
|
"""
|
|
return tuple(self.cfg["labels"])
|
|
|
|
@property
|
|
def label_data(self):
|
|
"""Data about the labels currently added to the component."""
|
|
return tuple(self.cfg["labels"])
|
|
|
|
def __call__(self, doc):
|
|
"""Apply the pipe to a Doc.
|
|
|
|
doc (Doc): The document to process.
|
|
RETURNS (Doc): The processed Doc.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#call
|
|
"""
|
|
tags = self.predict([doc])
|
|
self.set_annotations([doc], tags)
|
|
return doc
|
|
|
|
def pipe(self, stream, *, batch_size=128):
|
|
"""Apply the pipe to a stream of documents. This usually happens under
|
|
the hood when the nlp object is called on a text and all components are
|
|
applied to the Doc.
|
|
|
|
stream (Iterable[Doc]): A stream of documents.
|
|
batch_size (int): The number of documents to buffer.
|
|
YIELDS (Doc): Processed documents in order.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#pipe
|
|
"""
|
|
for docs in util.minibatch(stream, size=batch_size):
|
|
tag_ids = self.predict(docs)
|
|
self.set_annotations(docs, tag_ids)
|
|
yield from docs
|
|
|
|
def predict(self, docs):
|
|
"""Apply the pipeline's model to a batch of docs, without modifying them.
|
|
|
|
docs (Iterable[Doc]): The documents to predict.
|
|
RETURNS: The models prediction for each document.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#predict
|
|
"""
|
|
if not any(len(doc) for doc in docs):
|
|
# Handle cases where there are no tokens in any docs.
|
|
n_labels = len(self.labels)
|
|
guesses = [self.model.ops.alloc((0, n_labels)) for doc in docs]
|
|
assert len(guesses) == len(docs)
|
|
return guesses
|
|
scores = self.model.predict(docs)
|
|
assert len(scores) == len(docs), (len(scores), len(docs))
|
|
guesses = self._scores2guesses(scores)
|
|
assert len(guesses) == len(docs)
|
|
return guesses
|
|
|
|
def _scores2guesses(self, scores):
|
|
guesses = []
|
|
for doc_scores in scores:
|
|
doc_guesses = doc_scores.argmax(axis=1)
|
|
if not isinstance(doc_guesses, numpy.ndarray):
|
|
doc_guesses = doc_guesses.get()
|
|
guesses.append(doc_guesses)
|
|
return guesses
|
|
|
|
def set_annotations(self, docs, batch_tag_ids):
|
|
"""Modify a batch of documents, using pre-computed scores.
|
|
|
|
docs (Iterable[Doc]): The documents to modify.
|
|
batch_tag_ids: The IDs to set, produced by Tagger.predict.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#set_annotations
|
|
"""
|
|
if isinstance(docs, Doc):
|
|
docs = [docs]
|
|
cdef Doc doc
|
|
cdef Vocab vocab = self.vocab
|
|
for i, doc in enumerate(docs):
|
|
doc_tag_ids = batch_tag_ids[i]
|
|
if hasattr(doc_tag_ids, "get"):
|
|
doc_tag_ids = doc_tag_ids.get()
|
|
for j, tag_id in enumerate(doc_tag_ids):
|
|
# Don't clobber preset POS tags
|
|
if doc.c[j].tag == 0:
|
|
doc.c[j].tag = self.vocab.strings[self.labels[tag_id]]
|
|
|
|
def update(self, examples, *, drop=0., sgd=None, losses=None, set_annotations=False):
|
|
"""Learn from a batch of documents and gold-standard information,
|
|
updating the pipe's model. Delegates to predict and get_loss.
|
|
|
|
examples (Iterable[Example]): A batch of Example objects.
|
|
drop (float): The dropout rate.
|
|
set_annotations (bool): Whether or not to update the Example objects
|
|
with the predictions.
|
|
sgd (thinc.api.Optimizer): The optimizer.
|
|
losses (Dict[str, float]): Optional record of the loss during training.
|
|
Updated using the component name as the key.
|
|
RETURNS (Dict[str, float]): The updated losses dictionary.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#update
|
|
"""
|
|
if losses is None:
|
|
losses = {}
|
|
losses.setdefault(self.name, 0.0)
|
|
validate_examples(examples, "Tagger.update")
|
|
if not any(len(eg.predicted) if eg.predicted else 0 for eg in examples):
|
|
# Handle cases where there are no tokens in any docs.
|
|
return
|
|
set_dropout_rate(self.model, drop)
|
|
tag_scores, bp_tag_scores = self.model.begin_update([eg.predicted for eg in examples])
|
|
for sc in tag_scores:
|
|
if self.model.ops.xp.isnan(sc.sum()):
|
|
raise ValueError(Errors.E940)
|
|
loss, d_tag_scores = self.get_loss(examples, tag_scores)
|
|
bp_tag_scores(d_tag_scores)
|
|
if sgd not in (None, False):
|
|
self.model.finish_update(sgd)
|
|
|
|
losses[self.name] += loss
|
|
if set_annotations:
|
|
docs = [eg.predicted for eg in examples]
|
|
self.set_annotations(docs, self._scores2guesses(tag_scores))
|
|
return losses
|
|
|
|
def rehearse(self, examples, *, drop=0., sgd=None, losses=None):
|
|
"""Perform a "rehearsal" update from a batch of data. Rehearsal updates
|
|
teach the current model to make predictions similar to an initial model,
|
|
to try to address the "catastrophic forgetting" problem. This feature is
|
|
experimental.
|
|
|
|
examples (Iterable[Example]): A batch of Example objects.
|
|
drop (float): The dropout rate.
|
|
sgd (thinc.api.Optimizer): The optimizer.
|
|
losses (Dict[str, float]): Optional record of the loss during training.
|
|
Updated using the component name as the key.
|
|
RETURNS (Dict[str, float]): The updated losses dictionary.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#rehearse
|
|
"""
|
|
validate_examples(examples, "Tagger.rehearse")
|
|
docs = [eg.predicted for eg in examples]
|
|
if self._rehearsal_model is None:
|
|
return
|
|
if not any(len(doc) for doc in docs):
|
|
# Handle cases where there are no tokens in any docs.
|
|
return
|
|
set_dropout_rate(self.model, drop)
|
|
guesses, backprop = self.model.begin_update(docs)
|
|
target = self._rehearsal_model(examples)
|
|
gradient = guesses - target
|
|
backprop(gradient)
|
|
self.model.finish_update(sgd)
|
|
if losses is not None:
|
|
losses.setdefault(self.name, 0.0)
|
|
losses[self.name] += (gradient**2).sum()
|
|
|
|
def get_loss(self, examples, scores):
|
|
"""Find the loss and gradient of loss for the batch of documents and
|
|
their predicted scores.
|
|
|
|
examples (Iterable[Examples]): The batch of examples.
|
|
scores: Scores representing the model's predictions.
|
|
RETUTNRS (Tuple[float, float]): The loss and the gradient.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#get_loss
|
|
"""
|
|
validate_examples(examples, "Tagger.get_loss")
|
|
loss_func = SequenceCategoricalCrossentropy(names=self.labels, normalize=False)
|
|
truths = [eg.get_aligned("TAG", as_string=True) for eg in examples]
|
|
d_scores, loss = loss_func(scores, truths)
|
|
if self.model.ops.xp.isnan(loss):
|
|
raise ValueError("nan value when computing loss")
|
|
return float(loss), d_scores
|
|
|
|
def initialize(self, get_examples, *, nlp=None, labels=None):
|
|
"""Initialize the pipe for training, using a representative set
|
|
of data examples.
|
|
|
|
get_examples (Callable[[], Iterable[Example]]): Function that
|
|
returns a representative sample of gold-standard Example objects..
|
|
nlp (Language): The current nlp object the component is part of.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#initialize
|
|
"""
|
|
self._ensure_examples(get_examples)
|
|
if labels is not None:
|
|
for tag in labels:
|
|
self.add_label(tag)
|
|
else:
|
|
tags = set()
|
|
for example in get_examples():
|
|
for token in example.y:
|
|
if token.tag_:
|
|
tags.add(token.tag_)
|
|
for tag in sorted(tags):
|
|
self.add_label(tag)
|
|
doc_sample = []
|
|
label_sample = []
|
|
for example in islice(get_examples(), 10):
|
|
doc_sample.append(example.x)
|
|
gold_tags = example.get_aligned("TAG", as_string=True)
|
|
gold_array = [[1.0 if tag == gold_tag else 0.0 for tag in self.labels] for gold_tag in gold_tags]
|
|
label_sample.append(self.model.ops.asarray(gold_array, dtype="float32"))
|
|
assert len(doc_sample) > 0, Errors.E923.format(name=self.name)
|
|
assert len(label_sample) > 0, Errors.E923.format(name=self.name)
|
|
self.model.initialize(X=doc_sample, Y=label_sample)
|
|
|
|
def add_label(self, label):
|
|
"""Add a new label to the pipe.
|
|
|
|
label (str): The label to add.
|
|
RETURNS (int): 0 if label is already present, otherwise 1.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#add_label
|
|
"""
|
|
if not isinstance(label, str):
|
|
raise ValueError(Errors.E187)
|
|
if label in self.labels:
|
|
return 0
|
|
self._allow_extra_label()
|
|
self.cfg["labels"].append(label)
|
|
self.vocab.strings.add(label)
|
|
return 1
|
|
|
|
def score(self, examples, **kwargs):
|
|
"""Score a batch of examples.
|
|
|
|
examples (Iterable[Example]): The examples to score.
|
|
RETURNS (Dict[str, Any]): The scores, produced by
|
|
Scorer.score_token_attr for the attributes "tag".
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#score
|
|
"""
|
|
validate_examples(examples, "Tagger.score")
|
|
return Scorer.score_token_attr(examples, "tag", **kwargs)
|
|
|
|
def to_bytes(self, *, exclude=tuple()):
|
|
"""Serialize the pipe to a bytestring.
|
|
|
|
exclude (Iterable[str]): String names of serialization fields to exclude.
|
|
RETURNS (bytes): The serialized object.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#to_bytes
|
|
"""
|
|
serialize = {}
|
|
serialize["model"] = self.model.to_bytes
|
|
serialize["vocab"] = self.vocab.to_bytes
|
|
serialize["cfg"] = lambda: srsly.json_dumps(self.cfg)
|
|
return util.to_bytes(serialize, exclude)
|
|
|
|
def from_bytes(self, bytes_data, *, exclude=tuple()):
|
|
"""Load the pipe from a bytestring.
|
|
|
|
bytes_data (bytes): The serialized pipe.
|
|
exclude (Iterable[str]): String names of serialization fields to exclude.
|
|
RETURNS (Tagger): The loaded Tagger.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#from_bytes
|
|
"""
|
|
def load_model(b):
|
|
try:
|
|
self.model.from_bytes(b)
|
|
except AttributeError:
|
|
raise ValueError(Errors.E149) from None
|
|
|
|
deserialize = {
|
|
"vocab": lambda b: self.vocab.from_bytes(b),
|
|
"cfg": lambda b: self.cfg.update(srsly.json_loads(b)),
|
|
"model": lambda b: load_model(b),
|
|
}
|
|
util.from_bytes(bytes_data, deserialize, exclude)
|
|
return self
|
|
|
|
def to_disk(self, path, *, exclude=tuple()):
|
|
"""Serialize the pipe to disk.
|
|
|
|
path (str / Path): Path to a directory.
|
|
exclude (Iterable[str]): String names of serialization fields to exclude.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#to_disk
|
|
"""
|
|
serialize = {
|
|
"vocab": lambda p: self.vocab.to_disk(p),
|
|
"model": lambda p: self.model.to_disk(p),
|
|
"cfg": lambda p: srsly.write_json(p, self.cfg),
|
|
}
|
|
util.to_disk(path, serialize, exclude)
|
|
|
|
def from_disk(self, path, *, exclude=tuple()):
|
|
"""Load the pipe from disk. Modifies the object in place and returns it.
|
|
|
|
path (str / Path): Path to a directory.
|
|
exclude (Iterable[str]): String names of serialization fields to exclude.
|
|
RETURNS (Tagger): The modified Tagger object.
|
|
|
|
DOCS: https://nightly.spacy.io/api/tagger#from_disk
|
|
"""
|
|
def load_model(p):
|
|
with p.open("rb") as file_:
|
|
try:
|
|
self.model.from_bytes(file_.read())
|
|
except AttributeError:
|
|
raise ValueError(Errors.E149) from None
|
|
|
|
deserialize = {
|
|
"vocab": lambda p: self.vocab.from_disk(p),
|
|
"cfg": lambda p: self.cfg.update(deserialize_config(p)),
|
|
"model": load_model,
|
|
}
|
|
util.from_disk(path, deserialize, exclude)
|
|
return self
|