code: def _reduce_graph_fn(ctx: _Context, node_id: str, graph_fn: _GraphFn):
      """Construct a new `GraphFn` with `GraphFn` and an `InputGraph.Node`.
  
      If `graph_fn` has a signature of ({x, y}) -> z and a node (of `node_id`) has
      a signature of ({w, v}) -> x, then the reduced graph function (return value)
      has a signature of ({w, v, y}) -> z.
  
      `_reduce_graph_fn` assumes the dependent nodes of the `node_id` is not yet
      reduced into the current `graph_fn`, so nodes must be reduced in a reversed
      topologically sorted order. This ensures each node is invoked exactly once in
      a correct order.
  
      Args:
        ctx: Current `_Context`.
        node_id: A `InputGraph.Node` ID.
        graph_fn: An accumulated `GraphFn`.
  
      Returns:
        A reduced `GraphFn`.
      """
      node_def = ctx.input_graph.nodes[node_id]
      if node_def.WhichOneof("kind") == "op_node":
          node_fn = functools.partial(_evaluate_op_node, ctx, node_id)
      elif node_def.WhichOneof("kind") == "dict_node":
          node_fn = functools.partial(_evaluate_dict_node, ctx, node_id)
      else:
          # Skip for kind == 'input_node' as the final facade graph function take
          # cares of them.
          return graph_fn
  
      def new_graph_fn(data: Mapping[str, _Data]):
          try:
              output = node_fn(data)
          except (
              exceptions.InputResolutionError,
              exceptions.InputResolutionSignal,
          ):  # pylint: disable=try-except-raise
              raise
          except Exception as e:  # pylint: disable=broad-except
              raise exceptions.InternalError(
                  "Internal error occurred during evaluating input_graph.nodes["
                  f"{node_id}]."
              ) from e
          return graph_fn({**data, node_id: output})
  
      return new_graph_fn
  
imports: import collections
  import dataclasses
  import functools
  from typing import Union, Sequence, Mapping, Tuple, List, Iterable, Callable
  from tfx import types
  from tfx.dsl.components.common import resolver
  from tfx.dsl.input_resolution import resolver_op
  from tfx.dsl.input_resolution.ops import ops
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import topsort
  from tfx.utils import typing_utils
label_desc: Use image node_id 5 for calling a global function with example usage: _reduce_graph_fn(ctx, node_id, graph_fn) and returns: new_graph_fn, graph_fn, graph_fn