Repository: deepset-ai/haystack
Pull request: feat: Add `NamedEntityExtractor`component
### Proposed Changes:

 <!--- In case of a bug: Describe what caused the issue and how you solved it -->
 <!--- In case of a feature: Describe what did you add and how it works -->
This PR introduces a new extractor component, namely `NamedEntityExtractor`. This component accepts a list of `Documents` as its input - the raw text in the documents are annotated by the extractor and the annotations are stored in the document's `meta` dictionary (under the key `named_entities`).

The component is designed to support multiple NER backends, and the current implementations support two at the moment: Hugging Face and spaCy. These two backends impelment support for any HF/spaCy model that supports token classification/NER respectively.

The `_NerBackend` class abstracts away the details of the NLP provider and is exposed to the public API by way of the `NamedEntityExtractorBackend` enum. Predicted annotations are stored as instances of the `NamedEntityAnnotation` dataclass. The component also supports basic serialization and GPU inference.

Closes #5679.

#### Potential future improvements
- Integration into Haystack's device management feature.
- New backends such as [flair](https://github.com/flairNLP/flair).
- Improve device context switching in the spaCy backend.

### How did you test it?

<!-- unit tests, integration tests, manual verification, instructions for manual tests -->
Includes E2E and unit tests.

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue

Problem statement:
`NamedEntityExtractor`
Very similar to our Haystack 1.x `EntityExtractor`: given some text, extracts all named entities it can find. It will take Documents and attach those entities into the Document’s metadata (together with the entity’s position in the original text).

Draft I/O for `NamedEntityExtractor`:
```python
@component
class NamedEntityExtractor:
    
    def __init__(self, model_name_or_path: str, entities: List[str]):
        self.model_name_or_path = model_name_or_path

    def warm_up(self):
        # init the model

    @component.output_types(documents=List[Document])
    def run(self, documents: List[Document]):
        # extracts the entities and puts them in the docs metadata
        return {"documents": documents}
```

Hints from issue discussion:
here I would like to see https://github.com/deepset-ai/haystack/blob/main/haystack/nodes/extractor/entity.py being migrated to 2.x. I like the pure NER approach but I am not sure if we need the "QA for NER" approach. Instead I could see a prompt based approach of extracting NERs instead of using extrative QA.

Commit message: feat: Add `NamedEntityExtractor`component

This component accepts a list of `Document`s which it annotates with named entities. The annotations are stored in the `meta` dictionary of each `Document` under a specific key.

The component currently support two backends for the annotation models: Hugging Face `transformers` and spaCy.

Patch to review:
diff --git a/e2e/pipelines/test_named_entity_extractor.py b/e2e/pipelines/test_named_entity_extractor.py
new file mode 100644
index 0000000000..401a9343a2
--- /dev/null
+++ b/e2e/pipelines/test_named_entity_extractor.py
@@ -0,0 +1,113 @@
+import pytest
+
+from haystack import Document, Pipeline, ComponentError
+from haystack.components.extractors import NamedEntityAnnotation, NamedEntityExtractor, NamedEntityExtractorBackend
+
+
+SPACY_TEST_MODEL_NAME = "en_core_web_trf"
+
+
+@pytest.fixture
+def raw_texts():
+    return [
+        "My name is Clara and I live in Berkeley, California.",
+        "I'm Merlin, the happy pig!",
+        "New York State declared a state of emergency after the announcement of the end of the world.",
+        "",  # Intentionally empty.
+    ]
+
+
+@pytest.fixture
+def hf_annotations():
+    return [
+        [
+            NamedEntityAnnotation(entity="PER", start=11, end=16),
+            NamedEntityAnnotation(entity="LOC", start=31, end=39),
+            NamedEntityAnnotation(entity="LOC", start=41, end=51),
+        ],
+        [NamedEntityAnnotation(entity="PER", start=4, end=10)],
+        [NamedEntityAnnotation(entity="LOC", start=0, end=14)],
+        [],
+    ]
+
+
+@pytest.fixture
+def spacy_annotations():
+    return [
+        [
+            NamedEntityAnnotation(entity="PERSON", start=11, end=16),
+            NamedEntityAnnotation(entity="GPE", start=31, end=39),
+            NamedEntityAnnotation(entity="GPE", start=41, end=51),
+        ],
+        [NamedEntityAnnotation(entity="PERSON", start=4, end=10)],
+        [NamedEntityAnnotation(entity="GPE", start=0, end=14)],
+        [],
+    ]
+
+
+def test_ner_extractor_init():
+    extractor = NamedEntityExtractor(
+        backend=NamedEntityExtractorBackend.HUGGING_FACE, model_name_or_path="dslim/bert-base-NER", device_id=-1
+    )
+
+    with pytest.raises(ComponentError, match=r"not initialized"):
+        extractor.run(documents=[])
+
+    assert not extractor.initialized
+    extractor.warm_up()
+    assert extractor.initialized
+
+
+@pytest.mark.parametrize("batch_size", [1, 3])
+def test_ner_extractor_hf_backend(raw_texts, hf_annotations, batch_size):
+    extractor = NamedEntityExtractor(
+        backend=NamedEntityExtractorBackend.HUGGING_FACE, model_name_or_path="dslim/bert-base-NER"
+    )
+    extractor.warm_up()
+
+    _extract_and_check_predictions(extractor, raw_texts, hf_annotations, batch_size)
+
+
+@pytest.mark.parametrize("batch_size", [1, 3])
+def test_ner_extractor_spacy_backend(raw_texts, spacy_annotations, batch_size):
+    extractor = NamedEntityExtractor(backend=NamedEntityExtractorBackend.SPACY, model_name_or_path="en_core_web_trf")
+    extractor.warm_up()
+
+    _extract_and_check_predictions(extractor, raw_texts, spacy_annotations, batch_size)
+
+
+@pytest.mark.parametrize("batch_size", [1, 3])
+def test_ner_extractor_in_pipeline(raw_texts, hf_annotations, batch_size):
+    pipeline = Pipeline()
+    pipeline.add_component(
+        name="ner_extractor",
+        instance=NamedEntityExtractor(
+            backend=NamedEntityExtractorBackend.HUGGING_FACE, model_name_or_path="dslim/bert-base-NER"
+        ),
+    )
+
+    outputs = pipeline.run(
+        {"ner_extractor": {"documents": [Document(content=text) for text in raw_texts], "batch_size": batch_size}}
+    )["ner_extractor"]["documents"]
+    predicted = [NamedEntityExtractor.get_stored_annotations(doc) for doc in outputs]
+    _check_predictions(predicted, hf_annotations)
+
+
+def _extract_and_check_predictions(extractor, texts, expected, batch_size):
+    docs = [Document(content=text) for text in texts]
+    outputs = extractor.run(documents=docs, batch_size=batch_size)["documents"]
+    assert all(id(a) == id(b) for a, b in zip(docs, outputs))
+    predicted = [NamedEntityExtractor.get_stored_annotations(doc) for doc in outputs]
+
+    _check_predictions(predicted, expected)
+
+
+def _check_predictions(predicted, expected):
+    assert len(predicted) == len(expected)
+    for pred, exp in zip(predicted, expected):
+        assert len(pred) == len(exp)
+
+        for a, b in zip(pred, exp):
+            assert a.entity == b.entity
+            assert a.start == b.start
+            assert a.end == b.end
diff --git a/haystack/components/extractors/__init__.py b/haystack/components/extractors/__init__.py
new file mode 100644
index 0000000000..9f11211b97
--- /dev/null
+++ b/haystack/components/extractors/__init__.py
@@ -0,0 +1,3 @@
+from .named_entity_extractor import NamedEntityAnnotation, NamedEntityExtractor, NamedEntityExtractorBackend
+
+__all__ = ["NamedEntityExtractor", "NamedEntityExtractorBackend", "NamedEntityAnnotation"]
diff --git a/haystack/components/extractors/named_entity_extractor.py b/haystack/components/extractors/named_entity_extractor.py
new file mode 100644
index 0000000000..771bd868e6
--- /dev/null
+++ b/haystack/components/extractors/named_entity_extractor.py
@@ -0,0 +1,399 @@
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import Enum, EnumMeta
+from typing import Any, Dict, List, Optional, Union
+
+from ... import ComponentError, DeserializationError, Document, component, default_from_dict, default_to_dict
+from ...lazy_imports import LazyImport
+
+with LazyImport(message="Run 'pip install transformers'") as transformers_import:
+    from transformers import AutoModelForTokenClassification, AutoTokenizer
+    from transformers import Pipeline as HfPipeline
+    from transformers import pipeline
+
+with LazyImport(message="Run 'pip install spacy'") as spacy_import:
+    import spacy
+    from spacy import Language as SpacyPipeline
+
+
+class _BackendEnumMeta(EnumMeta):
+    """
+    Metaclass for fine-grained error handling of backend enums.
+    """
+
+    def __call__(cls, value, names=None, *, module=None, qualname=None, type=None, start=1):
+        if names is None:
+            try:
+                return EnumMeta.__call__(cls, value, names, module=module, qualname=qualname, type=type, start=start)
+            except ValueError:
+                supported_backends = ", ".join(sorted(v.value for v in cls))  # pylint: disable=not-an-iterable
+                raise ComponentError(
+                    f"Invalid backend `{value}` for named entity extractor. "
+                    f"Supported backends: {supported_backends}"
+                )
+        else:
+            return EnumMeta.__call__(  # pylint: disable=too-many-function-args
+                cls, value, names, module, qualname, type, start
+            )
+
+
+class NamedEntityExtractorBackend(Enum, metaclass=_BackendEnumMeta):
+    """
+    NLP backend to use for Named Entity Recognition.
+    """
+
+    #: Hugging Face.
+    #:
+    #: Uses an Hugging Face model and pipeline.
+    HUGGING_FACE = "hugging_face"
+
+    #: spaCy.
+    #:
+    #: Uses a spaCy model and pipeline.
+    SPACY = "spacy"
+
+
+@dataclass
+class NamedEntityAnnotation:
+    """
+    Describes a single NER annotation.
+
+    :param entity:
+        Entity label.
+    :param start:
+        Start index of the entity in the document.
+    :param end:
+        End index of the entity in the document.
+    :param score:
+        Score calculated by the model.
+    """
+
+    entity: str
+    start: int
+    end: int
+    score: Optional[float] = None
+
+
+@component
+class NamedEntityExtractor:
+    _METADATA_KEY = "named_entities"
+
+    def __init__(
+        self, *, backend: Union[str, NamedEntityExtractorBackend], model_name_or_path: str, device_id: int = -1
+    ) -> None:
+        """
+        Construct a Named Entity extractor component.
+
+        :param backend:
+            Backend to use for NER.
+        :param model_name_or_path:
+            Name of the model or a path to the model on
+            the local disk.
+
+            Dependent on the backend.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+
+        if isinstance(backend, str):
+            backend = NamedEntityExtractorBackend(backend)
+
+        self._backend: _NerBackend
+        if backend == NamedEntityExtractorBackend.HUGGING_FACE:
+            self._backend = _HfBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        elif backend == NamedEntityExtractorBackend.SPACY:
+            self._backend = _SpacyBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        else:
+            raise ComponentError(f"Unknown NER backend '{type(backend).__name__}' for extractor")
+
+    def warm_up(self):
+        try:
+            self._backend.initialize()
+        except BaseException as e:
+            raise ComponentError(
+                f"Named entity extractor with backend '{self._backend.type} failed to initialize."
+            ) from e
+
+    @component.output_types(documents=List[Document])
+    def run(self, documents: List[Document], batch_size: int = 1) -> Dict[str, Any]:
+        texts = [doc.content if doc.content is not None else "" for doc in documents]
+        annotations = self._backend.annotate(texts, batch_size=batch_size)
+
+        if len(annotations) != len(documents):
+            raise ComponentError(
+                "NER backend did not return the correct number of annotations; "
+                f"got {len(annotations)} but expected {len(documents)}"
+            )
+
+        for doc, doc_annotations in zip(documents, annotations):
+            doc.meta[self._METADATA_KEY] = doc_annotations
+
+        return {"documents": documents}
+
+    def to_dict(self) -> Dict[str, Any]:
+        if isinstance(self._backend, _HfBackend):
+            backend_type = NamedEntityExtractorBackend.HUGGING_FACE
+        elif isinstance(self._backend, _SpacyBackend):
+            backend_type = NamedEntityExtractorBackend.SPACY
+        else:
+            # Unreachable.
+            raise ComponentError(f"Unknown backend '{type(self._backend).__name__}'")
+
+        return default_to_dict(
+            self, backend=backend_type, model_name_or_path=self._backend.model_name, device_id=self._backend.device_id
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "NamedEntityExtractor":
+        try:
+            return default_from_dict(cls, data)
+        except BaseException as e:
+            raise DeserializationError(f"Couldn't deserialize {cls.__name__} instance") from e
+
+    @property
+    def initialized(self) -> bool:
+        """
+        Returns if the extractor is ready to annotate text.
+        """
+        return self._backend.initialized
+
+    @classmethod
+    def get_stored_annotations(cls, document: Document) -> Optional[List[NamedEntityAnnotation]]:
+        """
+        Returns the document's named entity annotations stored
+        in its metadata, if any.
+
+        :param document:
+            Document whose annotations are to be fetched.
+        :returns:
+            The stored annotations.
+        """
+
+        return document.meta.get(cls._METADATA_KEY)
+
+
+class _NerBackend(ABC):
+    """
+    Base class for NER backends.
+    """
+
+    def __init__(self, type: NamedEntityExtractorBackend) -> None:
+        super().__init__()
+
+        self._type = type
+
+    @abstractmethod
+    def initialize(self):
+        """
+        Initializes the backend. This would usually
+        entail loading models, pipelines, etc.
+        """
+
+    @property
+    @abstractmethod
+    def initialized(self) -> bool:
+        """
+        Returns if the backend has been initialized, i.e,
+        ready to annotate text.
+        """
+
+    @abstractmethod
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        """
+        Predict annotations for a collection of documents.
+
+        :param texts:
+            Raw texts to be annotated.
+        :param batch_size:
+            Size of text batches that are
+            passed to the model.
+        :returns:
+            NER annotations.
+        """
+
+    @property
+    @abstractmethod
+    def model_name(self) -> str:
+        """
+        Returns the model name or path on the local disk.
+        """
+
+    @property
+    @abstractmethod
+    def device_id(self) -> int:
+        """
+        Returns the identifier of the device on which
+        the backend's model is loaded.
+        """
+
+    @property
+    def type(self) -> NamedEntityExtractorBackend:
+        """
+        Returns the type of the backend.
+        """
+        return self._type
+
+
+class _HfBackend(_NerBackend):
+    """
+    Hugging Face backend for NER.
+    """
+
+    def __init__(self, *, model_name_or_path: str, device_id: int) -> None:
+        """
+        Construct a Hugging Face NER backend.
+
+        :param model_name_or_path:
+            Name of the model or a path to the Hugging Face
+            model on the local disk.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+        super().__init__(NamedEntityExtractorBackend.HUGGING_FACE)
+
+        transformers_import.check()
+
+        self._model_name_or_path = model_name_or_path
+        self._device_id = device_id
+
+        self.tokenizer: Optional[AutoTokenizer] = None
+        self.model: Optional[AutoModelForTokenClassification] = None
+        self.pipeline: Optional[HfPipeline] = None
+
+    def initialize(self):
+        self.tokenizer = AutoTokenizer.from_pretrained(self._model_name_or_path)
+        self.model = AutoModelForTokenClassification.from_pretrained(self._model_name_or_path)
+        self.pipeline = pipeline(
+            "ner", model=self.model, tokenizer=self.tokenizer, aggregation_strategy="simple", device=self.device_id
+        )
+
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        if not self.initialized:
+            raise ComponentError("Hugging Face NER backend was not initialized")
+
+        assert self.pipeline is not None
+        outputs = self.pipeline(texts, batch_size=batch_size)
+        return [
+            [
+                NamedEntityAnnotation(
+                    entity=annotation["entity"] if "entity" in annotation else annotation["entity_group"],
+                    start=annotation["start"],
+                    end=annotation["end"],
+                    score=annotation["score"],
+                )
+                for annotation in annotations
+            ]
+            for annotations in outputs
+        ]
+
+    @property
+    def initialized(self) -> bool:
+        return self.tokenizer is not None and self.model is not None or self.pipeline is not None
+
+    @property
+    def model_name(self) -> str:
+        return self._model_name_or_path
+
+    @property
+    def device_id(self) -> int:
+        return self._device_id
+
+
+class _SpacyBackend(_NerBackend):
+    """
+    spaCy backend for NER.
+    """
+
+    def __init__(self, *, model_name_or_path: str, device_id: int) -> None:
+        """
+        Construct a spaCy NER backend.
+
+        :param model_name_or_path:
+            Name of the model or a path to the spaCy
+            model on the local disk.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+        super().__init__(NamedEntityExtractorBackend.SPACY)
+
+        spacy_import.check()
+
+        self._model_name_or_path = model_name_or_path
+        self._device_id = device_id
+
+        self.pipeline: Optional[SpacyPipeline] = None
+
+    def initialize(self):
+        # We need to initialize the model on the GPU if needed.
+        with self._select_device():
+            self.pipeline = spacy.load(self._model_name_or_path)
+
+        if not self.pipeline.has_pipe("ner"):
+            raise ComponentError(f"spaCy pipeline '{self._model_name_or_path}' does not contain an NER component")
+
+        # Disable unnecessary pipes.
+        pipes_to_keep = ("ner", "tok2vec", "transformer", "curated_transformer")
+        for name in self.pipeline.pipe_names:
+            if name not in pipes_to_keep:
+                self.pipeline.disable_pipe(name)
+
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        if not self.initialized:
+            raise ComponentError("spaCy NER backend was not initialized")
+
+        assert self.pipeline is not None
+        with self._select_device():
+            outputs = list(self.pipeline.pipe(texts, batch_size=batch_size))
+
+        return [
+            [
+                NamedEntityAnnotation(entity=entity.label_, start=entity.start_char, end=entity.end_char)
+                for entity in doc.ents
+            ]
+            for doc in outputs
+        ]
+
+    @property
+    def initialized(self) -> bool:
+        return self.pipeline is not None
+
+    @property
+    def model_name(self) -> str:
+        return self._model_name_or_path
+
+    @property
+    def device_id(self) -> int:
+        return self._device_id
+
+    @contextmanager
+    def _select_device(self):
+        """
+        Context manager used to run spaCy models on a specific
+        GPU in a scoped manner.
+        """
+
+        # TODO: This won't restore the active device.
+        # Since there are no opaque API functions to determine
+        # the active device in spaCy/Thinc, we can't do much
+        # about it as a consumer unless we start poking into their
+        # internals.
+        try:
+            if self._device_id >= 0:
+                spacy.require_gpu(self._device_id)
+            yield
+        finally:
+            if self._device_id >= 0:
+                spacy.require_cpu()
diff --git a/releasenotes/notes/named-entity-extractor-component-8fd647ee748892ca.yaml b/releasenotes/notes/named-entity-extractor-component-8fd647ee748892ca.yaml
new file mode 100644
index 0000000000..1b15e5bcf5
--- /dev/null
+++ b/releasenotes/notes/named-entity-extractor-component-8fd647ee748892ca.yaml
@@ -0,0 +1,3 @@
+---
+features:
+  - New NamedEntityExtractor component.
diff --git a/test/components/extractors/__init__.py b/test/components/extractors/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/test/components/extractors/test_named_entity_extractor.py b/test/components/extractors/test_named_entity_extractor.py
new file mode 100644
index 0000000000..9ac2c3e437
--- /dev/null
+++ b/test/components/extractors/test_named_entity_extractor.py
@@ -0,0 +1,36 @@
+import pytest
+
+from haystack import ComponentError, DeserializationError
+from haystack.components.extractors import NamedEntityExtractor, NamedEntityExtractorBackend
+
+
+@pytest.mark.unit
+def test_named_entity_extractor_backend():
+    _ = NamedEntityExtractor(backend=NamedEntityExtractorBackend.HUGGING_FACE, model_name_or_path="dslim/bert-base-NER")
+
+    _ = NamedEntityExtractor(backend="hugging_face", model_name_or_path="dslim/bert-base-NER")
+
+    _ = NamedEntityExtractor(backend=NamedEntityExtractorBackend.SPACY, model_name_or_path="en_core_web_sm")
+
+    _ = NamedEntityExtractor(backend="spacy", model_name_or_path="en_core_web_sm")
+
+    with pytest.raises(ComponentError, match=r"Invalid backend"):
+        NamedEntityExtractor(backend="random_backend", model_name_or_path="dslim/bert-base-NER")
+
+
+@pytest.mark.unit
+def test_named_entity_extractor_serde():
+    extractor = NamedEntityExtractor(
+        backend=NamedEntityExtractorBackend.HUGGING_FACE, model_name_or_path="dslim/bert-base-NER", device_id=-1
+    )
+
+    serde_data = extractor.to_dict()
+    new_extractor = NamedEntityExtractor.from_dict(serde_data)
+
+    assert type(new_extractor._backend) == type(extractor._backend)
+    assert new_extractor._backend.model_name == extractor._backend.model_name
+    assert new_extractor._backend.device_id == extractor._backend.device_id
+
+    with pytest.raises(DeserializationError, match=r"Couldn't deserialize"):
+        serde_data["init_parameters"].pop("backend")
+        _ = NamedEntityExtractor.from_dict(serde_data)
diff --git a/test/test_requirements.txt b/test/test_requirements.txt
index 3d094068e0..5e9c19f707 100644
--- a/test/test_requirements.txt
+++ b/test/test_requirements.txt
@@ -3,6 +3,9 @@
 # Package                                        Components
 
 transformers[torch,sentencepiece]==4.36.2       # ExtractiveReader, TransformersSimilarityRanker, LocalWhisperTranscriber, HFGenerators...
+spacy>=3.7,<3.8                                 # NamedEntityExtractor
+spacy-curated-transformers>=0.2,<=0.3           # NamedEntityExtractor
+https://github.com/explosion/spacy-models/releases/download/en_core_web_trf-3.7.3/en_core_web_trf-3.7.3-py3-none-any.whl # NamedEntityExtractor
 
 # Converters
 pypdf                                           # PyPDFConverter


Code review comments:
diff_hunk: @@ -0,0 +1,399 @@
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import Enum, EnumMeta
+from typing import Any, Dict, List, Optional, Union
+
+from ... import ComponentError, DeserializationError, Document, component, default_from_dict, default_to_dict
+from ...lazy_imports import LazyImport
+
+with LazyImport(message="Run 'pip install transformers'") as transformers_import:
+    from transformers import AutoModelForTokenClassification, AutoTokenizer
+    from transformers import Pipeline as HfPipeline
+    from transformers import pipeline
+
+with LazyImport(message="Run 'pip install spacy'") as spacy_import:
+    import spacy
+    from spacy import Language as SpacyPipeline
+
+
+class _BackendEnumMeta(EnumMeta):
+    """
+    Metaclass for fine-grained error handling of backend enums.
+    """
+
+    def __call__(cls, value, names=None, *, module=None, qualname=None, type=None, start=1):
+        if names is None:
+            try:
+                return EnumMeta.__call__(cls, value, names, module=module, qualname=qualname, type=type, start=start)
+            except ValueError:
+                supported_backends = ", ".join(sorted(v.value for v in cls))  # pylint: disable=not-an-iterable
+                raise ComponentError(
+                    f"Invalid backend `{value}` for named entity extractor. "
+                    f"Supported backends: {supported_backends}"
+                )
+        else:
+            return EnumMeta.__call__(  # pylint: disable=too-many-function-args
+                cls, value, names, module, qualname, type, start
+            )
+
+
+class NamedEntityExtractorBackend(Enum, metaclass=_BackendEnumMeta):
+    """
+    NLP backend to use for Named Entity Recognition.
+    """
+
+    #: Hugging Face.
+    #:
+    #: Uses an Hugging Face model and pipeline.
+    HUGGING_FACE = "hugging_face"
+
+    #: spaCy.
+    #:
+    #: Uses a spaCy model and pipeline.
+    SPACY = "spacy"
+
+
+@dataclass
+class NamedEntityAnnotation:
+    """
+    Describes a single NER annotation.
+
+    :param entity:
+        Entity label.
+    :param start:
+        Start index of the entity in the document.
+    :param end:
+        End index of the entity in the document.
+    :param score:
+        Score calculated by the model.
+    """
+
+    entity: str
+    start: int
+    end: int
+    score: Optional[float] = None
+
+
+@component
+class NamedEntityExtractor:
+    _METADATA_KEY = "named_entities"
+
+    def __init__(
+        self, *, backend: Union[str, NamedEntityExtractorBackend], model_name_or_path: str, device_id: int = -1
+    ) -> None:
+        """
+        Construct a Named Entity extractor component.
+
+        :param backend:
+            Backend to use for NER.
+        :param model_name_or_path:
+            Name of the model or a path to the model on
+            the local disk.
+
+            Dependent on the backend.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+
+        if isinstance(backend, str):
+            backend = NamedEntityExtractorBackend(backend)
+
+        self._backend: _NerBackend
+        if backend == NamedEntityExtractorBackend.HUGGING_FACE:
+            self._backend = _HfBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        elif backend == NamedEntityExtractorBackend.SPACY:
+            self._backend = _SpacyBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        else:
+            raise ComponentError(f"Unknown NER backend '{type(backend).__name__}' for extractor")
+
+    def warm_up(self):
+        try:
+            self._backend.initialize()
+        except BaseException as e:
+            raise ComponentError(
+                f"Named entity extractor with backend '{self._backend.type} failed to initialize."
+            ) from e
+
+    @component.output_types(documents=List[Document])
+    def run(self, documents: List[Document], batch_size: int = 1) -> Dict[str, Any]:
+        texts = [doc.content if doc.content is not None else "" for doc in documents]
+        annotations = self._backend.annotate(texts, batch_size=batch_size)
+
+        if len(annotations) != len(documents):
+            raise ComponentError(
+                "NER backend did not return the correct number of annotations; "
+                f"got {len(annotations)} but expected {len(documents)}"
+            )
+
+        for doc, doc_annotations in zip(documents, annotations):
+            doc.meta[self._METADATA_KEY] = doc_annotations
+
+        return {"documents": documents}
+
+    def to_dict(self) -> Dict[str, Any]:
+        if isinstance(self._backend, _HfBackend):
+            backend_type = NamedEntityExtractorBackend.HUGGING_FACE
+        elif isinstance(self._backend, _SpacyBackend):
+            backend_type = NamedEntityExtractorBackend.SPACY
+        else:
+            # Unreachable.
+            raise ComponentError(f"Unknown backend '{type(self._backend).__name__}'")
+
+        return default_to_dict(
+            self, backend=backend_type, model_name_or_path=self._backend.model_name, device_id=self._backend.device_id
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "NamedEntityExtractor":
+        try:
+            return default_from_dict(cls, data)
+        except BaseException as e:
+            raise DeserializationError(f"Couldn't deserialize {cls.__name__} instance") from e
+
+    @property
+    def initialized(self) -> bool:
+        """
+        Returns if the extractor is ready to annotate text.
+        """
+        return self._backend.initialized
+
+    @classmethod
+    def get_stored_annotations(cls, document: Document) -> Optional[List[NamedEntityAnnotation]]:
+        """
+        Returns the document's named entity annotations stored
+        in its metadata, if any.
+
+        :param document:
+            Document whose annotations are to be fetched.
+        :returns:
+            The stored annotations.
+        """
+
+        return document.meta.get(cls._METADATA_KEY)
+
+
+class _NerBackend(ABC):
+    """
+    Base class for NER backends.
+    """
+
+    def __init__(self, type: NamedEntityExtractorBackend) -> None:
+        super().__init__()
+
+        self._type = type
+
+    @abstractmethod
+    def initialize(self):
+        """
+        Initializes the backend. This would usually
+        entail loading models, pipelines, etc.
+        """
+
+    @property
+    @abstractmethod
+    def initialized(self) -> bool:
+        """
+        Returns if the backend has been initialized, i.e,
+        ready to annotate text.
+        """
+
+    @abstractmethod
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        """
+        Predict annotations for a collection of documents.
+
+        :param texts:
+            Raw texts to be annotated.
+        :param batch_size:
+            Size of text batches that are
+            passed to the model.
+        :returns:
+            NER annotations.
+        """
+
+    @property
+    @abstractmethod
+    def model_name(self) -> str:
+        """
+        Returns the model name or path on the local disk.
+        """
+
+    @property
+    @abstractmethod
+    def device_id(self) -> int:
+        """
+        Returns the identifier of the device on which
+        the backend's model is loaded.
+        """
+
+    @property
+    def type(self) -> NamedEntityExtractorBackend:
+        """
+        Returns the type of the backend.
+        """
+        return self._type
+
+
+class _HfBackend(_NerBackend):
+    """
+    Hugging Face backend for NER.
+    """
+
+    def __init__(self, *, model_name_or_path: str, device_id: int) -> None:
+        """
+        Construct a Hugging Face NER backend.
+
+        :param model_name_or_path:
+            Name of the model or a path to the Hugging Face
+            model on the local disk.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+        super().__init__(NamedEntityExtractorBackend.HUGGING_FACE)
+
+        transformers_import.check()
+
+        self._model_name_or_path = model_name_or_path
+        self._device_id = device_id
+
+        self.tokenizer: Optional[AutoTokenizer] = None
+        self.model: Optional[AutoModelForTokenClassification] = None
+        self.pipeline: Optional[HfPipeline] = None
+
+    def initialize(self):
+        self.tokenizer = AutoTokenizer.from_pretrained(self._model_name_or_path)
+        self.model = AutoModelForTokenClassification.from_pretrained(self._model_name_or_path)
+        self.pipeline = pipeline(
path: haystack/components/extractors/named_entity_extractor.py
text: @user1:
We sometimes have users that come with a request "How come I can't pass this and this arg" to pipeline creation. Consider adding kwargs or perhaps HF has some standardized way to add other options to pipeline creation function

@user1:
I'm sure @user2 faced the same issue and has thought this through. Perhaps we just follow the same approach as we have in HuggingFaceLocalGenerator

@author:
Added a `pipeline_kwargs` parameter a la `HuggingFaceLocalGenerator`.

diff_hunk: @@ -0,0 +1,399 @@
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import Enum, EnumMeta
+from typing import Any, Dict, List, Optional, Union
+
+from ... import ComponentError, DeserializationError, Document, component, default_from_dict, default_to_dict
+from ...lazy_imports import LazyImport
+
+with LazyImport(message="Run 'pip install transformers'") as transformers_import:
+    from transformers import AutoModelForTokenClassification, AutoTokenizer
+    from transformers import Pipeline as HfPipeline
+    from transformers import pipeline
+
+with LazyImport(message="Run 'pip install spacy'") as spacy_import:
+    import spacy
+    from spacy import Language as SpacyPipeline
+
+
+class _BackendEnumMeta(EnumMeta):
+    """
+    Metaclass for fine-grained error handling of backend enums.
+    """
+
+    def __call__(cls, value, names=None, *, module=None, qualname=None, type=None, start=1):
+        if names is None:
+            try:
+                return EnumMeta.__call__(cls, value, names, module=module, qualname=qualname, type=type, start=start)
+            except ValueError:
+                supported_backends = ", ".join(sorted(v.value for v in cls))  # pylint: disable=not-an-iterable
+                raise ComponentError(
+                    f"Invalid backend `{value}` for named entity extractor. "
+                    f"Supported backends: {supported_backends}"
+                )
+        else:
+            return EnumMeta.__call__(  # pylint: disable=too-many-function-args
+                cls, value, names, module, qualname, type, start
+            )
+
+
+class NamedEntityExtractorBackend(Enum, metaclass=_BackendEnumMeta):
+    """
+    NLP backend to use for Named Entity Recognition.
+    """
+
+    #: Hugging Face.
+    #:
+    #: Uses an Hugging Face model and pipeline.
+    HUGGING_FACE = "hugging_face"
+
+    #: spaCy.
+    #:
+    #: Uses a spaCy model and pipeline.
+    SPACY = "spacy"
+
+
+@dataclass
+class NamedEntityAnnotation:
+    """
+    Describes a single NER annotation.
+
+    :param entity:
+        Entity label.
+    :param start:
+        Start index of the entity in the document.
+    :param end:
+        End index of the entity in the document.
+    :param score:
+        Score calculated by the model.
+    """
+
+    entity: str
+    start: int
+    end: int
+    score: Optional[float] = None
+
+
+@component
+class NamedEntityExtractor:
+    _METADATA_KEY = "named_entities"
+
+    def __init__(
+        self, *, backend: Union[str, NamedEntityExtractorBackend], model_name_or_path: str, device_id: int = -1
+    ) -> None:
+        """
+        Construct a Named Entity extractor component.
+
+        :param backend:
+            Backend to use for NER.
+        :param model_name_or_path:
+            Name of the model or a path to the model on
+            the local disk.
+
+            Dependent on the backend.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+
+        if isinstance(backend, str):
+            backend = NamedEntityExtractorBackend(backend)
+
+        self._backend: _NerBackend
+        if backend == NamedEntityExtractorBackend.HUGGING_FACE:
+            self._backend = _HfBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        elif backend == NamedEntityExtractorBackend.SPACY:
+            self._backend = _SpacyBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        else:
+            raise ComponentError(f"Unknown NER backend '{type(backend).__name__}' for extractor")
+
+    def warm_up(self):
+        try:
+            self._backend.initialize()
+        except BaseException as e:
+            raise ComponentError(
+                f"Named entity extractor with backend '{self._backend.type} failed to initialize."
+            ) from e
+
+    @component.output_types(documents=List[Document])
+    def run(self, documents: List[Document], batch_size: int = 1) -> Dict[str, Any]:
+        texts = [doc.content if doc.content is not None else "" for doc in documents]
+        annotations = self._backend.annotate(texts, batch_size=batch_size)
+
+        if len(annotations) != len(documents):
+            raise ComponentError(
+                "NER backend did not return the correct number of annotations; "
+                f"got {len(annotations)} but expected {len(documents)}"
+            )
+
+        for doc, doc_annotations in zip(documents, annotations):
+            doc.meta[self._METADATA_KEY] = doc_annotations
+
+        return {"documents": documents}
+
+    def to_dict(self) -> Dict[str, Any]:
+        if isinstance(self._backend, _HfBackend):
+            backend_type = NamedEntityExtractorBackend.HUGGING_FACE
+        elif isinstance(self._backend, _SpacyBackend):
+            backend_type = NamedEntityExtractorBackend.SPACY
+        else:
+            # Unreachable.
+            raise ComponentError(f"Unknown backend '{type(self._backend).__name__}'")
+
+        return default_to_dict(
+            self, backend=backend_type, model_name_or_path=self._backend.model_name, device_id=self._backend.device_id
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "NamedEntityExtractor":
+        try:
+            return default_from_dict(cls, data)
+        except BaseException as e:
+            raise DeserializationError(f"Couldn't deserialize {cls.__name__} instance") from e
+
+    @property
+    def initialized(self) -> bool:
+        """
+        Returns if the extractor is ready to annotate text.
+        """
+        return self._backend.initialized
+
+    @classmethod
+    def get_stored_annotations(cls, document: Document) -> Optional[List[NamedEntityAnnotation]]:
+        """
+        Returns the document's named entity annotations stored
+        in its metadata, if any.
+
+        :param document:
+            Document whose annotations are to be fetched.
+        :returns:
+            The stored annotations.
+        """
+
+        return document.meta.get(cls._METADATA_KEY)
+
+
+class _NerBackend(ABC):
+    """
+    Base class for NER backends.
+    """
+
+    def __init__(self, type: NamedEntityExtractorBackend) -> None:
+        super().__init__()
+
+        self._type = type
+
+    @abstractmethod
+    def initialize(self):
+        """
+        Initializes the backend. This would usually
+        entail loading models, pipelines, etc.
+        """
+
+    @property
+    @abstractmethod
+    def initialized(self) -> bool:
+        """
+        Returns if the backend has been initialized, i.e,
+        ready to annotate text.
+        """
+
+    @abstractmethod
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        """
+        Predict annotations for a collection of documents.
+
+        :param texts:
+            Raw texts to be annotated.
+        :param batch_size:
+            Size of text batches that are
+            passed to the model.
+        :returns:
+            NER annotations.
+        """
+
+    @property
+    @abstractmethod
+    def model_name(self) -> str:
+        """
+        Returns the model name or path on the local disk.
+        """
+
+    @property
+    @abstractmethod
+    def device_id(self) -> int:
+        """
+        Returns the identifier of the device on which
+        the backend's model is loaded.
+        """
+
+    @property
+    def type(self) -> NamedEntityExtractorBackend:
+        """
+        Returns the type of the backend.
+        """
+        return self._type
+
+
+class _HfBackend(_NerBackend):
+    """
+    Hugging Face backend for NER.
+    """
+
+    def __init__(self, *, model_name_or_path: str, device_id: int) -> None:
+        """
+        Construct a Hugging Face NER backend.
+
+        :param model_name_or_path:
+            Name of the model or a path to the Hugging Face
+            model on the local disk.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+        super().__init__(NamedEntityExtractorBackend.HUGGING_FACE)
+
+        transformers_import.check()
+
+        self._model_name_or_path = model_name_or_path
+        self._device_id = device_id
+
+        self.tokenizer: Optional[AutoTokenizer] = None
+        self.model: Optional[AutoModelForTokenClassification] = None
+        self.pipeline: Optional[HfPipeline] = None
+
+    def initialize(self):
+        self.tokenizer = AutoTokenizer.from_pretrained(self._model_name_or_path)
+        self.model = AutoModelForTokenClassification.from_pretrained(self._model_name_or_path)
+        self.pipeline = pipeline(
+            "ner", model=self.model, tokenizer=self.tokenizer, aggregation_strategy="simple", device=self.device_id
+        )
+
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        if not self.initialized:
+            raise ComponentError("Hugging Face NER backend was not initialized")
+
+        assert self.pipeline is not None
+        outputs = self.pipeline(texts, batch_size=batch_size)
+        return [
+            [
+                NamedEntityAnnotation(
+                    entity=annotation["entity"] if "entity" in annotation else annotation["entity_group"],
+                    start=annotation["start"],
+                    end=annotation["end"],
+                    score=annotation["score"],
+                )
+                for annotation in annotations
+            ]
+            for annotations in outputs
+        ]
+
+    @property
+    def initialized(self) -> bool:
+        return self.tokenizer is not None and self.model is not None or self.pipeline is not None
+
+    @property
+    def model_name(self) -> str:
+        return self._model_name_or_path
+
+    @property
+    def device_id(self) -> int:
+        return self._device_id
+
+
+class _SpacyBackend(_NerBackend):
+    """
+    spaCy backend for NER.
+    """
+
+    def __init__(self, *, model_name_or_path: str, device_id: int) -> None:
+        """
+        Construct a spaCy NER backend.
+
+        :param model_name_or_path:
+            Name of the model or a path to the spaCy
+            model on the local disk.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+        super().__init__(NamedEntityExtractorBackend.SPACY)
+
+        spacy_import.check()
+
+        self._model_name_or_path = model_name_or_path
+        self._device_id = device_id
+
+        self.pipeline: Optional[SpacyPipeline] = None
+
+    def initialize(self):
+        # We need to initialize the model on the GPU if needed.
+        with self._select_device():
+            self.pipeline = spacy.load(self._model_name_or_path)
+
+        if not self.pipeline.has_pipe("ner"):
+            raise ComponentError(f"spaCy pipeline '{self._model_name_or_path}' does not contain an NER component")
+
+        # Disable unnecessary pipes.
+        pipes_to_keep = ("ner", "tok2vec", "transformer", "curated_transformer")
+        for name in self.pipeline.pipe_names:
+            if name not in pipes_to_keep:
+                self.pipeline.disable_pipe(name)
+
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        if not self.initialized:
+            raise ComponentError("spaCy NER backend was not initialized")
path: haystack/components/extractors/named_entity_extractor.py
text: @user1:
Also, please call warm_up etc etc

@author:
Fixed.

diff_hunk: @@ -0,0 +1,399 @@
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import Enum, EnumMeta
+from typing import Any, Dict, List, Optional, Union
+
+from ... import ComponentError, DeserializationError, Document, component, default_from_dict, default_to_dict
+from ...lazy_imports import LazyImport
+
+with LazyImport(message="Run 'pip install transformers'") as transformers_import:
+    from transformers import AutoModelForTokenClassification, AutoTokenizer
+    from transformers import Pipeline as HfPipeline
+    from transformers import pipeline
+
+with LazyImport(message="Run 'pip install spacy'") as spacy_import:
+    import spacy
+    from spacy import Language as SpacyPipeline
+
+
+class _BackendEnumMeta(EnumMeta):
+    """
+    Metaclass for fine-grained error handling of backend enums.
+    """
+
+    def __call__(cls, value, names=None, *, module=None, qualname=None, type=None, start=1):
+        if names is None:
+            try:
+                return EnumMeta.__call__(cls, value, names, module=module, qualname=qualname, type=type, start=start)
+            except ValueError:
+                supported_backends = ", ".join(sorted(v.value for v in cls))  # pylint: disable=not-an-iterable
+                raise ComponentError(
+                    f"Invalid backend `{value}` for named entity extractor. "
+                    f"Supported backends: {supported_backends}"
+                )
+        else:
+            return EnumMeta.__call__(  # pylint: disable=too-many-function-args
+                cls, value, names, module, qualname, type, start
+            )
+
+
+class NamedEntityExtractorBackend(Enum, metaclass=_BackendEnumMeta):
+    """
+    NLP backend to use for Named Entity Recognition.
+    """
+
+    #: Hugging Face.
+    #:
+    #: Uses an Hugging Face model and pipeline.
+    HUGGING_FACE = "hugging_face"
+
+    #: spaCy.
+    #:
+    #: Uses a spaCy model and pipeline.
+    SPACY = "spacy"
+
+
+@dataclass
+class NamedEntityAnnotation:
+    """
+    Describes a single NER annotation.
+
+    :param entity:
+        Entity label.
+    :param start:
+        Start index of the entity in the document.
+    :param end:
+        End index of the entity in the document.
+    :param score:
+        Score calculated by the model.
+    """
+
+    entity: str
+    start: int
+    end: int
+    score: Optional[float] = None
+
+
+@component
+class NamedEntityExtractor:
+    _METADATA_KEY = "named_entities"
+
+    def __init__(
+        self, *, backend: Union[str, NamedEntityExtractorBackend], model_name_or_path: str, device_id: int = -1
+    ) -> None:
+        """
+        Construct a Named Entity extractor component.
+
+        :param backend:
+            Backend to use for NER.
+        :param model_name_or_path:
+            Name of the model or a path to the model on
+            the local disk.
+
+            Dependent on the backend.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+
+        if isinstance(backend, str):
+            backend = NamedEntityExtractorBackend(backend)
+
+        self._backend: _NerBackend
+        if backend == NamedEntityExtractorBackend.HUGGING_FACE:
+            self._backend = _HfBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        elif backend == NamedEntityExtractorBackend.SPACY:
+            self._backend = _SpacyBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        else:
+            raise ComponentError(f"Unknown NER backend '{type(backend).__name__}' for extractor")
+
+    def warm_up(self):
+        try:
+            self._backend.initialize()
+        except BaseException as e:
+            raise ComponentError(
+                f"Named entity extractor with backend '{self._backend.type} failed to initialize."
+            ) from e
+
+    @component.output_types(documents=List[Document])
+    def run(self, documents: List[Document], batch_size: int = 1) -> Dict[str, Any]:
+        texts = [doc.content if doc.content is not None else "" for doc in documents]
+        annotations = self._backend.annotate(texts, batch_size=batch_size)
+
+        if len(annotations) != len(documents):
+            raise ComponentError(
+                "NER backend did not return the correct number of annotations; "
+                f"got {len(annotations)} but expected {len(documents)}"
+            )
+
+        for doc, doc_annotations in zip(documents, annotations):
+            doc.meta[self._METADATA_KEY] = doc_annotations
+
+        return {"documents": documents}
+
+    def to_dict(self) -> Dict[str, Any]:
+        if isinstance(self._backend, _HfBackend):
+            backend_type = NamedEntityExtractorBackend.HUGGING_FACE
+        elif isinstance(self._backend, _SpacyBackend):
+            backend_type = NamedEntityExtractorBackend.SPACY
+        else:
+            # Unreachable.
+            raise ComponentError(f"Unknown backend '{type(self._backend).__name__}'")
+
+        return default_to_dict(
+            self, backend=backend_type, model_name_or_path=self._backend.model_name, device_id=self._backend.device_id
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "NamedEntityExtractor":
+        try:
+            return default_from_dict(cls, data)
+        except BaseException as e:
+            raise DeserializationError(f"Couldn't deserialize {cls.__name__} instance") from e
+
+    @property
+    def initialized(self) -> bool:
+        """
+        Returns if the extractor is ready to annotate text.
+        """
+        return self._backend.initialized
+
+    @classmethod
+    def get_stored_annotations(cls, document: Document) -> Optional[List[NamedEntityAnnotation]]:
+        """
+        Returns the document's named entity annotations stored
+        in its metadata, if any.
+
+        :param document:
+            Document whose annotations are to be fetched.
+        :returns:
+            The stored annotations.
+        """
+
+        return document.meta.get(cls._METADATA_KEY)
+
+
+class _NerBackend(ABC):
+    """
+    Base class for NER backends.
+    """
+
+    def __init__(self, type: NamedEntityExtractorBackend) -> None:
+        super().__init__()
+
+        self._type = type
+
+    @abstractmethod
+    def initialize(self):
+        """
+        Initializes the backend. This would usually
+        entail loading models, pipelines, etc.
+        """
+
+    @property
+    @abstractmethod
+    def initialized(self) -> bool:
+        """
+        Returns if the backend has been initialized, i.e,
+        ready to annotate text.
+        """
+
+    @abstractmethod
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        """
+        Predict annotations for a collection of documents.
+
+        :param texts:
+            Raw texts to be annotated.
+        :param batch_size:
+            Size of text batches that are
+            passed to the model.
+        :returns:
+            NER annotations.
+        """
+
+    @property
+    @abstractmethod
+    def model_name(self) -> str:
+        """
+        Returns the model name or path on the local disk.
+        """
+
+    @property
+    @abstractmethod
+    def device_id(self) -> int:
+        """
+        Returns the identifier of the device on which
+        the backend's model is loaded.
+        """
+
+    @property
+    def type(self) -> NamedEntityExtractorBackend:
+        """
+        Returns the type of the backend.
+        """
+        return self._type
+
+
+class _HfBackend(_NerBackend):
+    """
+    Hugging Face backend for NER.
+    """
+
+    def __init__(self, *, model_name_or_path: str, device_id: int) -> None:
+        """
+        Construct a Hugging Face NER backend.
+
+        :param model_name_or_path:
+            Name of the model or a path to the Hugging Face
+            model on the local disk.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+        super().__init__(NamedEntityExtractorBackend.HUGGING_FACE)
+
+        transformers_import.check()
+
+        self._model_name_or_path = model_name_or_path
+        self._device_id = device_id
+
+        self.tokenizer: Optional[AutoTokenizer] = None
+        self.model: Optional[AutoModelForTokenClassification] = None
+        self.pipeline: Optional[HfPipeline] = None
+
+    def initialize(self):
+        self.tokenizer = AutoTokenizer.from_pretrained(self._model_name_or_path)
+        self.model = AutoModelForTokenClassification.from_pretrained(self._model_name_or_path)
+        self.pipeline = pipeline(
+            "ner", model=self.model, tokenizer=self.tokenizer, aggregation_strategy="simple", device=self.device_id
+        )
+
+    def annotate(self, texts: List[str], *, batch_size: int = 1) -> List[List[NamedEntityAnnotation]]:
+        if not self.initialized:
+            raise ComponentError("Hugging Face NER backend was not initialized")
path: haystack/components/extractors/named_entity_extractor.py
text: @user1:
Perhaps, let the user know they need to invoke warm_up?

@author:
Expanded the error message to mention `warm_up`.

diff_hunk: @@ -0,0 +1,399 @@
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import Enum, EnumMeta
+from typing import Any, Dict, List, Optional, Union
+
+from ... import ComponentError, DeserializationError, Document, component, default_from_dict, default_to_dict
+from ...lazy_imports import LazyImport
+
+with LazyImport(message="Run 'pip install transformers'") as transformers_import:
+    from transformers import AutoModelForTokenClassification, AutoTokenizer
+    from transformers import Pipeline as HfPipeline
+    from transformers import pipeline
+
+with LazyImport(message="Run 'pip install spacy'") as spacy_import:
+    import spacy
+    from spacy import Language as SpacyPipeline
+
+
+class _BackendEnumMeta(EnumMeta):
+    """
+    Metaclass for fine-grained error handling of backend enums.
+    """
+
+    def __call__(cls, value, names=None, *, module=None, qualname=None, type=None, start=1):
+        if names is None:
+            try:
+                return EnumMeta.__call__(cls, value, names, module=module, qualname=qualname, type=type, start=start)
+            except ValueError:
+                supported_backends = ", ".join(sorted(v.value for v in cls))  # pylint: disable=not-an-iterable
+                raise ComponentError(
+                    f"Invalid backend `{value}` for named entity extractor. "
+                    f"Supported backends: {supported_backends}"
+                )
+        else:
+            return EnumMeta.__call__(  # pylint: disable=too-many-function-args
+                cls, value, names, module, qualname, type, start
+            )
+
+
+class NamedEntityExtractorBackend(Enum, metaclass=_BackendEnumMeta):
+    """
+    NLP backend to use for Named Entity Recognition.
+    """
+
+    #: Hugging Face.
+    #:
+    #: Uses an Hugging Face model and pipeline.
+    HUGGING_FACE = "hugging_face"
+
+    #: spaCy.
+    #:
+    #: Uses a spaCy model and pipeline.
+    SPACY = "spacy"
+
+
+@dataclass
+class NamedEntityAnnotation:
+    """
+    Describes a single NER annotation.
+
+    :param entity:
+        Entity label.
+    :param start:
+        Start index of the entity in the document.
+    :param end:
+        End index of the entity in the document.
+    :param score:
+        Score calculated by the model.
+    """
+
+    entity: str
+    start: int
+    end: int
+    score: Optional[float] = None
+
+
+@component
+class NamedEntityExtractor:
+    _METADATA_KEY = "named_entities"
+
+    def __init__(
+        self, *, backend: Union[str, NamedEntityExtractorBackend], model_name_or_path: str, device_id: int = -1
+    ) -> None:
+        """
+        Construct a Named Entity extractor component.
+
+        :param backend:
+            Backend to use for NER.
+        :param model_name_or_path:
+            Name of the model or a path to the model on
+            the local disk.
+
+            Dependent on the backend.
+        :param device_id:
+            Identifier of the device on which the backend
+            is executed.
+
+            To execute on the CPU, pass a value of `-1`.
+            To execute on the GPU, pass the GPU identifier.
+        """
+
+        if isinstance(backend, str):
+            backend = NamedEntityExtractorBackend(backend)
+
+        self._backend: _NerBackend
+        if backend == NamedEntityExtractorBackend.HUGGING_FACE:
+            self._backend = _HfBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        elif backend == NamedEntityExtractorBackend.SPACY:
+            self._backend = _SpacyBackend(model_name_or_path=model_name_or_path, device_id=device_id)
+        else:
+            raise ComponentError(f"Unknown NER backend '{type(backend).__name__}' for extractor")
+
+    def warm_up(self):
+        try:
+            self._backend.initialize()
+        except BaseException as e:
path: haystack/components/extractors/named_entity_extractor.py
text: @user1:
We usually catch simply `Exception` here, although it really doesn't matter BaseException includes SystemExit, KeyboardInterrupt, and GeneratorExit. 

@author:
Fixed.

diff_hunk: @@ -0,0 +1,3 @@
+from .named_entity_extractor import NamedEntityAnnotation, NamedEntityExtractor, NamedEntityExtractorBackend
path: haystack/components/extractors/__init__.py
text: @user1:
We usually use the full absolute path here instead of this relative.

@author:
Switched to absolute paths. Is this a question of style or function? I have rarely come across libraries using absolute paths in submodules.

@user2:
It's about style, developed after our linting tools more than after an explicit decision we made. We initially banned the usage of complex relative imports through flake8/ruff, and while relative imports to siblings are accepted in PEP8 (like the one here), to avoid explaining the difference we just went with "always absolute". If we find the time to define a better style (e.g. admitting relative imports to siblings) and configure our tooling accordingly, I would be supportive of the initiative!