wip: feat: new llm extractor pipe

Author: percevalw

edsnlp/core/stream.py

@@ -701,37 +701,41 @@ def map_pipeline(
         """
         new_ops = []
         tokenizer = model.tokenizer
-        has_batches = batch_by is not None
         for op in self.ops:
             # check if the pipe has a "tokenizer" kwarg and update the kwargs if needed
             op = copy(op)
             if (
-                isinstance(op, MapOp)
-                and "tokenizer" in signature(op.pipe).parameters
-                and "tokenizer" not in op.kwargs
-            ):
-                op.kwargs["tokenizer"] = tokenizer
-            elif (
-                isinstance(op, MapBatchesOp)
-                and hasattr(op.pipe, "batch_process")
-                and "tokenizer" in signature(op.pipe.batch_process).parameters
-                and "tokenizer" not in op.kwargs
-            ) or (
-                isinstance(op, MapBatchesOp)
-                and callable(op.pipe)
-                and "tokenizer" in signature(op.pipe).parameters
-                and "tokenizer" not in op.kwargs
+                (
+                    isinstance(op, MapOp)
+                    and "tokenizer" in signature(op.pipe).parameters
+                    and "tokenizer" not in op.kwargs
+                )
+                or (
+                    isinstance(op, MapBatchesOp)
+                    and hasattr(op.pipe, "batch_process")
+                    and "tokenizer" in signature(op.pipe.batch_process).parameters
+                    and "tokenizer" not in op.kwargs
+                )
+                or (
+                    isinstance(op, MapBatchesOp)
+                    and callable(op.pipe)
+                    and "tokenizer" in signature(op.pipe).parameters
+                    and "tokenizer" not in op.kwargs
+                )
             ):
                 op.kwargs["tokenizer"] = tokenizer
-                has_batches = True
             if isinstance(op, (MapOp, MapBatchesOp)):
                 op.context["tokenizer"] = tokenizer
             new_ops.append(op)
+        has_batches = batch_by is not None or (
+            hasattr(p, "batch_process") for n, p in model.pipeline
+        )
         new_ops.append(MapOp(model._ensure_doc, {}))
         if has_batches:
             batch_size, batch_by = self.validate_batching(batch_size, batch_by)
             batch_by = batchify_fns.get(batch_by, batch_by)
             new_ops.append(BatchifyOp(batch_size, batch_by))
+
         for name, pipe in model.pipeline:
             if name not in model._disabled:
                 op = (