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 Specelized Operators

Some operators are specielized in specific task such as:

loaders for loading data.
splitters for fixing data splits.

Other specelized operators are used by unitxt internally:

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

The rest of this section is dedicated for general operators.

General Operaotrs List:

class unitxt.operators.AddConstant(*argv, **kwargs)

Bases: FieldOperator

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

Parameters:: add – the constant to add.

class unitxt.operators.AddFields(*argv, **kwargs)

Bases: StreamInstanceOperator

Adds specified fields to each instance in a given stream or all streams (default) If fields exist, updates them.

Parameters:

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

Examples

# Add a ‘classes’ field with a value of a list “positive” and “negative” to all streams AddFields(fields={“classes”: [“positive”,”negatives”]})

# Add a ‘start’ field under the ‘span’ field with a value of 0 to all streams AddFields(fields={“span/start”: 0}

# Add a ‘classes’ field with a value of a list “positive” and “negative” to ‘train’ stream AddFields(fields={“classes”: [“positive”,”negatives”], apply_to_stream=[“train”]})

# Add a ‘classes’ field on a given list, prevent modification of original list # from changing the instance. AddFields(fields={“classes”: alist}), use_deepcopy=True) # if now alist is modified, still the instances remain intact.

class unitxt.operators.AddID(*argv, **kwargs)

Bases: StreamInstanceOperator

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

class unitxt.operators.Apply(*argv, **kwargs)

Bases: StreamInstanceOperator

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

Parameters:

function (additional arguments are field names passed to the) – name of function.
to_field (str) – the field to store the result
function –

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(*argv, **kwargs)

Bases: SingleStreamOperator, 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(*argv, **kwargs)

Bases: StreamInstanceOperator, ArtifactFetcherMixin

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(*argv, **kwargs)

Bases: SingleStreamOperator, 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

Bases: object

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

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

cache: Dict[str, Artifact] = {}

classmethod get_artifact(artifact_identifier: str) → Artifact

class unitxt.operators.AugmentPrefixSuffix(*argv, **kwargs)

Bases: Augmentor

Augments the input by prepending and appending to it a randomly selected (typically, whitespace) patterns.

Parameters:

prefixes (list or dict) – the potential (typically, whitespace) patterns to select from. The dictionary version allows to specify relative weights of the different patterns.
suffixes (list or dict) – the potential (typically, whitespace) patterns to select from. The dictionary version allows to specify relative weights of the different patterns.
prefix_len (positive int) – The added prefix or suffix will be of length prefix_len of suffix_len, respectively, repetitions of the randomly selected patterns.
suffix_len (positive int) – The added prefix or suffix will be of length prefix_len of suffix_len, respectively, repetitions of the randomly selected patterns.
remove_existing_whitespaces – allows to first clean any existing leading and trailing whitespaces. The strings made of repetitions of the selected pattern(s) are then prepended and/or appended to the potentially trimmed input.
needed (If only one of prefixes/suffixes is) –
None. (set the other to) –

Examples

To prepend the input with a prefix made of 4 ‘n’-s or ‘t’-s, employ AugmentPrefixSuffix(augment_model_input=True, prefixes=[’n’,’t’], prefix_len=4, suffixes = None) To append the input with a suffix made of 3 ‘n’-s or ‘t’-s, with triple ‘n’ suffixes being preferred over triple ‘t’, at 2:1 ratio, employ AugmentPrefixSuffix(augment_model_input=True, suffixes={’n’:2,’t’:1}, suffix_len=3, prefixes = None) which will append ‘n’-s twice as often as ‘t’-s.

prefixes: List[str] | Dict[str, int] | None = {'': 30, ' ': 20, '\\n': 40, '\\t': 10}

suffixes: List[str] | Dict[str, int] | None = {'': 30, ' ': 20, '\\n': 40, '\\t': 10}

class unitxt.operators.AugmentWhitespace(*argv, **kwargs)

Bases: Augmentor

Augments the inputs by replace existing whitespace with other whitespace.

Currently each whitespace is replaced by a random choice of 1-3 whitespace charaters (spcae, tab, newline).

class unitxt.operators.Augmentor(*argv, **kwargs)

Bases: StreamInstanceOperator

A stream that augments the values of either the task input fields before rendering with the template, or the input passed to the model after rendering of the template.

Parameters:

augment_model_input – Whether to augment the input to the model.
augment_task_input – Whether to augment the task input fields. The specific fields are defined in the FormTask operator.

class unitxt.operators.CastFields(*argv, **kwargs)

Bases: StreamInstanceOperator

Casts specified fields to specified types.

Parameters:

use_nested_query (bool) – Whether to cast nested fields, expressed in dpath. Defaults to False.
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.

Examples

CastFields(: fields={“a/d”: “float”, “b”: “int”}, failure_defaults={“a/d”: 0.0, “b”: 0}, process_every_value=True, use_nested_query=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.CopyFields(*argv, **kwargs)

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. use_query (bool): Whether to use dpath for accessing fields. Defaults to False.

Examples

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

with use_query=True, we can also copy inside the field: CopyFields(field_to_field={“a/0”: “a”}, use_query=True) would process instance {“a”: [1, 3]} into {“a”: 1}

class unitxt.operators.DeterministicBalancer(*argv, **kwargs)

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.

fields

A list of field names to be used in producing the instance’s signature.

Type:: List[str]

max_instances

Type:: Optional, int

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.DivideAllFieldsBy(*argv, **kwargs)

Bases: StreamInstanceOperator

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): Bases: Exception

class unitxt.operators.DownloadOperator(*argv, **kwargs)

Bases: SideEffectOperator

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

source

URL of the file to be downloaded.

Type:: str

target

Local path where the downloaded file should be saved.

Type:: str

class unitxt.operators.EncodeLabels(*argv, **kwargs)

Bases: StreamInstanceOperator

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: dpath is 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.ExecuteQuery(*argv, **kwargs)

Bases: StreamInstanceOperator

Compute an expression (query), expressed 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:

query (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

Examples

When instance {“a”: 2, “b”: 3} is process-ed by operator ExecuteQuery(query=”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 ExecuteQuery(query = “a+’ ‘+b”, to_field = “c”) the result is {“a”: “hello”, “b”: “world”, “c”: “hello world”}

class unitxt.operators.ExtractFieldValues(*argv, **kwargs): Bases: ExtractMostCommonFieldValues

class unitxt.operators.ExtractMostCommonFieldValues(*argv, **kwargs): Bases: MultiStreamOperator

class unitxt.operators.ExtractZipFile(*argv, **kwargs)

Bases: SideEffectOperator

Operator for extracting files from a zip archive.

zip_file

Path of the zip file to be extracted.

Type:: str

target_dir

Directory where the contents of the zip file will be extracted.

Type:: str

class unitxt.operators.FieldOperator(*argv, **kwargs)

Bases: StreamInstanceOperator

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 *var. Defaults to False
use_query (bool) – Whether to use dpath style queries. 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
'use_query'=True) (prefix if) –
'field' (then the result of the operation is saved within) –

class unitxt.operators.FilterByCondition(*argv, **kwargs)

Bases: SingleStreamOperator

Filters a stream, yielding only instances for which the required values follows the required condition operator.

Raises an error if a required key is missing.

Parameters:

values (Dict[str, Any]) – Values that instances must match using the condition to be included in the output.
condition – the name of the desired condition operator between the key and the value in values (“gt”, “ge”, “lt”, “le”, “ne”, “eq”)
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 “a”>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” different from 4 or 8

condition_to_func = {'eq': <built-in function eq>, '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.FilterByQuery(*argv, **kwargs)

Bases: SingleStreamOperator

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:

query (str) – a condition over fields of the instance, to be processed by python’s eval()
error_on_filtered_all (bool, optional) – If True, raises an error if all instances are filtered out. Defaults to True.

Examples

FilterByQuery(query = “a > 4”) will yield only instances where “a”>4 FilterByQuery(query = “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 FilterByQuery(query = “a in [4, 8]”) will yield only instances where “a” is 4 or 8 FilterByQuery(query = “a not in [4, 8]”) will yield only instances where “a” is neither 4 nor 8

class unitxt.operators.FlattenInstances(*argv, **kwargs)

Bases: StreamInstanceOperator

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(*argv, **kwargs)

Bases: StreamInitializerOperator

Creates a MultiStream from a dict of named iterables.

Example

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

class unitxt.operators.IndexOf(*argv, **kwargs)

Bases: StreamInstanceOperator

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

class unitxt.operators.Intersect(*argv, **kwargs)

Bases: FieldOperator

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

Parameters:: allowed_values (list) –

class unitxt.operators.IterableSource(*argv, **kwargs)

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(*argv, **kwargs)

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(*argv, **kwargs)

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 maps a given total length
index (into the index of the least of them that exceeds the total length. (If none exceeds -- into one) –
beyond –
namely –
segments_boudaries) (the length of) –
fields (Optional, List[str]) –

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(*argv, **kwargs)

Bases: StreamInstanceOperator

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

class unitxt.operators.MapInstanceValues(*argv, **kwargs)

Bases: StreamInstanceOperator

A class used to map instance values into other values.

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

mappers

The mappers to use for mapping instance values. Keys are the names of the fields to be mapped, and values are dictionaries that define the mapping from old values to new values.

Type:: Dict[str, Dict[str, str]]

strict

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.

Type:: bool

process_every_value

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.

Type:: bool

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}.

MapInstanceValues(mappers={“a”: {“1”: “hi”, “2”: “bye”}}, process_every_element=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’. Note that mapped values are defined by their string representation, so mapped values must be converted to strings.

class unitxt.operators.MergeStreams(*argv, **kwargs)

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.NullAugmentor(*argv, **kwargs)

Bases: Augmentor

Does not change the input string.

class unitxt.operators.Perturbate(*argv, **kwargs)

Bases: FieldOperator

Slightly perturbates 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

select_from: List[Any] = []

class unitxt.operators.RemoveFields(*argv, **kwargs)

Bases: StreamInstanceOperator

Remove specified fields from each instance in a stream.

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

class unitxt.operators.RemoveValues(*argv, **kwargs)

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.RenameFields(*argv, **kwargs)

Bases: FieldOperator

Renames fields.

Move value from one field to another, potentially, if ‘use_query’=True, from one branch into another. Remove the from field, potentially part of it in case of use_query.

Examples

RenameFields(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}]

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

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

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

class unitxt.operators.Shuffle(*argv, **kwargs)

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(*argv, **kwargs)

Bases: FieldOperator

Shuffles a list of values found in a field.

class unitxt.operators.SplitByValue(*argv, **kwargs)

Bases: MultiStreamOperator

Splits a MultiStream into multiple streams based on unique values in specified fields.

Parameters:: fields (List[str]) – The fields to use when splitting the MultiStream.

class unitxt.operators.StreamRefiner(*argv, **kwargs)

Bases: SingleStreamOperator

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.

Args: 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(*argv, **kwargs)

Bases: StreamInstanceOperator

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): Bases: Exception

class unitxt.operators.Unique(*argv, **kwargs)

Bases: SingleStreamReducer

Reduces a stream to unique instances based on specified fields.

Parameters:: fields (List[str]) – The fields that should be unique in each instance.

class unitxt.operators.ZipFieldValues(*argv, **kwargs)

Bases: StreamInstanceOperator

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’=False, the length of the zipped result is determined by the longest input, padding shorter inputs with None -s.