unitxt.operators module¶

This section describes unitxt operators.

Operators: Building Blocks of Unitxt Processing Pipelines¶

Within the Unitxt framework, operators serve as the foundational elements used to assemble processing pipelines. Each operator is designed to perform specific manipulations on dictionary structures within a stream. These operators are callable entities that receive a MultiStream as input. The output is a MultiStream, augmented with the operator’s manipulations, which are then systematically applied to each instance in the stream when pulled.

Creating Custom Operators¶

To enhance the functionality of Unitxt, users are encouraged to develop custom operators. This can be achieved by inheriting from any of the existing operators listed below or from one of the fundamental base operators. The primary task in any operator development is to implement the process function, which defines the unique manipulations the operator will perform.

General or Specialized Operators¶

Some operators are specialized in specific data or specific operations such as:

loaders for accessing data from various sources.
splitters for fixing data splits.
stream_operators for changing joining and mixing streams.
struct_data_operators for structured data operators.
collections_operators for handling collections such as lists and dictionaries.
dialog_operators for handling dialogs.
string_operators for handling strings.
span_labeling_operators for handling strings.
fusion for fusing and mixing datasets.

Other specialized operators are used by unitxt internally:

templates for verbalizing data examples.
formats for preparing data for models.

The rest of this section is dedicated to general operators.

General Operators List:¶

class unitxt.operators.AddConstant(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, add: Any = __required__)[source]¶

Bases: FieldOperator

Adds a constant, being argument ‘add’, to the processed value.

Parameters:: add – the constant to add.

class unitxt.operators.AddID(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, id_field_name: str = 'id')[source]¶

Bases: InstanceOperator

Stores a unique id value in the designated ‘id_field_name’ field of the given instance.

class unitxt.operators.AddIncrementalId(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, to_field: str = __required__)[source]¶: Bases: StreamOperator

class unitxt.operators.Apply(data_classification_policy: List[str] = None, _requirements_list: Union[List[str], Dict[str, str]] = [], requirements: Union[List[str], Dict[str, str]] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, function: Callable = <class 'unitxt.dataclass.Undefined'>, _argv: tuple = (), _kwargs: dict = {}, to_field: str = <class 'unitxt.dataclass.Undefined'>)[source]¶

Bases: InstanceOperator

A class used to apply a python function and store the result in a field.

Parameters:

function (str) – name of function.
to_field (str) – the field to store the result

any additional arguments are field names whose values will be passed directly to the function specified

Examples: Store in field “b” the uppercase string of the value in field “a”: Apply("a", function=str.upper, to_field="b")

Dump the json representation of field “t” and store back in the same field: Apply("t", function=json.dumps, to_field="t")

Set the time in a field ‘b’: Apply(function=time.time, to_field="b")

class unitxt.operators.ApplyMetric(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, metric_field: str = __required__, calc_confidence_intervals: bool = __required__)[source]¶

Bases: StreamOperator, ArtifactFetcherMixin

Applies metric operators to a stream based on a metric field specified in each instance.

Parameters:

metric_field (str) – The field containing the metrics to be applied.
calc_confidence_intervals (bool) – Whether the applied metric should calculate confidence intervals or not.

class unitxt.operators.ApplyOperatorsField(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, operators_field: str = __required__, default_operators: List[str] = None)[source]¶

Bases: InstanceOperator

Applies value operators to each instance in a stream based on specified fields.

Parameters:

operators_field (str) – name of the field that contains a single name, or a list of names, of the operators to be applied, one after the other, for the processing of the instance. Each operator is equipped with ‘process_instance()’ method.
default_operators (List[str]) – A list of default operators to be used if no operators are found in the instance.

Example

when instance {“prediction”: 111, “references”: [222, 333] , “c”: [“processors.to_string”, “processors.first_character”]} is processed by operator (please look up the catalog that these operators, they are tuned to process fields “prediction” and “references”): operator = ApplyOperatorsField(operators_field=”c”), the resulting instance is: {“prediction”: “1”, “references”: [“2”, “3”], “c”: [“processors.to_string”, “processors.first_character”]}

class unitxt.operators.ApplyStreamOperatorsField(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str = __required__, reversed: bool = False)[source]¶

Bases: StreamOperator, ArtifactFetcherMixin

Applies stream operators to a stream based on specified fields in each instance.

Parameters:

field (str) – The field containing the operators to be applied.
reversed (bool) – Whether to apply the operators in reverse order.

class unitxt.operators.ArtifactFetcherMixin[source]¶

Bases: object

Provides a way to fetch and cache artifacts in the system.

Parameters:: cache (Dict[str, Artifact]) – A cache for storing fetched artifacts.

classmethod get_artifact(artifact_identifier: str) → Artifact[source]¶

class unitxt.operators.Balance(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] | NoneType = None, dont_apply_to_streams: List[str] = None, max_instances: int = None, fields: List[str] = __required__)[source]¶

Bases: StreamRefiner

A class used to balance streams deterministically.

For each instance, a signature is constructed from the values of the instance in specified input ‘fields’. By discarding instances from the input stream, DeterministicBalancer maintains equal number of instances for all signatures. When also input ‘max_instances’ is specified, DeterministicBalancer maintains a total instance count not exceeding ‘max_instances’. The total number of discarded instances is as few as possible.

Parameters:

fields (List[str]) – A list of field names to be used in producing the instance’s signature.
max_instances (Optional, int) – overall max.

