(self, inputs: list[dict[str, Any]],
dataset: lit_dataset.Dataset,
dataset_spec_to_annotate: Optional[types.Spec] = None)
| 37 | """ |
| 38 | |
| 39 | def annotate(self, inputs: list[dict[str, Any]], |
| 40 | dataset: lit_dataset.Dataset, |
| 41 | dataset_spec_to_annotate: Optional[types.Spec] = None): |
| 42 | if len(self._annotator_model.input_spec().items()) != 1: |
| 43 | raise ValueError('Annotator model provided to PerFieldAnnotator does not ' |
| 44 | 'operate on a single field') |
| 45 | |
| 46 | datasets = {} |
| 47 | for input_name, input_type in self._annotator_model.input_spec().items(): |
| 48 | # Do remap of inputs based on input name needed by annotator. |
| 49 | ds_keys = utils.find_spec_keys(dataset.spec(), type(input_type)) |
| 50 | for ds_key in ds_keys: |
| 51 | temp_ds = lit_dataset.Dataset(examples=inputs, base=dataset) |
| 52 | datasets[ds_key] = temp_ds.remap({ds_key: input_name}) |
| 53 | |
| 54 | for ds_key, ds in datasets.items(): |
| 55 | outputs = self._annotator_model.predict(ds.examples) |
| 56 | for output_name, output_type in self._annotator_model.output_spec( |
| 57 | ).items(): |
| 58 | # Update dataset spec with new annotated field. |
| 59 | field_name = f'{self._name}:{output_name}:{ds_key}' |
| 60 | if dataset_spec_to_annotate: |
| 61 | dataset_spec_to_annotate[field_name] = attr.evolve( |
| 62 | output_type, annotated=True) |
| 63 | |
| 64 | # Update all examples with annotator output. |
| 65 | for example, output in zip(inputs, outputs): |
| 66 | example[field_name] = output[output_name] |
nothing calls this directly
no test coverage detected