def _approximate_vocabulary_analyzer_nodes(
    analyzer_inputs: Collection[tf.Tensor],
    input_dtype: tf.dtypes.DType,
    vocab_filename: str,
    top_k: int,
    store_frequency: bool,
    file_format: common_types.VocabularyFileFormatType,
    vocabulary_key: str,
    reserved_tokens: Optional[Union[Sequence[str], tf.Tensor]] = None,
) -> common_types.TemporaryAnalyzerOutputType:
  """Internal helper for analyzing vocab. See `vocabulary` doc string."""
  # TODO(b/208879020): Add vocabulary size annotation for this analyzer.
  analyzers.register_vocab(
      vocab_filename, vocabulary_key=vocabulary_key, file_format=file_format)

  outputs_value_nodes = analyzers.apply_cacheable_combine_operation(
      _VocabularyCombiner(
          top_k, input_dtype, output_pylist=reserved_tokens is not None
      ),
      *analyzer_inputs
  )

  flattened_outputs_value_node = nodes.apply_operation(
      analyzer_nodes.FlattenLists, *outputs_value_nodes)

  extra_apply_order_and_write_op_args = []
  if reserved_tokens is not None:
    tf_utils.register_vocabulary_reserved_tokens(
        vocab_filename, reserved_tokens
    )
    extra_apply_order_and_write_op_args.append(
        nodes.apply_operation(
            analyzer_nodes.ExtractVocabularyReservedTokens, name=vocab_filename
        )
    )

  vocab_filename_node = nodes.apply_operation(
      analyzer_nodes.VocabularyOrderAndWrite,
      flattened_outputs_value_node,
      *extra_apply_order_and_write_op_args,
      vocab_filename=vocab_filename,
      store_frequency=store_frequency,
      input_dtype=input_dtype,
      file_format=file_format,
      fingerprint_shuffle=False,
      input_is_sorted=True
  )

  return analyzer_nodes.wrap_as_tensor(vocab_filename_node)
Internal helper for analyzing vocab. See `vocabulary` doc string.
tensorflow_transform/experimental/analyzers.py