Usage:: balancer = DeterministicBalancer(fields=["field1", "field2"], max_instances=200) balanced_stream = balancer.process(stream)

Example

When input [{"a": 1, "b": 1},{"a": 1, "b": 2},{"a": 2},{"a": 3},{"a": 4}] is fed into DeterministicBalancer(fields=["a"]) the resulting stream will be: [{"a": 1, "b": 1},{"a": 2},{"a": 3},{"a": 4}]

class unitxt.operators.BytesToString(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶: Bases: FieldOperator

class unitxt.operators.Cast(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, to: str = __required__, failure_default: Any | NoneType = '__UNDEFINED__')[source]¶

Bases: FieldOperator

Casts specified fields to specified types.

Parameters:

default (object) – A dictionary mapping field names to default values for cases of casting failure.
process_every_value (bool) – If true, all fields involved must contain lists, and each value in the list is then casted. Defaults to False.

class unitxt.operators.CastFields(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: Dict[str, str] = {}, failure_defaults: Dict[str, object] = {}, use_nested_query: bool = None, process_every_value: bool = False)[source]¶

Bases: InstanceOperator

Casts specified fields to specified types.

Parameters:

fields (Dict[str, str]) – A dictionary mapping field names to the names of the types to cast the fields to. e.g: “int”, “str”, “float”, “bool”. Basic names of types
defaults (Dict[str, object]) – A dictionary mapping field names to default values for cases of casting failure.
process_every_value (bool) – If true, all fields involved must contain lists, and each value in the list is then casted. Defaults to False.

Example

CastFields(
    fields={"a/d": "float", "b": "int"},
    failure_defaults={"a/d": 0.0, "b": 0},
    process_every_value=True,
)

would process the input instance: {"a": {"d": ["half", "0.6", 1, 12]}, "b": ["2"]} into {"a": {"d": [0.0, 0.6, 1.0, 12.0]}, "b": [2]}.

class unitxt.operators.CollateInstances(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, batch_size: int = __required__)[source]¶

Bases: StreamOperator

Operator which collates values from multiple instances to a single instance.

Each field becomes the list of values of corresponding field of collated batch_size of instances.

batch_size¶

Type:: int

Example

CollateInstances(batch_size=2)

Given inputs = [
    {"a": 1, "b": 2},
    {"a": 2, "b": 2},
    {"a": 3, "b": 2},
    {"a": 4, "b": 2},
    {"a": 5, "b": 2}
]

Returns targets = [
    {"a": [1,2], "b": [2,2]},
    {"a": [3,4], "b": [2,2]},
    {"a": [5], "b": [2]},
]

class unitxt.operators.CollateInstancesByField(data_classification_policy: List[str] = None, _requirements_list: Union[List[str], Dict[str, str]] = [], requirements: Union[List[str], Dict[str, str]] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, by_field: str = <class 'unitxt.dataclass.Undefined'>, aggregate_fields: List[str] = <class 'unitxt.dataclass.Undefined'>)[source]¶

Bases: StreamOperator

Groups a list of instances by a specified field, aggregates specified fields into lists, and ensures consistency for all other non-aggregated fields.

Parameters:

str (by_field) – the name of the field to group data by.
list (aggregate_fields) – the field names to aggregate into lists.

Returns:

A stream of instances grouped and aggregated by the specified field.

Raises:

UnitxtError – If non-aggregate fields have inconsistent values.

Example

Collate the instances based on field “category” and aggregate fields “value” and “id”.

CollateInstancesByField(by_field="category", aggregate_fields=["value", "id"])

given input:
[
    {"id": 1, "category": "A", "value": 10", "flag" : True},
    {"id": 2, "category": "B", "value": 20", "flag" : False},
    {"id": 3, "category": "A", "value": 30", "flag" : True},
    {"id": 4, "category": "B", "value": 40", "flag" : False}
]

the output is:
[
    {"category": "A", "id": [1, 3], "value": [10, 30], "info": True},
    {"category": "B", "id": [2, 4], "value": [20, 40], "info": False}
]

Note that the “flag” field is not aggregated, and must be the same in all instances in the same category, or an error is raised.

class unitxt.operators.ComputeExpressionMixin(data_classification_policy: List[str] = None, expression: str = __required__, imports_list: List[str] = [])[source]¶

Bases: Artifact

Computes an expression expressed over fields of an instance.

Parameters:

expression (str) – the expression, in terms of names of fields of an instance
imports_list (List[str]) – list of names of imports needed for the evaluation of the expression

class unitxt.operators.Copy(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶

Bases: FieldOperator

Copies values from specified fields to specified fields.

Args (of parent class):: field_to_field (Union[List[List], Dict[str, str]]): A list of lists, where each sublist contains the source field and the destination field, or a dictionary mapping source fields to destination fields.

Examples

An input instance {“a”: 2, “b”: 3}, when processed by Copy(field_to_field={"a": "b"}) would yield {“a”: 2, “b”: 2}, and when processed by Copy(field_to_field={"a": "c"}) would yield {“a”: 2, “b”: 3, “c”: 2}

with field names containing / , we can also copy inside the field: Copy(field="a/0",to_field="a") would process instance {“a”: [1, 3]} into {“a”: 1}

class unitxt.operators.Deduplicate(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, by: List[str] = __required__)[source]¶

Bases: StreamOperator

Deduplicate the stream based on the given fields.

Parameters:: by (List[str]) – A list of field names to deduplicate by. The combination of these fields’ values will be used to determine uniqueness.

Examples

>>> dedup = Deduplicate(by=["field1", "field2"])

class unitxt.operators.DefaultPlaceHolder[source]¶: Bases: object

class unitxt.operators.DeterministicBalancer(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] | NoneType = None, dont_apply_to_streams: List[str] = None, max_instances: int = None, fields: List[str] = __required__)[source]¶: Bases: Balance

class unitxt.operators.DivideAllFieldsBy(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, divisor: float = 1.0, strict: bool = False)[source]¶

Bases: InstanceOperator

Recursively reach down to all fields that are float, and divide each by ‘divisor’.

The given instance is viewed as a tree whose internal nodes are dictionaries and lists, and the leaves are either ‘float’ and then divided, or other basic type, in which case, a ValueError is raised if input flag ‘strict’ is True, or – left alone, if ‘strict’ is False.

Parameters:

divisor (float) –
strict (bool) –

Example

when instance {“a”: 10.0, “b”: [2.0, 4.0, 7.0], “c”: 5} is processed by operator: operator = DivideAllFieldsBy(divisor=2.0) the output is: {“a”: 5.0, “b”: [1.0, 2.0, 3.5], “c”: 5} If the operator were defined with strict=True, through: operator = DivideAllFieldsBy(divisor=2.0, strict=True), the processing of the above instance would raise a ValueError, for the integer at “c”.

exception unitxt.operators.DownloadError(message)[source]¶: Bases: Exception

class unitxt.operators.DownloadOperator(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], source: str = __required__, target: str = __required__)[source]¶

Bases: SideEffectOperator

Operator for downloading a file from a given URL to a specified local path.

Parameters:

source (str) – URL of the file to be downloaded.
target (str) – Local path where the downloaded file should be saved.

class unitxt.operators.DuplicateInstances(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, num_duplications: int = __required__, duplication_index_field: str | NoneType = None)[source]¶

Bases: StreamOperator

Operator which duplicates each instance in stream a given number of times.

Parameters:

num_duplications (int) – How many times each instance should be duplicated (1 means no duplication).
duplication_index_field (Optional[str]) – If given, then additional field with specified name is added to each duplicated instance, which contains id of a given duplication. Defaults to None, so no field is added.

class unitxt.operators.EncodeLabels(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: List[str] = __required__)[source]¶

Bases: InstanceOperator

Encode each value encountered in any field in ‘fields’ into the integers 0,1,…

Encoding is determined by a str->int map that is built on the go, as different values are first encountered in the stream, either as list members or as values in single-value fields.

Parameters:: fields (List[str]) – The fields to encode together.

Example

applying EncodeLabels(fields = ["a", "b/*"]) on input stream = [{"a": "red", "b": ["red", "blue"], "c":"bread"}, {"a": "blue", "b": ["green"], "c":"water"}] will yield the output stream = [{'a': 0, 'b': [0, 1], 'c': 'bread'}, {'a': 1, 'b': [2], 'c': 'water'}]

Note: dict_utils are applied here, and hence, fields that are lists, should be included in input ‘fields’ with the appendix "/*" as in the above example.

class unitxt.operators.ExecuteExpression(data_classification_policy: List[str] = None, expression: str = __required__, imports_list: List[str] = [], _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, to_field: str = __required__)[source]¶

Bases: InstanceOperator, ComputeExpressionMixin

Compute an expression, specified as a string to be eval-uated, over the instance’s fields, and store the result in field to_field.

Raises an error if a field mentioned in the query is missing from the instance.

Parameters:

expression (str) – an expression to be evaluated over the fields of the instance
to_field (str) – the field where the result is to be stored into
imports_list (List[str]) – names of imports needed for the eval of the query (e.g. ‘re’, ‘json’)

Examples

When instance {“a”: 2, “b”: 3} is process-ed by operator ExecuteExpression(expression=”a+b”, to_field = “c”) the result is {“a”: 2, “b”: 3, “c”: 5}

When instance {“a”: “hello”, “b”: “world”} is process-ed by operator ExecuteExpression(expression = “a+’ ‘+b”, to_field = “c”) the result is {“a”: “hello”, “b”: “world”, “c”: “hello world”}

class unitxt.operators.ExtractFieldValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, field: str = __required__, stream_name: str = __required__, overall_top_frequency_percent: int | NoneType = 100, min_frequency_percent: int | NoneType = 0, to_field: str = __required__, process_every_value: bool | NoneType = False)[source]¶: Bases: ExtractMostCommonFieldValues

class unitxt.operators.ExtractMostCommonFieldValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, field: str = __required__, stream_name: str = __required__, overall_top_frequency_percent: int | NoneType = 100, min_frequency_percent: int | NoneType = 0, to_field: str = __required__, process_every_value: bool | NoneType = False)[source]¶: Bases: MultiStreamOperator

class unitxt.operators.ExtractZipFile(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], zip_file: str = __required__, target_dir: str = __required__)[source]¶

Bases: SideEffectOperator

Operator for extracting files from a zip archive.

Parameters:

zip_file (str) – Path of the zip file to be extracted.
target_dir (str) – Directory where the contents of the zip file will be extracted.

class unitxt.operators.FeatureGroupedShuffle(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, page_size: int = 1000, random_generator: random.Random = None, grouping_features: List[str] = None, shuffle_within_group: bool = False)[source]¶

Bases: Shuffle

Class for shuffling an input dataset by instance ‘blocks’, not on the individual instance level.

Example is if the dataset consists of questions with paraphrases of it, and each question falls into a topic. All paraphrases have the same ID value as the original. In this case, we may want to shuffle on grouping_features = [‘question ID’], to keep the paraphrases and original question together. We may also want to group by both ‘question ID’ and ‘topic’, if the question IDs are repeated between topics. In this case, grouping_features = [‘question ID’, ‘topic’]

Parameters:

grouping_features (list of strings) – list of feature names to use to define the groups. a group is defined by each unique observed combination of data values for features in grouping_features
shuffle_within_group (bool) – whether to further shuffle the instances within each group block, keeping the block order

Args (of superclass):

page_size (int): The size of each page in the stream. Defaults to 1000.: Note: shuffle_by_grouping_features determines the unique groups (unique combinations of values of grouping_features) separately by page (determined by page_size). If a block of instances in the same group are split into separate pages (either by a page break falling in the group, or the dataset was not sorted by grouping_features), these instances will be shuffled separately and thus the grouping may be broken up by pages. If the user wants to ensure the shuffle does the grouping and shuffling across all pages, set the page_size to be larger than the dataset size. See outputs_2features_bigpage and outputs_2features_smallpage in test_grouped_shuffle.

class unitxt.operators.FieldOperator(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶: Bases: InstanceFieldOperator

class unitxt.operators.Fillna(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, value: Any = __required__)[source]¶: Bases: FieldOperator

class unitxt.operators.FilterByCondition(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, values: Dict[str, Any] = __required__, condition: str = __required__, error_on_filtered_all: bool = True)[source]¶

Bases: StreamOperator

Filters a stream, yielding only instances in which the values in required fields follow the required condition operator.

Raises an error if a required field name is missing from the input instance.

Parameters:

values (Dict[str, Any]) – Field names and respective Values that instances must match according the condition, to be included in the output.
condition – the name of the desired condition operator between the specified (sub) field’s value and the provided constant value. Supported conditions are (“gt”, “ge”, “lt”, “le”, “ne”, “eq”, “in”,”not in”)
error_on_filtered_all (bool, optional) – If True, raises an error if all instances are filtered out. Defaults to True.

Examples

FilterByCondition(values = {"a":4}, condition = "gt") will yield only instances where field "a" contains a value > 4
FilterByCondition(values = {"a":4}, condition = "le") will yield only instances where "a"<=4
FilterByCondition(values = {"a":[4,8]}, condition = "in") will yield only instances where "a" is 4 or 8
FilterByCondition(values = {"a":[4,8]}, condition = "not in") will yield only instances where "a" is different from 4 or 8
FilterByCondition(values = {"a/b":[4,8]}, condition = "not in") will yield only instances where "a" is a dict in which key "b" is mapped to a value that is neither 4 nor 8
FilterByCondition(values = {"a[2]":4}, condition = "le") will yield only instances where “a” is a list whose 3-rd element is <= 4
FilterByCondition(values = {"a":False}, condition = "exists") will yield only instances which do not contain a field named "a"
FilterByCondition(values = {"a/b":True}, condition = "exists") will yield only instances which contain a field named "a" whose value is a dict containing, in turn, a field named "b"

condition_to_func = {'eq': <built-in function eq>, 'exists': None, 'ge': <built-in function ge>, 'gt': <built-in function gt>, 'in': None, 'le': <built-in function le>, 'lt': <built-in function lt>, 'ne': <built-in function ne>, 'not in': None}¶

class unitxt.operators.FilterByConditionBasedOnFields(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, values: Dict[str, Any] = __required__, condition: str = __required__, error_on_filtered_all: bool = True)[source]¶

Bases: FilterByCondition

Filters a stream based on a condition between 2 fields values.

Raises an error if either of the required fields names is missing from the input instance.

Parameters:

values (Dict[str, str]) – The fields names that the filter operation is based on.
condition – the name of the desired condition operator between the specified field’s values. Supported conditions are (“gt”, “ge”, “lt”, “le”, “ne”, “eq”, “in”,”not in”)
error_on_filtered_all (bool, optional) – If True, raises an error if all instances are filtered out. Defaults to True.

Examples

FilterByCondition(values = {“a”:”b}, condition = “gt”) will yield only instances where field “a” contains a value greater then the value in field “b”. FilterByCondition(values = {“a”:”b}, condition = “le”) will yield only instances where “a”<=”b”

class unitxt.operators.FilterByExpression(data_classification_policy: List[str] = None, expression: str = __required__, imports_list: List[str] = [], _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, error_on_filtered_all: bool = True)[source]¶

Bases: StreamOperator, ComputeExpressionMixin

Filters a stream, yielding only instances which fulfil a condition specified as a string to be python’s eval-uated.

Raises an error if a field participating in the specified condition is missing from the instance

Parameters:

expression (str) – a condition over fields of the instance, to be processed by python’s eval()
imports_list (List[str]) – names of imports needed for the eval of the query (e.g. ‘re’, ‘json’)
error_on_filtered_all (bool, optional) – If True, raises an error if all instances are filtered out. Defaults to True.

Examples

FilterByExpression(expression = "a > 4") will yield only instances where “a”>4
FilterByExpression(expression = "a <= 4 and b > 5") will yield only instances where the value of field “a” is not exceeding 4 and in field “b” – greater than 5
FilterByExpression(expression = "a in [4, 8]") will yield only instances where “a” is 4 or 8
FilterByExpression(expression = "a not in [4, 8]") will yield only instances where “a” is neither 4 nor 8
FilterByExpression(expression = "a['b'] not in [4, 8]") will yield only instances where “a” is a dict in which key ‘b’ is mapped to a value that is neither 4 nor 8

class unitxt.operators.FixJsonSchemaOfParameterTypes(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, main_field: str = __required__)[source]¶: Bases: InstanceOperator

class unitxt.operators.FlattenInstances(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, parent_key: str = '', sep: str = '_')[source]¶

Bases: InstanceOperator

Flattens each instance in a stream, making nested dictionary entries into top-level entries.

Parameters:

parent_key (str) – A prefix to use for the flattened keys. Defaults to an empty string.
sep (str) – The separator to use when concatenating nested keys. Defaults to “_”.

class unitxt.operators.FromIterables(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None)[source]¶

Bases: StreamInitializerOperator

Creates a MultiStream from a dict of named iterables.

Example

operator = FromIterables() ms = operator.process(iterables)

class unitxt.operators.GetItemByIndex(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, items_list: List[Any] = __required__)[source]¶

Bases: FieldOperator

Get the element from the fixed list by the index in the given field and store in another field.

Example

GetItemByIndex(items_list=[“dog”,cat”],field=”animal_index”,to_field=”animal”)

on instance {“animal_index” : 1} will change the instance to {“animal_index” : 1, “animal” : “cat”}

class unitxt.operators.IndexOf(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, search_in: str = __required__, index_of: str = __required__, to_field: str = __required__, use_query: bool | NoneType = None)[source]¶

Bases: InstanceOperator

For a given instance, finds the offset of value of field ‘index_of’, within the value of field ‘search_in’.

class unitxt.operators.InstanceFieldOperator(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶

Bases: InstanceOperator

A general stream instance operator that processes the values of a field (or multiple ones).

Parameters:

field (Optional[str]) – The field to process, if only a single one is passed. Defaults to None
to_field (Optional[str]) – Field name to save result into, if only one field is processed, if None is passed the operation would happen in-place and its result would replace the value of field. Defaults to None
field_to_field (Optional[Union[List[List[str]], Dict[str, str]]]) – Mapping from names of fields to process, to names of fields to save the results into. Inner List, if used, should be of length 2. A field is processed by feeding its value into method process_value and storing the result in to_field that is mapped to the field. When the type of argument field_to_field is List, the order by which the fields are processed is their order in the (outer) List. But when the type of argument field_to_field is Dict, there is no uniquely determined order. The end result might depend on that order if either (1) two different fields are mapped to the same to_field, or (2) a field shows both as a key and as a value in different mappings. The operator throws an AssertionError in either of these cases. field_to_field defaults to None.
process_every_value (bool) – Processes the values in a list instead of the list as a value, similar to python’s *var. Defaults to False

Note: if field and to_field (or both members of a pair in field_to_field ) are equal (or share a common prefix if field and to_field contain a / ), then the result of the operation is saved within field .

class unitxt.operators.InterleaveListsToDialogOperator(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, user_turns_field: str = __required__, assistant_turns_field: str = __required__, user_role_label: str = 'user', assistant_role_label: str = 'assistant', to_field: str = __required__)[source]¶

Bases: InstanceOperator

Interleaves two lists, one of user dialog turns and one of assistant dialog turns, into a single list of tuples, alternating between “user” and “assistant”.

The list of tuples if of format (role, turn_content), where the role label is specified by the ‘user_role_label’ and ‘assistant_role_label’ fields (default to “user” and “assistant”).

The user turns and assistant turns field are specified in the arguments. The value of each of the ‘fields’ is assumed to be a list.

class unitxt.operators.Intersect(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, allowed_values: List[Any] = __required__)[source]¶

Bases: FieldOperator

Intersects the value of a field, which must be a list, with a given list.

Parameters:: allowed_values (list) –

class unitxt.operators.IntersectCorrespondingFields(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str = __required__, allowed_values: List[str] = __required__, corresponding_fields_to_intersect: List[str] = __required__)[source]¶

Bases: InstanceOperator

Intersects the value of a field, which must be a list, with a given list , and removes corresponding elements from other list fields.

For example:

Assume the instances contain a field of ‘labels’ and a field with the labels’ corresponding ‘positions’ in the text.

IntersectCorrespondingFields(field="label",
                            allowed_values=["b", "f"],
                            corresponding_fields_to_intersect=["position"])

would keep only “b” and “f” values in ‘labels’ field and their respective values in the ‘position’ field. (All other fields are not effected)

Given this input:

[
    {"label": ["a", "b"],"position": [0,1],"other" : "not"},
    {"label": ["a", "c", "d"], "position": [0,1,2], "other" : "relevant"},
    {"label": ["a", "b", "f"], "position": [0,1,2], "other" : "field"}
]

So the output would be:
[
        {"label": ["b"], "position":[1],"other" : "not"},
        {"label": [], "position": [], "other" : "relevant"},
        {"label": ["b", "f"],"position": [1,2], "other" : "field"},
]

Parameters:

intersected (field - the field to) –
allowed_values (list) –
corresponding_fields_to_intersect (list) –
'field' (are removed based the corresponding indices of values removed from the) –

class unitxt.operators.IterableSource(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, iterables: Dict[str, Iterable] = __required__)[source]¶

Bases: SourceOperator

Creates a MultiStream from a dict of named iterables.

It is a callable.

Parameters:: iterables (Dict[str, Iterable]) – A dictionary mapping stream names to iterables.

Example

operator = IterableSource(input_dict) ms = operator()

class unitxt.operators.JoinStr(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, separator: str = ',')[source]¶

Bases: FieldOperator

Joins a list of strings (contents of a field), similar to str.join().

Parameters:: separator (str) – text to put between values

class unitxt.operators.LengthBalancer(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] | NoneType = None, dont_apply_to_streams: List[str] = None, max_instances: int = None, fields: List[str] | NoneType = __required__, segments_boundaries: List[int] = __required__)[source]¶

Bases: DeterministicBalancer

Balances by a signature that reflects the total length of the fields’ values, quantized into integer segments.

Parameters:

segments_boundaries (List[int]) – distinct integers sorted in increasing order, that map a given total length into the index of the least of them that exceeds the given total length. (If none exceeds – into one index beyond, namely, the length of segments_boundaries)
fields (Optional, List[str]) – the total length of the values of these fields goes through the quantization described above

Example

when input [{"a": [1, 3], "b": 0, "id": 0}, {"a": [1, 3], "b": 0, "id": 1}, {"a": [], "b": "a", "id": 2}] is fed into LengthBalancer(fields=["a"], segments_boundaries=[1]), input instances will be counted and balanced against two categories: empty total length (less than 1), and non-empty.

class unitxt.operators.ListFieldValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: List[str] = __required__, to_field: str = __required__, use_query: bool | NoneType = None)[source]¶

Bases: InstanceOperator

Concatenates values of multiple fields into a list, and assigns it to a new field.

class unitxt.operators.MapInstanceValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, mappers: Dict[str, Dict[str, str]] = __required__, strict: bool = True, process_every_value: bool = False)[source]¶

