GliNER

Bases: Engine[PromptSignature, Result, Model, InferenceMode]

Source code in sieves/engines/glix_.py

class GliX(Engine[PromptSignature, Result, Model, InferenceMode]):
    def __init__(
        self,
        model: Model,
        init_kwargs: dict[str, Any] | None = None,
        inference_kwargs: dict[str, Any] | None = None,
        strict_mode: bool = False,
        batch_size: int = -1,
    ):
        super().__init__(model, init_kwargs, inference_kwargs, strict_mode, batch_size)
        self._model_wrappers: dict[InferenceMode, gliner.multitask.base.GLiNERBasePipeline] = {}

    @property
    def inference_modes(self) -> type[InferenceMode]:
        return InferenceMode

    @property
    def supports_few_shotting(self) -> bool:
        return False

    def build_executable(
        self,
        inference_mode: InferenceMode,
        prompt_template: str | None,
        prompt_signature: type[PromptSignature] | PromptSignature,
        fewshot_examples: Iterable[pydantic.BaseModel] = (),
    ) -> Executable[Result]:
        assert isinstance(prompt_signature, list)
        cls_name = self.__class__.__name__
        if len(list(fewshot_examples)):
            warnings.warn(f"Few-shot examples are not supported by engine {cls_name}.")

        # Lazily initialize multi-task wrapper for underlying GliNER model.
        if inference_mode not in self._model_wrappers:
            self._model_wrappers[inference_mode] = inference_mode.value(model=self._model)
        model = self._model_wrappers[inference_mode]

        # Overwrite prompt default template, if template specified. Note that this is a static prompt and GliX doesn't
        # do few-shotting, so we don't inject anything into the template.
        if prompt_template:
            self._model.prompt = jinja2.Template(prompt_template).render()

        def execute(values: Iterable[dict[str, Any]]) -> Iterable[Result]:
            """Execute prompts with engine for given values.
            :param values: Values to inject into prompts.
            :return Iterable[Result]: Results for prompts.
            """
            try:
                params = {
                    InferenceMode.classification: {"classes": prompt_signature, "multi_label": True},
                    InferenceMode.question_answering: {"questions": prompt_signature},
                    InferenceMode.summarization: {},
                }[inference_mode]
            except KeyError:
                raise ValueError(f"Inference mode {inference_mode} not supported by {cls_name} engine.")

            batch_size = self._batch_size if self._batch_size != -1 else sys.maxsize
            # Ensure values are read as generator for standardized batch handling (otherwise we'd have to use
            # different batch handling depending on whether lists/tuples or generators are used).
            values = (v for v in values)

            while batch := [vals["text"] for vals in itertools.islice(values, batch_size)]:
                if len(batch) == 0:
                    break

                assert isinstance(params, dict)
                yield from model(batch, **(params | self._inference_kwargs))

        return execute

_attributes property

Returns attributes to serialize.

Returns:

Type	Description
dict[str, Attribute]	Dict of attributes to serialize.

model property

Return model instance.

Returns:

Type	Description
EngineModel	Model instance.

_convert_fewshot_examples(fewshot_examples) staticmethod

Convert fewshot examples from pydantic.BaseModel instance to dicts.

Parameters:

Name	Type	Description	Default
fewshot_examples	Iterable[BaseModel]	Fewshot examples to convert.	required

Returns:

Type	Description
list[dict[str, Any]]	Fewshot examples as dicts.

Source code in sieves/engines/core.py

@staticmethod
def _convert_fewshot_examples(fewshot_examples: Iterable[pydantic.BaseModel]) -> list[dict[str, Any]]:
    """
    Convert fewshot examples from pydantic.BaseModel instance to dicts.
    :param fewshot_examples: Fewshot examples to convert.
    :return: Fewshot examples as dicts.
    """
    return [fs_example.model_dump(serialize_as_any=True) for fs_example in fewshot_examples]

_execute_async_calls(calls) async staticmethod

Executes batch of async functions.

Parameters:

Name	Type	Description	Default
calls	list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]	Async calls to execute.	required

Returns:

Type	Description
Any	Parsed response objects.

Source code in sieves/engines/core.py

@staticmethod
async def _execute_async_calls(calls: list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]) -> Any:
    """Executes batch of async functions.
    :param calls: Async calls to execute.
    :return: Parsed response objects.
    """
    return await asyncio.gather(*calls)

_validate_batch_size(batch_size)

Validates batch_size. Noop by default.

Parameters:

Name	Type	Description	Default
batch_size	int	Specified batch size.	required

Returns:

Type	Description
int	Validated batch size.

Source code in sieves/engines/core.py

def _validate_batch_size(self, batch_size: int) -> int:
    """Validates batch_size. Noop by default.
    :param batch_size: Specified batch size.
    :returns int: Validated batch size.
    """
    return batch_size

deserialize(config, **kwargs) classmethod

Generate Engine instance from config.

Parameters:

Name	Type	Description	Default
config	Config	Config to generate instance from.	required
kwargs	dict[str, Any]	Values to inject into loaded config.	{}

Returns:

Type	Description
Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]	Deserialized Engine instance.

Source code in sieves/engines/core.py

@classmethod
def deserialize(
    cls, config: Config, **kwargs: dict[str, Any]
) -> Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]:
    """Generate Engine instance from config.
    :param config: Config to generate instance from.
    :param kwargs: Values to inject into loaded config.
    :return: Deserialized Engine instance.
    """
    return cls(**config.to_init_dict(cls, **kwargs))

serialize()

Serializes engine.

Returns:

Type	Description
Config	Config instance.

Source code in sieves/engines/core.py

def serialize(self) -> Config:
    """Serializes engine.
    :return: Config instance.
    """
    return Config.create(self.__class__, self._attributes)