• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

IBM / unitxt / 13098237558

02 Feb 2025 11:12AM UTC coverage: 79.223% (-0.1%) from 79.32%
13098237558

Pull #1564

github

web-flow
Merge c2ed1b6ad into 912dc2aad
Pull Request #1564: Revisit huggingface cache policy

1451 of 1825 branches covered (79.51%)

Branch coverage included in aggregate %.

9149 of 11555 relevant lines covered (79.18%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.04
src/unitxt/api.py
1
import inspect
1✔
2
import json
1✔
3
from datetime import datetime
1✔
4
from functools import lru_cache
1✔
5
from typing import Any, Dict, List, Optional, Union
1✔
6

7
from datasets import Dataset, DatasetDict, IterableDataset, IterableDatasetDict
1✔
8

9
from .artifact import fetch_artifact
1✔
10
from .card import TaskCard
1✔
11
from .dataset_utils import get_dataset_artifact
1✔
12
from .error_utils import UnitxtError
1✔
13
from .inference import (
1✔
14
    InferenceEngine,
15
    LogProbInferenceEngine,
16
    OptionSelectingByLogProbsInferenceEngine,
17
)
18
from .loaders import LoadFromDictionary
1✔
19
from .logging_utils import get_logger
1✔
20
from .metric_utils import EvaluationResults, _compute, _inference_post_process
1✔
21
from .operator import SourceOperator
1✔
22
from .schema import UNITXT_DATASET_SCHEMA, loads_instance
1✔
23
from .settings_utils import get_constants, get_settings
1✔
24
from .standard import DatasetRecipe
1✔
25
from .task import Task
1✔
26

27
logger = get_logger()
1✔
28
constants = get_constants()
1✔
29
settings = get_settings()
1✔
30

31

32
def load(source: Union[SourceOperator, str]):
1✔
33
    assert isinstance(
×
34
        source, (SourceOperator, str)
35
    ), "source must be a SourceOperator or a string"
36
    if isinstance(source, str):
×
37
        source, _ = fetch_artifact(source)
×
38
    return source().to_dataset()
×
39

40

41
def _get_recipe_from_query(dataset_query: str) -> DatasetRecipe:
1✔
42
    dataset_query = dataset_query.replace("sys_prompt", "instruction")
1✔
43
    try:
1✔
44
        dataset_stream, _ = fetch_artifact(dataset_query)
1✔
45
    except:
1✔
46
        dataset_stream = get_dataset_artifact(dataset_query)
1✔
47
    return dataset_stream
1✔
48

49

50
def _get_recipe_from_dict(dataset_params: Dict[str, Any]) -> DatasetRecipe:
1✔
51
    recipe_attributes = list(DatasetRecipe.__dict__["__fields__"].keys())
1✔
52
    for param in dataset_params.keys():
1✔
53
        assert param in recipe_attributes, (
1✔
54
            f"The parameter '{param}' is not an attribute of the 'DatasetRecipe' class. "
55
            f"Please check if the name is correct. The available attributes are: '{recipe_attributes}'."
56
        )
57
    return DatasetRecipe(**dataset_params)
1✔
58

59

60
def _verify_dataset_args(dataset_query: Optional[str] = None, dataset_args=None):
1✔
61
    if dataset_query and dataset_args:
1✔
62
        raise ValueError(
×
63
            "Cannot provide 'dataset_query' and key-worded arguments at the same time. "
64
            "If you want to load dataset from a card in local catalog, use query only. "
65
            "Otherwise, use key-worded arguments only to specify properties of dataset."
66
        )
67

68
    if dataset_query:
1✔
69
        if not isinstance(dataset_query, str):
1✔
70
            raise ValueError(
×
71
                f"If specified, 'dataset_query' must be a string, however, "
72
                f"'{dataset_query}' was provided instead, which is of type "
73
                f"'{type(dataset_query)}'."
74
            )
75

76
    if not dataset_query and not dataset_args:
1✔
77
        raise ValueError(
×
78
            "Either 'dataset_query' or key-worded arguments must be provided."
79
        )
80

81

82
def load_recipe(dataset_query: Optional[str] = None, **kwargs) -> DatasetRecipe:
1✔
83
    if isinstance(dataset_query, DatasetRecipe):
1✔
84
        return dataset_query
×
85

86
    _verify_dataset_args(dataset_query, kwargs)
1✔
87

88
    if dataset_query:
1✔
89
        recipe = _get_recipe_from_query(dataset_query)
1✔
90

91
    if kwargs:
1✔
92
        recipe = _get_recipe_from_dict(kwargs)
1✔
93

94
    return recipe
1✔
95

96

97
def create_dataset(
1✔
98
    task: Union[str, Task],
99
    test_set: List[Dict[Any, Any]],
100
    train_set: Optional[List[Dict[Any, Any]]] = None,
101
    validation_set: Optional[List[Dict[Any, Any]]] = None,
102
    split: Optional[str] = None,
103
    **kwargs,
104
) -> Union[DatasetDict, IterableDatasetDict, Dataset, IterableDataset]:
105
    """Creates dataset from input data based on a specific task.
106

107
    Args:
108
        task:  The name of the task from the Unitxt Catalog (https://www.unitxt.ai/en/latest/catalog/catalog.tasks.__dir__.html)
109
        test_set : required list of instances
110
        train_set : optional train_set
111
        validation_set: optional validation set
112
        split: optional one split to choose
113
        **kwargs: Arguments used to load dataset from provided datasets (see load_dataset())
114

115
    Returns:
116
        DatasetDict
117

118
    Example:
119
        template = Template(...)
120
        dataset = create_dataset(task="tasks.qa.open", template=template, format="formats.chatapi")
121
    """
122
    data = {"test": test_set}
×
123
    if train_set is not None:
×
124
        data["train"] = train_set
×
125
    if validation_set is not None:
×
126
        data["validation"] = validation_set
×
127
    task, _ = fetch_artifact(task)
×
128

129
    if "template" not in kwargs and task.default_template is None:
×
130
        raise Exception(
×
131
            f"No 'template' was passed to the create_dataset() and the given task ('{task.__id__}') has no 'default_template' field."
132
        )
133

134
    card = TaskCard(loader=LoadFromDictionary(data=data), task=task)
×
135
    return load_dataset(card=card, split=split, **kwargs)
×
136

137

138
def load_dataset(
1✔
139
    dataset_query: Optional[str] = None,
140
    split: Optional[str] = None,
141
    streaming: bool = False,
142
    disable_cache: Optional[bool] = True,
143
    **kwargs,
144
) -> Union[DatasetDict, IterableDatasetDict, Dataset, IterableDataset]:
145
    """Loads dataset.
146

147
    If the 'dataset_query' argument is provided, then dataset is loaded from a card
148
    in local catalog based on parameters specified in the query.
149

150
    Alternatively, dataset is loaded from a provided card based on explicitly
151
    given parameters.
152

153
    Args:
154
        dataset_query (str, optional):
155
            A string query which specifies a dataset to load from
156
            local catalog or name of specific recipe or benchmark in the catalog. For
157
            example, ``"card=cards.wnli,template=templates.classification.multi_class.relation.default"``.
158
        streaming (bool, False):
159
            When True yields the data as Unitxt streams dictionary
160
        split (str, optional):
161
            The split of the data to load
162
        disable_cache (str, optional):
163
            Disable caching process of the data
164
        **kwargs:
165
            Arguments used to load dataset from provided card, which is not present in local catalog.
166

167
    Returns:
168
        DatasetDict
169

170
    :Example:
171

172
        .. code-block:: python
173

174
            dataset = load_dataset(
175
                dataset_query="card=cards.stsb,template=templates.regression.two_texts.simple,max_train_instances=5"
176
            )  # card and template must be present in local catalog
177

178
            # or built programmatically
179
            card = TaskCard(...)
180
            template = Template(...)
181
            loader_limit = 10
182
            dataset = load_dataset(card=card, template=template, loader_limit=loader_limit)
183

184
    """
185
    recipe = load_recipe(dataset_query, **kwargs)
1✔
186

187
    stream = recipe()
1✔
188
    if split is not None:
1✔
189
        stream = stream[split]
1✔
190

191
    if streaming:
1✔
192
        dataset = stream.to_iterable_dataset(
×
193
            features=UNITXT_DATASET_SCHEMA,
194
        ).map(loads_instance, batched=True)
195
    else:
196
        dataset = stream.to_dataset(
1✔
197
            features=UNITXT_DATASET_SCHEMA, disable_cache=disable_cache
198
        ).with_transform(loads_instance)
199

200
    frame = inspect.currentframe()
1✔
201
    args, _, _, values = inspect.getargvalues(frame)
1✔
202
    all_kwargs = {key: values[key] for key in args if key != "kwargs"}
1✔
203
    all_kwargs.update(kwargs)
1✔
204
    metadata = fill_metadata(**all_kwargs)
1✔
205
    if isinstance(dataset, dict):
1✔
206
        for ds in dataset.values():
1✔
207
            ds.info.description = metadata.copy()
1✔
208
    else:
209
        dataset.info.description = metadata
1✔
210
    return dataset
1✔
211

212

213
def fill_metadata(**kwargs):
1✔
214
    metadata = kwargs.copy()
1✔
215
    metadata["unitxt_version"] = get_constants().version
1✔
216
    metadata["creation_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
1✔
217
    return metadata
1✔
218

219

220
def evaluate(
1✔
221
    predictions, dataset: Union[Dataset, IterableDataset] = None, data=None
222
) -> EvaluationResults:
223
    if dataset is None and data is None:
1✔
224
        raise UnitxtError(message="Specify 'dataset' in evaluate")
×
225
    if data is not None:
1✔
226
        dataset = data  # for backward compatibility
1✔
227
    evaluation_result = _compute(predictions=predictions, references=dataset)
1✔
228
    if hasattr(dataset, "info") and hasattr(dataset.info, "description"):
1✔
229
        evaluation_result.metadata["dataset"] = dataset.info.description
1✔
230
    if hasattr(predictions, "metadata"):
1✔
231
        evaluation_result.metadata["predictions"] = predictions.metadata
×
232
    evaluation_result.metadata["creation_time"] = datetime.now().strftime(
1✔
233
        "%Y-%m-%d %H:%M:%S.%f"
234
    )[:-3]
235
    return evaluation_result
1✔
236

237

238
def post_process(predictions, data) -> List[Dict[str, Any]]:
1✔
239
    return _inference_post_process(predictions=predictions, references=data)
1✔
240

241

242
@lru_cache
1✔
243
def _get_produce_with_cache(dataset_query: Optional[str] = None, **kwargs):
1✔
244
    return load_recipe(dataset_query, **kwargs).produce
1✔
245

246

247
def produce(
1✔
248
    instance_or_instances, dataset_query: Optional[str] = None, **kwargs
249
) -> Union[Dataset, Dict[str, Any]]:
250
    is_list = isinstance(instance_or_instances, list)
1✔
251
    if not is_list:
1✔
252
        instance_or_instances = [instance_or_instances]
1✔
253
    result = _get_produce_with_cache(dataset_query, **kwargs)(instance_or_instances)
1✔
254
    if not is_list:
1✔
255
        return result[0]
1✔
256
    return Dataset.from_list(result).with_transform(loads_instance)
1✔
257

258

259
def infer(
1✔
260
    instance_or_instances,
261
    engine: InferenceEngine,
262
    dataset_query: Optional[str] = None,
263
    return_data: bool = False,
264
    return_log_probs: bool = False,
265
    return_meta_data: bool = False,
266
    previous_messages: Optional[List[Dict[str, str]]] = None,
267
    **kwargs,
268
):
269
    dataset = produce(instance_or_instances, dataset_query, **kwargs)
1✔
270
    if previous_messages is not None:
1✔
271

272
        def add_previous_messages(example, index):
×
273
            example["source"] = previous_messages[index] + example["source"]
×
274
            return example
×
275

276
        dataset = dataset.map(add_previous_messages, with_indices=True)
×
277
    engine, _ = fetch_artifact(engine)
1✔
278
    if return_log_probs:
1✔
279
        if not isinstance(engine, LogProbInferenceEngine):
×
280
            raise NotImplementedError(
×
281
                f"Error in infer: return_log_probs set to True but supplied engine "
282
                f"{engine.__class__.__name__} does not support logprobs."
283
            )
284
        infer_outputs = engine.infer_log_probs(dataset, return_meta_data)
×
285
        raw_predictions = (
×
286
            [output.prediction for output in infer_outputs]
287
            if return_meta_data
288
            else infer_outputs
289
        )
290
        raw_predictions = [
×
291
            json.dumps(raw_prediction) for raw_prediction in raw_predictions
292
        ]
293
    else:
294
        infer_outputs = engine.infer(dataset, return_meta_data)
1✔
295
        raw_predictions = (
1✔
296
            [output.prediction for output in infer_outputs]
297
            if return_meta_data
298
            else infer_outputs
299
        )
300
    predictions = post_process(raw_predictions, dataset)
1✔
301
    if return_data:
1✔
302
        if return_meta_data:
1✔
303
            infer_output_list = [
×
304
                infer_output.__dict__ for infer_output in infer_outputs
305
            ]
306
            for infer_output in infer_output_list:
×
307
                del infer_output["prediction"]
×
308
            dataset = dataset.add_column("infer_meta_data", infer_output_list)
×
309
        dataset = dataset.add_column("prediction", predictions)
1✔
310
        return dataset.add_column("raw_prediction", raw_predictions)
1✔
311
    return predictions
1✔
312

313

314
def select(
1✔
315
    instance_or_instances,
316
    engine: OptionSelectingByLogProbsInferenceEngine,
317
    dataset_query: Optional[str] = None,
318
    return_data: bool = False,
319
    previous_messages: Optional[List[Dict[str, str]]] = None,
320
    **kwargs,
321
):
322
    dataset = produce(instance_or_instances, dataset_query, **kwargs)
×
323
    if previous_messages is not None:
×
324

325
        def add_previous_messages(example, index):
×
326
            example["source"] = previous_messages[index] + example["source"]
×
327
            return example
×
328

329
        dataset = dataset.map(add_previous_messages, with_indices=True)
×
330
    engine, _ = fetch_artifact(engine)
×
331
    predictions = engine.select(dataset)
×
332
    # predictions = post_process(raw_predictions, dataset)
333
    if return_data:
×
334
        return dataset.add_column("prediction", predictions)
×
335
    return predictions
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc