Fix divide by zero

examples/experiments/joint-dep-pos.cfg

@@ -0,0 +1,78 @@
+[training]
+patience = 10000
+eval_frequency = 50
+dropout = 0.2
+init_tok2vec = null
+vectors = null
+max_epochs = 100
+orth_variant_level = 0.0
+gold_preproc = true
+max_length = 0
+use_gpu = -1
+scores = ["tags_acc", "uas", "las"]
+score_weights = {"las": 0.8, "tags_acc": 0.2}
+limit = 0
+seed = 0
+accumulate_gradient = 1
+discard_oversize = false
+batch_size = 1000
+batch_by = "words"
+
+#[training.batch_size]
+#@schedules = "compounding.v1"
+#start = 100
+#stop = 1000
+#compound = 1.001
+
+[training.optimizer]
+@optimizers = "Adam.v1"
+learn_rate = 0.001
+beta1 = 0.9
+beta2 = 0.999
+
+[nlp]
+lang = "en"
+vectors = ${training:vectors}
+
+[nlp.pipeline]
+
+[nlp.pipeline.tok2vec]
+factory = "tok2vec"
+
+[nlp.pipeline.tagger]
+factory = "tagger"
+
+[nlp.pipeline.parser]
+factory = "parser"
+learn_tokens = false
+min_action_freq = 1
+beam_width = 1
+beam_update_prob = 1.0
+
+[nlp.pipeline.tagger.model]
+@architectures = "spacy.Tagger.v1"
+
+[nlp.pipeline.tagger.model.tok2vec]
+@architectures = "spacy.Tok2VecTensors.v1"
+width = ${nlp.pipeline.tok2vec.model:width}
+
+[nlp.pipeline.parser.model]
+@architectures = "spacy.TransitionBasedParser.v1"
+nr_feature_tokens = 8
+hidden_width = 64
+maxout_pieces = 3
+
+[nlp.pipeline.parser.model.tok2vec]
+@architectures = "spacy.Tok2VecTensors.v1"
+width = ${nlp.pipeline.tok2vec.model:width}
+
+[nlp.pipeline.tok2vec.model]
+@architectures = "spacy.HashEmbedCNN.v1"
+pretrained_vectors = ${nlp:vectors}
+width = 128
+depth = 4
+window_size = 1
+embed_size = 2000
+maxout_pieces = 3
+subword_features = true
+dropout = null

spacy/cli/train.py

