Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup_printer(config):
score_cols = config["training"]["scores"]
score_widths = [max(len(col), 6) for col in score_cols]
loss_cols = ["Loss {}".format(pipe) for pipe in config["nlp"]["pipeline"]]
loss_widths = [max(len(col), 8) for col in loss_cols]
table_header = ["#"] + loss_cols + score_cols + ["Score"]
table_header = [col.upper() for col in table_header]
table_widths = [6] + loss_widths + score_widths + [6]
table_aligns = ["r" for _ in table_widths]
msg.row(table_header, widths=table_widths)
msg.row(["-" * width for width in table_widths])
def print_row(info):
losses = [
"{0:.2f}".format(info["losses"].get(col, 0.0))
for col in config["nlp"]["pipeline"]
]
scores = [
"{0:.2f}".format(info["other_scores"].get(col, 0.0))
for col in config["training"]["scores"]
]
data = [info["step"]] + losses + scores + ["{0:.2f}".format(info["score"])]
msg.row(data, widths=table_widths, aligns=table_aligns)
return print_row
else:
nlp = spacy.load(model)
msg.good(f"Loaded model '{model}'")
train_data, labels = format_data(srsly.read_jsonl(train_path))
eval_data, _ = format_data(srsly.read_jsonl(eval_path))
textcat = nlp.create_pipe("textcat")
for label in labels:
textcat.add_label(label)
nlp.add_pipe(textcat)
cfg = {"textcat": {"exclusive_classes": True}}
optimizer = nlp.begin_training(component_cfg=cfg)
batch_size = spacy.util.compounding(1.0, 16.0, 1.001)
best_acc = 0
best_model = None
row_widths = (2, 8, 8)
msg.row(("#", "L", "F"), widths=row_widths)
for i in range(n_iter):
random.shuffle(train_data)
losses = {}
data = tqdm.tqdm(train_data, leave=False)
for batch in spacy.util.minibatch(data, size=batch_size):
texts, annots = zip(*batch)
nlp.update(texts, annots, drop=0.2, losses=losses)
with nlp.use_params(optimizer.averages):
sc = nlp.evaluate(eval_data)
if sc.textcat_score > best_acc:
best_acc = sc.textcat_score
if output:
best_model = nlp.to_bytes()
acc = f"{sc.textcat_score:.3f}"
msg.row((i + 1, f"{losses['textcat']:.2f}", acc), widths=row_widths)
msg.text(f"Best F-Score: {best_acc:.3f}")
"the labels '{}'".format(", ".join(textcat_labels))
)
else:
msg.fail(
"Unsupported textcat configuration. Use `spacy debug-data` "
"for more information."
)
# fmt: off
row_head, output_stats = _configure_training_output(pipeline, use_gpu, has_beam_widths)
row_widths = [len(w) for w in row_head]
row_settings = {"widths": row_widths, "aligns": tuple(["r" for i in row_head]), "spacing": 2}
# fmt: on
print("")
msg.row(row_head, **row_settings)
msg.row(["-" * width for width in row_settings["widths"]], **row_settings)
try:
iter_since_best = 0
best_score = 0.0
for i in range(n_iter):
train_docs = corpus.train_docs(
nlp,
noise_level=noise_level,
orth_variant_level=orth_variant_level,
gold_preproc=gold_preproc,
max_length=0,
ignore_misaligned=True,
)
if raw_text:
random.shuffle(raw_text)
raw_batches = util.minibatch(
(nlp.make_doc(rt["text"]) for rt in raw_text), size=8
def print_row(info):
losses = [
"{0:.2f}".format(info["losses"].get(col, 0.0))
for col in config["nlp"]["pipeline"]
]
scores = [
"{0:.2f}".format(info["other_scores"].get(col, 0.0))
for col in config["training"]["scores"]
]
data = [info["step"]] + losses + scores + ["{0:.2f}".format(info["score"])]
msg.row(data, widths=table_widths, aligns=table_aligns)
def setup_printer(config):
score_cols = config["training"]["scores"]
score_widths = [max(len(col), 6) for col in score_cols]
loss_cols = ["Loss {}".format(pipe) for pipe in config["nlp"]["pipeline"]]
loss_widths = [max(len(col), 8) for col in loss_cols]
table_header = ["#"] + loss_cols + score_cols + ["Score"]
table_header = [col.upper() for col in table_header]
table_widths = [6] + loss_widths + score_widths + [6]
table_aligns = ["r" for _ in table_widths]
msg.row(table_header, widths=table_widths)
msg.row(["-" * width for width in table_widths])
def print_row(info):
losses = [
"{0:.2f}".format(info["losses"].get(col, 0.0))
for col in config["nlp"]["pipeline"]
]
scores = [
"{0:.2f}".format(info["other_scores"].get(col, 0.0))
for col in config["training"]["scores"]
]
data = [info["step"]] + losses + scores + ["{0:.2f}".format(info["score"])]
msg.row(data, widths=table_widths, aligns=table_aligns)
return print_row
ner.add_label(label)
nlp.add_pipe(ner)
t2v_cfg = {
"embed_rows": 10000,
"token_vector_width": 128,
"conv_depth": 8,
"nr_feature_tokens": 3,
}
optimizer = nlp.begin_training(component_cfg={"ner": t2v_cfg} if tok2vec else {})
if tok2vec:
_load_pretrained_tok2vec(nlp, Path(tok2vec))
batch_size = spacy.util.compounding(1.0, 16.0, 1.001)
best_acc = 0
best_model = None
row_widths = (2, 8, 8, 8, 8)
msg.row(("#", "L", "P", "R", "F"), widths=row_widths)
for i in range(n_iter):
random.shuffle(train_data)
losses = {}
data = tqdm.tqdm(train_data, leave=False)
for batch in spacy.util.minibatch(data, size=batch_size):
texts, annots = zip(*batch)
nlp.update(texts, annots, drop=0.2, losses=losses)
with nlp.use_params(optimizer.averages):
sc = nlp.evaluate(eval_data)
if sc.ents_f > best_acc:
best_acc = sc.ents_f
if output:
best_model = nlp.to_bytes()
acc = (f"{sc.ents_p:.3f}", f"{sc.ents_r:.3f}", f"{sc.ents_f:.3f}")
msg.row((i + 1, f"{losses['ner']:.2f}", *acc), widths=row_widths)
msg.text(f"Best F-Score: {best_acc:.3f}")
losses,
scorer.scores,
output_stats,
beam_width=beam_width if has_beam_widths else None,
cpu_wps=cpu_wps,
gpu_wps=gpu_wps,
)
if i == 0 and "textcat" in pipeline:
textcats_per_cat = scorer.scores.get("textcats_per_cat", {})
for cat, cat_score in textcats_per_cat.items():
if cat_score.get("roc_auc_score", 0) < 0:
msg.warn(
"Textcat ROC AUC score is undefined due to "
"only one value in label '{}'.".format(cat)
)
msg.row(progress, **row_settings)
# Early stopping
if n_early_stopping is not None:
current_score = _score_for_model(meta)
if current_score < best_score:
iter_since_best += 1
else:
iter_since_best = 0
best_score = current_score
if iter_since_best >= n_early_stopping:
msg.text(
"Early stopping, best iteration "
"is: {}".format(i - iter_since_best)
)
msg.text(
"Best score = {}; Final iteration "
"score = {}".format(best_score, current_score)
"Textcat evaluation score: F1-score macro-averaged across "
"the labels '{}'".format(", ".join(textcat_labels))
)
else:
msg.fail(
"Unsupported textcat configuration. Use `spacy debug-data` "
"for more information."
)
# fmt: off
row_head, output_stats = _configure_training_output(pipeline, use_gpu, has_beam_widths)
row_widths = [len(w) for w in row_head]
row_settings = {"widths": row_widths, "aligns": tuple(["r" for i in row_head]), "spacing": 2}
# fmt: on
print("")
msg.row(row_head, **row_settings)
msg.row(["-" * width for width in row_settings["widths"]], **row_settings)
try:
iter_since_best = 0
best_score = 0.0
for i in range(n_iter):
train_docs = corpus.train_docs(
nlp,
noise_level=noise_level,
orth_variant_level=orth_variant_level,
gold_preproc=gold_preproc,
max_length=0,
ignore_misaligned=True,
)
if raw_text:
random.shuffle(raw_text)
raw_batches = util.minibatch(