class_docstr: ResolverFunction represents a traceable function of resolver operators.
  
  Resolver function returns some form of channel depending on the function
  definition.
  
  It can return a single channel:
  
      trainer = Trainer(
          examples=latest_created(example_gen.outputs['examples']))
  
  or a dictionary of channels:
  
      k_fold_inputs = k_fold(example_gen.outputs['examples'], splits=5)
      trainer = Trainer(
          examples=k_fold_inputs['train'],
      )
      evaluator = Evaluator(
          examples=k_fold_inputs['eval'],
          model=trainer.outputs['model'],
      )
  
  or a ForEach-loopable channels:
  
    with ForEach(
        tfx.dsl.inputs.sequential_rolling_range(
            example_gens.outputs['examples'], n=3)) as train_window:
      trainer = Trainer(
          examples=train_window['examples'])
docstr: Trace resolver function with given node arguments.
  
  Do not call this function directly; Use __call__ only.
  
  Tracing happens by substituting the input arguments (from BaseChannel to
  InputNode) and calling the inner python function. Traced result is the
  return value of the inner python function. Since ResolverOp invocation
  stores all the input arguments (which originated from InputNode), we can
  analyze the full ResolverOp invocation graph from the return value.
  
  Trace happens only once during the resolver function invocation. Traced
  resolver function (which is a resolver_op.Node) is serialized to the
  pipeline IR during compilation (i.e. inner python function is serialized),
  and the inner python function is not invoked again on IR interpretation.
  
  Args:
    *args: Substituted arguments to the resolver function.
    **kwargs: Substitued keyword arguments to the resolver function.
  
  Raises:
    RuntimeError: if the tracing fails.
  
  Returns:
    A traced result, which is a resolver_op.Node.
docstr_tok: ["Trace","resolver","function","with","given","node","arguments",".","Do","not","call","this","function","directly",";","Use","__call__","only",".","Tracing","happens","by","substituting","the","input","arguments","(","from","BaseChannel","to","InputNode",")","and","calling","the","inner","python","function",".","Traced","result","is","the","return","value","of","the","inner","python","function",".","Since","ResolverOp","invocation","stores","all","the","input","arguments","(","which","originated","from","InputNode",")",",","we","can","analyze","the","full","ResolverOp","invocation","graph","from","the","return","value",".","Trace","happens","only","once","during","the","resolver","function","invocation",".","Traced","resolver","function","(","which","is","a","resolver_op.Node",")","is","serialized","to","the","pipeline","IR","during","compilation","(","i.e",".","inner","python","function","is","serialized",")",",","and","the","inner","python","function","is","not","invoked","again","on","IR","interpretation",".","Args",":","*","args",":","Substituted","arguments","to","the","resolver","function",".","*","*","kwargs",":","Substitued","keyword","arguments","to","the","resolver","function",".","Raises",":","RuntimeError",":","if","the","tracing","fails",".","Returns",":","A","traced","result",",","which","is","a","resolver_op.Node","."]
code: def trace(
      self, *args: resolver_op.Node, **kwargs: Any
  ) -> resolver_op.Node:
      """Trace resolver function with given node arguments.
  
      Do not call this function directly; Use __call__ only.
  
      Tracing happens by substituting the input arguments (from BaseChannel to
      InputNode) and calling the inner python function. Traced result is the
      return value of the inner python function. Since ResolverOp invocation
      stores all the input arguments (which originated from InputNode), we can
      analyze the full ResolverOp invocation graph from the return value.
  
      Trace happens only once during the resolver function invocation. Traced
      resolver function (which is a resolver_op.Node) is serialized to the
      pipeline IR during compilation (i.e. inner python function is serialized),
      and the inner python function is not invoked again on IR interpretation.
  
      Args:
        *args: Substituted arguments to the resolver function.
        **kwargs: Substitued keyword arguments to the resolver function.
  
      Raises:
        RuntimeError: if the tracing fails.
  
      Returns:
        A traced result, which is a resolver_op.Node.
      """
      # TODO(b/188023509): Better debug support & error message.
      result = self.__wrapped__(*args, **kwargs)
      if typing_utils.is_compatible(
          result, Mapping[str, resolver_op.Node]
      ):
          result = resolver_op.DictNode(
              cast(Mapping[str, resolver_op.Node], result)
          )
      if not isinstance(result, resolver_op.Node):
          raise RuntimeError(
              f"Invalid resolver function trace result {result}. Expected to "
              "return an output of ResolverOp or a dict of outputs."
          )
      return result
  
imports: import contextlib
  import inspect
  from typing import Callable, Type, Union, Mapping, Any, Optional, Sequence, cast, overload
  from tfx.dsl.control_flow import for_each_internal
  from tfx.dsl.input_resolution import resolver_op
  from tfx.types import artifact
  from tfx.types import channel
  from tfx.types import resolved_channel
  from tfx.utils import doc_controls
  from tfx.utils import typing_utils
label_desc: Use image node_id 7 for calling the ResolverFunction obj's underlying member method code with example usage: obj.trace() and returns: result