class_docstr: A wrapper for `seqio.Task` using a `seqio.FunctionDataSource`.
  
  For backward compatibility with the original T5 task format, this task
  separates preprocessing into text and token stages, with optional caching in
  between.
class_docstr_tok: ["A","wrapper","for","`","seqio.Task","`","using","a","`","seqio.FunctionDataSource","`",".","For","backward","compatibility","with","the","original","T5","task","format",",","this","task","separates","preprocessing","into","text","and","token","stages",",","with","optional","caching","in","between","."]
code: def __init__(
      self,
      name,
      dataset_fn,
      splits,
      text_preprocessor,
      metric_fns=None,
      postprocess_fn=None,
      token_preprocessor=None,
      output_features=None,
      num_input_examples=None,
      supports_caching=True,
      shuffle_buffer_size=SHUFFLE_BUFFER_SIZE,
      source=None,
  ):
      if (dataset_fn, source).count(None) != 1:
          raise ValueError(
              "Exactly one of either `dataset_fn` or `source` must be provided."
          )
  
      if source and (splits or num_input_examples):
          raise ValueError(
              "If `source` is provided, `splits` and `num_input_examples` should "
              "not also be provided to the Task."
          )
      source = source or seqio.FunctionDataSource(
          dataset_fn=dataset_fn,
          splits=splits,
          num_input_examples=num_input_examples,
      )
  
      if text_preprocessor and not hasattr(
          text_preprocessor, "__iter__"
      ):
          text_preprocessor = [text_preprocessor]
      if token_preprocessor and not hasattr(
          token_preprocessor, "__iter__"
      ):
          token_preprocessor = [token_preprocessor]
  
      preprocessors = list(text_preprocessor or [])
      preprocessors.append(seqio.preprocessors.tokenize)
      if supports_caching:
          preprocessors.append(seqio.CacheDatasetPlaceholder())
      preprocessors.extend(token_preprocessor or [])
      preprocessors.append(seqio.preprocessors.append_eos_after_trim)
  
      if hasattr(output_features, "__len__") and not output_features:
          raise ValueError("output_features must be non-empty.")
      if output_features is None:
          output_features = seqio.Feature(
              utils.get_default_vocabulary()
          )
      if isinstance(output_features, Mapping):
          pass
      elif isinstance(output_features, seqio.Feature):
          output_features = {
              k: output_features for k in _DEFAULT_FEATURE_KEYS
          }
      elif isinstance(output_features, list) and all(
          isinstance(f, str) for f in output_features
      ):
          output_features = {
              k: seqio.Feature(utils.get_default_vocabulary())
              for k in output_features
          }
      else:
          raise ValueError(
              "output_features must be a dict, Feature, list of str, or None"
          )
  
      if hasattr(postprocess_fn, "__iter__"):
          postprocess_fns = postprocess_fn
  
          def postprocess_fn(
              x, **postprocess_kwargs
          ):  # pylint:disable=function-redefined
              for post_fn in postprocess_fns:
                  x = post_fn(x, **postprocess_kwargs)
              return x
  
      super().__init__(
          name=name,
          source=source,
          output_features=output_features,
          preprocessors=preprocessors,
          postprocess_fn=postprocess_fn,
          metric_fns=metric_fns,
          shuffle_buffer_size=shuffle_buffer_size,
      )
  
imports: from collections.abc import Mapping
  import re
  import seqio
  from t5.data import utils
  import tensorflow.compat.v2
label_desc: Use image node_id 1 to create a new FunctionTask object from inherited base classes: seqio with example: obj = FunctionTask(name, dataset_fn, splits, text_preprocessor, metric_fns, postprocess_fn, token_preprocessor, output_features, num_input_examples, supports_caching, shuffle_buffer_size, source)