Bases: InstanceOperator

A class used to map instance values into other values.

This class is a type of InstanceOperator, it maps values of instances in a stream using predefined mappers.

Parameters:

mappers (Dict[str, Dict[str, Any]]) – The mappers to use for mapping instance values. Keys are the names of the fields to undergo mapping, and values are dictionaries that define the mapping from old values to new values. Note that mapped values are defined by their string representation, so mapped values are converted to strings before being looked up in the mappers.
strict (bool) – If True, the mapping is applied strictly. That means if a value does not exist in the mapper, it will raise a KeyError. If False, values that are not present in the mapper are kept as they are.
process_every_value (bool) – If True, all fields to be mapped should be lists, and the mapping is to be applied to their individual elements. If False, mapping is only applied to a field containing a single value.

Examples

MapInstanceValues(mappers={"a": {"1": "hi", "2": "bye"}}) replaces "1" with "hi" and "2" with "bye" in field "a" in all instances of all streams: instance {"a": 1, "b": 2} becomes {"a": "hi", "b": 2}. Note that the value of "b" remained intact, since field-name "b" does not participate in the mappers, and that 1 was casted to "1" before looked up in the mapper of "a".

MapInstanceValues(mappers={"a": {"1": "hi", "2": "bye"}}, process_every_value=True): Assuming field "a" is a list of values, potentially including "1"-s and "2"-s, this replaces each such "1" with "hi" and "2" – with "bye" in all instances of all streams: instance {"a": ["1", "2"], "b": 2} becomes {"a": ["hi", "bye"], "b": 2}.

MapInstanceValues(mappers={"a": {"1": "hi", "2": "bye"}}, strict=True): To ensure that all values of field "a" are mapped in every instance, use strict=True. Input instance {"a":"3", "b": 2} will raise an exception per the above call, because "3" is not a key in the mapper of "a".

MapInstanceValues(mappers={"a": {str([1,2,3,4]): "All", str([]): "None"}}, strict=True) replaces a list [1,2,3,4] with the string "All" and an empty list by string "None".

class unitxt.operators.MapValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, mapping: Dict[str, str] = __required__)[source]¶: Bases: FieldOperator

class unitxt.operators.MergeStreams(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, streams_to_merge: List[str] = None, new_stream_name: str = 'all', add_origin_stream_name: bool = True, origin_stream_name_field_name: str = 'origin')[source]¶

Bases: MultiStreamOperator

Merges multiple streams into a single stream.

Parameters:

new_stream_name (str) – The name of the new stream resulting from the merge.
add_origin_stream_name (bool) – Whether to add the origin stream name to each instance.
origin_stream_name_field_name (str) – The field name for the origin stream name.

class unitxt.operators.MinimumOneExamplePerLabelRefiner(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] | NoneType = None, dont_apply_to_streams: List[str] = None, max_instances: int = None, fields: List[str] = __required__)[source]¶

Bases: StreamRefiner

A class used to return a specified number instances ensuring at least one example per label.

For each instance, a signature value is constructed from the values of the instance in specified input fields. MinimumOneExamplePerLabelRefiner takes first instance that appears from each label (each unique signature), and then adds more elements up to the max_instances limit. In general, the refiner takes the first elements in the stream that meet the required conditions. MinimumOneExamplePerLabelRefiner then shuffles the results to avoid having one instance from each class first and then the rest . If max instance is not set, the original stream will be used

Parameters:

fields (List[str]) – A list of field names to be used in producing the instance’s signature.
max_instances (Optional, int) – Number of elements to select. Note that max_instances of StreamRefiners that are passed to the recipe (e.g. train_refiner. test_refiner) are overridden by the recipe parameters ( max_train_instances, max_test_instances)

Usage:: balancer = MinimumOneExamplePerLabelRefiner(fields=["field1", "field2"], max_instances=200)

balanced_stream = balancer.process(stream)

Example

When input [{"a": 1, "b": 1},{"a": 1, "b": 2},{"a": 1, "b": 3},{"a": 1, "b": 4},{"a": 2, "b": 5}] is fed into MinimumOneExamplePerLabelRefiner(fields=["a"], max_instances=3) the resulting stream will be: [{'a': 1, 'b': 1}, {'a': 1, 'b': 2}, {'a': 2, 'b': 5}] (order may be different)