@@ -130,6 +130,9 @@ def train_cli(
     raw_text: Optional[Path] = Opt(None, "--raw-text", "-rt", help="Path to jsonl file with unlabelled text documents."),
     verbose: bool = Opt(False, "--verbose", "-V", "-VV", help="Display more information for debugging purposes"),
     use_gpu: int = Opt(-1, "--use-gpu", "-g", help="Use GPU"),
+    num_workers: int = Opt(None, "-j", help="Parallel Workers"),
+    strategy: str = Opt("allreduce", "--strategy", help="Distributed training strategy (requires spacy_ray)"),
+    ray_address: str = Opt(None, "--address", help="Address of the Ray cluster. Multi-node training (requires spacy_ray)"),
     tag_map_path: Optional[Path] = Opt(None, "--tag-map-path", "-tm", help="Location of JSON-formatted tag map"),
     omit_extra_lookups: bool = Opt(False, "--omit-extra-lookups", "-OEL", help="Don't include extra lookups in model"),
     # fmt: on
@@ -153,34 +156,47 @@ def train_cli(
         with init_tok2vec.open("rb") as file_:
             weights_data = file_.read()
 
-    if use_gpu >= 0:
-        msg.info(f"Using GPU: {use_gpu}")
-        require_gpu(use_gpu)
+    if num_workers and num_workers >= 1:
+        from _ray_async_utils import distributed_setup_and_train
+        distributed_setup_and_train(
+            use_gpu,
+            num_workers,
+            strategy,
+            ray_address,
+            {
+                "config": config_path,
+                "train": train_path,
+                "dev": dev_path,
+                "output": output_path
+            }
+        )
     else:
-        msg.info("Using CPU")
+        msg.info(f"Loading config from: {config_path}")
+        if use_gpu >= 0:
+            require_gpu(use_gpu)
+        nlp, config = load_nlp_and_config(config_path)
+        corpus = Corpus(train_path, dev_path, limit=config["training"]["limit"])
 
-    train(
-        config_path,
-        {"train": train_path, "dev": dev_path},
-        output_path=output_path,
-        raw_text=raw_text,
-        tag_map=tag_map,
-        weights_data=weights_data,
-        omit_extra_lookups=omit_extra_lookups,
-    )
+        train_args = dict(
+            nlp=nlp,
+            config=config,
+            corpus=corpus,
+            output_path=output_path,
+            raw_text=raw_text,
+            tag_map=tag_map,
+            weights_data=weights_data,
+            omit_extra_lookups=omit_extra_lookups
+        )
+
+        if use_gpu >= 0:
+            msg.info(f"Using GPU: {use_gpu}")
+            require_gpu(use_gpu)
+        else:
+            msg.info("Using CPU")
+        train(**train_args)
 
 
-def train(
-    config_path: Path,
-    data_paths: Dict[str, Path],
-    raw_text: Optional[Path] = None,
-    output_path: Optional[Path] = None,
-    tag_map: Optional[Path] = None,
-    weights_data: Optional[bytes] = None,
-    omit_extra_lookups: bool = False,
-) -> None:
-    msg.info(f"Loading config from: {config_path}")
-    # Read the config first without creating objects, to get to the original nlp_config
+def load_nlp_and_config(config_path):
     config = util.load_config(config_path, create_objects=False)
     if config["training"].get("seed"):
         fix_random_seed(config["training"]["seed"])
@@ -189,12 +205,30 @@ def train(
         use_pytorch_for_gpu_memory()
     nlp_config = config["nlp"]
     config = util.load_config(config_path, create_objects=True)
-    training = config["training"]
-    msg.info("Creating nlp from config")
     nlp = util.load_model_from_config(nlp_config)
+    # TODO: This is hacky, but temporary convenience...
+    config["_nlp_config"] = nlp_config
+    return nlp, config
+
+
+def train(
+    nlp,
+    config,
+    corpus,
+    raw_text: Optional[Path] = None,
+    output_path: Optional[Path] = None,
+    tag_map: Optional[Path] = None,
+    weights_data: Optional[bytes] = None,
+    omit_extra_lookups: bool = False,
+    disable_tqdm: bool = False,
+    worker_id: int = 0,
+    num_workers=1,
+) -> None:
+    # Read the config first without creating objects, to get to the original nlp_config
+    training = config["training"]
+    nlp_config = config["_nlp_config"]
+    msg.info("Creating nlp from config")
     optimizer = training["optimizer"]
-    limit = training["limit"]
-    corpus = Corpus(data_paths["train"], data_paths["dev"], limit=limit)
     if "textcat" in nlp_config["pipeline"]:
         verify_textcat_config(nlp, nlp_config)
     if training.get("resume", False):
@@ -242,7 +276,7 @@ def train(
         tok2vec.from_bytes(weights_data)
 
     msg.info("Loading training corpus")
-    train_batches = create_train_batches(nlp, corpus, training)
+    train_batches = create_train_batches(nlp, corpus, training, worker_id)
     evaluate = create_evaluation_callback(nlp, optimizer, corpus, training)
 
     # Create iterator, which yields out info after each optimization step.
@@ -259,12 +293,11 @@ def train(
         eval_frequency=training["eval_frequency"],
         raw_text=raw_text,
     )
-
     msg.info(f"Training. Initial learn rate: {optimizer.learn_rate}")
-    print_row = setup_printer(training, nlp)
-
+    print_row = setup_printer(training, nlp.pipe_names)
+    tqdm_args = dict(total=training["eval_frequency"], leave=False, disable=disable_tqdm)
     try:
-        progress = tqdm.tqdm(total=training["eval_frequency"], leave=False)
+        progress = tqdm.tqdm(**tqdm_args)
         for batch, info, is_best_checkpoint in training_step_iterator:
             progress.update(1)
             if is_best_checkpoint is not None:
@@ -273,7 +306,7 @@ def train(
                 if is_best_checkpoint and output_path is not None:
                     update_meta(training, nlp, info)
                     nlp.to_disk(output_path / "model-best")
-                progress = tqdm.tqdm(total=training["eval_frequency"], leave=False)
+                progress = tqdm.tqdm(**tqdm_args)
     except Exception as e:
         if output_path is not None:
             msg.warn(
@@ -294,7 +327,7 @@ def train(
             msg.good(f"Saved model to output directory {final_model_path}")
 
 
-def create_train_batches(nlp, corpus, cfg):
+def create_train_batches(nlp, corpus, cfg, randomization_index):
     max_epochs = cfg.get("max_epochs", 0)
     train_examples = list(
         corpus.train_dataset(
@@ -310,6 +343,11 @@ def create_train_batches(nlp, corpus, cfg):
     while True:
         if len(train_examples) == 0:
             raise ValueError(Errors.E988)
+        # This is used when doing parallel training to
+        # ensure that the dataset is shuffled differently across all workers.
+        for _ in range(randomization_index):
+            random.random()
+        random.shuffle(train_examples)
         epoch += 1
         if batch_strategy == "padded":
             batches = util.minibatch_by_padded_size(
@@ -443,6 +481,8 @@ def train_while_improving(
         ]
         raw_batches = util.minibatch(raw_examples, size=8)
 
+    start_time = timer()
+    words_seen = 0
     for step, (epoch, batch) in enumerate(train_data):
         dropout = next(dropouts)
         with nlp.select_pipes(enable=to_enable):
@@ -464,13 +504,16 @@ def train_while_improving(
         else:
             score, other_scores = (None, None)
             is_best_checkpoint = None
+        words_seen += sum(len(eg) for eg in batch)
         info = {
             "epoch": epoch,
             "step": step,
+            "words": words_seen,
             "score": score,
             "other_scores": other_scores,
             "losses": losses,
             "checkpoints": results,
+            "seconds": int(timer() - start_time)
         }
         yield batch, info, is_best_checkpoint
         if is_best_checkpoint is not None:
@@ -499,14 +542,16 @@ def subdivide_batch(batch, accumulate_gradient):
         yield subbatch
 
 
-def setup_printer(training, nlp):
+def setup_printer(training, pipe_names):
     score_cols = training["scores"]
     score_widths = [max(len(col), 6) for col in score_cols]
-    loss_cols = [f"Loss {pipe}" for pipe in nlp.pipe_names]
+    loss_cols = [f"Loss {pipe}" for pipe in pipe_names]
     loss_widths = [max(len(col), 8) for col in loss_cols]
     table_header = ["E", "#"] + loss_cols + score_cols + ["Score"]
     table_header = [col.upper() for col in table_header]
     table_widths = [3, 6] + loss_widths + score_widths + [6]
+    table_header.append("WPS (TRAIN)")
+    table_widths.append(len(table_header[-1]))
     table_aligns = ["r" for _ in table_widths]
 
     msg.row(table_header, widths=table_widths)
@@ -516,7 +561,7 @@ def setup_printer(training, nlp):
         try:
             losses = [
                 "{0:.2f}".format(float(info["losses"][pipe_name]))
-                for pipe_name in nlp.pipe_names
+                for pipe_name in pipe_names
             ]
         except KeyError as e:
             raise KeyError(
@@ -542,6 +587,7 @@ def setup_printer(training, nlp):
             + losses
             + scores
             + ["{0:.2f}".format(float(info["score"]))]
+            + ["%d" % (info["words"] / (info["seconds"] + 1e-6))]
         )
         msg.row(data, widths=table_widths, aligns=table_aligns)
 
@@ -567,6 +613,9 @@ def verify_cli_args(
     raw_text=None,
     verbose=False,
     use_gpu=-1,
+    num_workers=None,
+    strategy=None,
+    ray_address=None,
     tag_map_path=None,
     omit_extra_lookups=False,
 ):
@@ -592,13 +641,16 @@ def verify_cli_args(
     if code_path is not None:
         if not code_path.exists():
             msg.fail("Path to Python code not found", code_path, exits=1)
-        try:
-            util.import_file("python_code", code_path)
-        except Exception as e:
-            msg.fail(f"Couldn't load Python code: {code_path}", e, exits=1)
+        util.import_file("python_code", code_path)
     if init_tok2vec is not None and not init_tok2vec.exists():
         msg.fail("Can't find pretrained tok2vec", init_tok2vec, exits=1)
 
+    if num_workers and num_workers > 1:
+        try:
+            import ray
+        except ImportError:
+            msg.fail("Need to `pip install ray` to use distributed training!", exits=1)
+
 
 def verify_textcat_config(nlp, nlp_config):
     # if 'positive_label' is provided: double check whether it's in the data and