class unitxt.operators.Move(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str = __required__, to_field: str = __required__)[source]¶: Bases: InstanceOperator

class unitxt.operators.Perturb(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, select_from: List[Any] = [], percentage_to_perturb: int = 1)[source]¶

Bases: FieldOperator

Slightly perturbs the contents of field. Could be Handy for imitating prediction from given target.

When task was classification, argument select_from can be used to list the other potential classes, as a relevant perturbation

Parameters:

percentage_to_perturb (int) – the percentage of the instances for which to apply this perturbation. Defaults to 1 (1 percent)
select_from – List[Any]: a list of values to select from, as a perturbation of the field’s value. Defaults to [].

select_from: List[Any] = []¶

class unitxt.operators.ReadFile(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, encoding: str = 'utf-8')[source]¶

Bases: FieldOperator

Reads file content from local path or URL.

This operator can read files from local filesystem paths or remote URLs. The content is returned as a string.

Parameters:: encoding (str) – Text encoding to use when reading the file. Defaults to ‘utf-8’.

Example

Reading a local file

ReadFile(field="file_path", to_field="content")

Reading from URL

ReadFile(field="url", to_field="content")

process_value(value: str) → str[source]¶: Read file content from local path or URL.

class unitxt.operators.RecursiveCopy(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶: Bases: FieldOperator

class unitxt.operators.RecursiveReplace(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, key: str = __required__, map_values: dict = __required__, remove_values: list | NoneType = None)[source]¶

Bases: InstanceOperator

An operator to recursively replace values in dictionary fields of instances based on a key and a mapping of values.

key¶

The key in the dictionary to start the replacement process.

Type:: str

map_values¶

A dictionary containing the key-value pairs to replace the original values.

Type:: dict

remove_values¶

An optional list of values to remove from the dictionary. Defaults to None.

Type:: Optional[list]

Example: RecursiveReplace(key=”a”, map_values={“1”: “hi”, “2”: “bye” }, remove_values=[“3”])

replaces the value of key “a” in all instances of all streams: instance {"field" : [{"a": "1", "b" : "2"}, {"a" : "3", "b:" "4"}}` becomes ``{"field" : [{"a": "hi", "b" : "2"}, {"b": "4"}}

Notice how the value of field "a" in the first instance is replaced with "hi" and the value of field "a" in the second instance is removed.

class unitxt.operators.RemoveFields(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: List[str] = __required__)[source]¶

Bases: InstanceOperator

Remove specified fields from each instance in a stream.

Parameters:: fields (List[str]) – The fields to remove from each instance.

class unitxt.operators.RemoveValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, unallowed_values: List[Any] = __required__)[source]¶

Bases: FieldOperator

Removes elements in a field, which must be a list, using a given list of unallowed.

Parameters:: unallowed_values (list) –

class unitxt.operators.Rename(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶

Bases: FieldOperator

Renames fields.

Move value from one field to another, potentially, if field name contains a /, from one branch into another. Remove the from field, potentially part of it in case of / in from_field.

Examples

Rename(field_to_field={“b”: “c”}) will change inputs [{“a”: 1, “b”: 2}, {“a”: 2, “b”: 3}] to [{“a”: 1, “c”: 2}, {“a”: 2, “c”: 3}]

Rename(field_to_field={“b”: “c/d”}) will change inputs [{“a”: 1, “b”: 2}, {“a”: 2, “b”: 3}] to [{“a”: 1, “c”: {“d”: 2}}, {“a”: 2, “c”: {“d”: 3}}]

Rename(field_to_field={“b”: “b/d”}) will change inputs [{“a”: 1, “b”: 2}, {“a”: 2, “b”: 3}] to [{“a”: 1, “b”: {“d”: 2}}, {“a”: 2, “b”: {“d”: 3}}]

Rename(field_to_field={“b/c/e”: “b/d”}) will change inputs [{“a”: 1, “b”: {“c”: {“e”: 2, “f”: 20}}}] to [{“a”: 1, “b”: {“c”: {“f”: 20}, “d”: 2}}]

class unitxt.operators.SelectFields(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: List[str] = __required__)[source]¶

Bases: InstanceOperator

Keep only specified fields from each instance in a stream.

Parameters:: fields (List[str]) – The fields to keep from each instance.

class unitxt.operators.Set(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: Dict[str, object] = __required__, use_query: bool | NoneType = None, use_deepcopy: bool = False)[source]¶

Bases: InstanceOperator

Sets specified fields in each instance, in a given stream or all streams (default), with specified values. If fields exist, updates them, if do not exist – adds them.

Parameters:

fields (Dict[str, object]) – The fields to add to each instance. Use ‘/’ to access inner fields
use_deepcopy (bool) – Deep copy the input value to avoid later modifications

Examples

# Set a value of a list consisting of “positive” and “negative” do field “classes” to each and every instance of all streams Set(fields={"classes": ["positive","negatives"]})

# In each and every instance of all streams, field “span” is to become a dictionary containing a field “start”, in which the value 0 is to be set Set(fields={"span/start": 0}

# In all instances of stream “train” only, Set field “classes” to have the value of a list consisting of “positive” and “negative” Set(fields={"classes": ["positive","negatives"], apply_to_stream=["train"]})

# Set field “classes” to have the value of a given list, preventing modification of original list from changing the instance. Set(fields={"classes": alist}), use_deepcopy=True) if now alist is modified, still the instances remain intact.

class unitxt.operators.Shuffle(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, page_size: int = 1000, random_generator: random.Random = None)[source]¶

Bases: PagedStreamOperator

Shuffles the order of instances in each page of a stream.

Args (of superclass):: page_size (int): The size of each page in the stream. Defaults to 1000.

class unitxt.operators.ShuffleFieldValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False)[source]¶

Bases: FieldOperator

An operator that shuffles the values of a list field.

the seed for shuffling in the is determined by the elements of the input field, ensuring that the shuffling operation produces different results for different input lists, but also that it is deterministic and reproducible.

None¶

process_value(value

Any) -> Any: Shuffles the elements of the input list and returns the shuffled list.

Parameters:: value (Any) – The input list to be shuffled.

Returns:: The shuffled list.
Return type:: Any

class unitxt.operators.SplitByNestedGroup(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, field_name_of_group: str = 'group', number_of_fusion_generations: int = 1)[source]¶

Bases: MultiStreamOperator

Splits a MultiStream that is small - for metrics, hence: whole stream can sit in memory, split by the value of field ‘group’.

Parameters:: number_of_fusion_generations – int

the value in field group is of the form “sourcen/sourcenminus1/…” describing the sources in which the instance sat when these were fused, potentially several phases of fusion. the name of the most recent source sits first in this value. (See BaseFusion and its extensions) number_of_fuaion_generations specifies the length of the prefix by which to split the stream. E.g. for number_of_fusion_generations = 1, only the most recent fusion in creating this multi_stream, affects the splitting. For number_of_fusion_generations = -1, take the whole history written in this field, ignoring number of generations.

class unitxt.operators.StreamRefiner(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] | NoneType = None, dont_apply_to_streams: List[str] = None, max_instances: int = None)[source]¶

Bases: StreamOperator

Discard from the input stream all instances beyond the leading ‘max_instances’ instances.

Thereby, if the input stream consists of no more than ‘max_instances’ instances, the resulting stream is the whole of the input stream. And if the input stream consists of more than ‘max_instances’ instances, the resulting stream only consists of the leading ‘max_instances’ of the input stream.

Parameters:

max_instances (int) –
apply_to_streams (optional, list(str)) – names of streams to refine.

Examples

when input = [{"a": 1},{"a": 2},{"a": 3},{"a": 4},{"a": 5},{"a": 6}] is fed into StreamRefiner(max_instances=4) the resulting stream is [{"a": 1},{"a": 2},{"a": 3},{"a": 4}]

class unitxt.operators.TakeByField(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str = __required__, index: str = __required__, to_field: str = None, use_query: bool | NoneType = None)[source]¶

Bases: InstanceOperator

From field ‘field’ of a given instance, select the member indexed by field ‘index’, and store to field ‘to_field’.

exception unitxt.operators.UnexpectedHttpCodeError(http_code)[source]¶: Bases: Exception

class unitxt.operators.WikipediaFetcher(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = ['Wikipedia-API'], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, field: str | NoneType = None, to_field: str | NoneType = None, field_to_field: List[List[str]] | Dict[str, str] | NoneType = None, use_query: bool | NoneType = None, process_every_value: bool = False, set_every_value: bool = False, get_default: Any = None, not_exist_ok: bool = False, not_exist_do_nothing: bool = False, mode: Literal['summary', 'text'] = 'text')[source]¶: Bases: FieldOperator

class unitxt.operators.ZipFieldValues(data_classification_policy: List[str] = None, _requirements_list: List[str] | Dict[str, str] = [], requirements: List[str] | Dict[str, str] = [], caching: bool = None, apply_to_streams: List[str] = None, dont_apply_to_streams: List[str] = None, fields: List[str] = __required__, to_field: str = __required__, longest: bool = False, use_query: bool | NoneType = None)[source]¶

Bases: InstanceOperator

Zips values of multiple fields in a given instance, similar to list(zip(*fields)).

The value in each of the specified ‘fields’ is assumed to be a list. The lists from all ‘fields’ are zipped, and stored into ‘to_field’.

If ‘longest’=False, the length of the zipped result is determined by the shortest input value.

If ‘longest’=True, the length of the zipped result is determined by the longest input, padding shorter inputs with None-s.

unitxt.operators.recursive_key_value_replace(data, target_key, value_map, value_remove=None)[source]¶

Recursively traverses a data structure (dicts and lists), replaces values of target_key using value_map, and removes values listed in value_remove.

Parameters:

data – The data structure (dict or list) to traverse.
target_key – The specific key whose value needs to be checked and replaced or removed.
value_map – A dictionary mapping old values to new values.
value_remove – A list of values to completely remove if found as values of target_key.

Returns:

The modified data structure. Modification is done in-place.

unitxt.operators.update_scores_of_stream_instances(stream: Stream, scores: List[dict]) → Generator[source]¶