code: def compile_forward_sampling_function(
      outputs: List[Variable],
      vars_in_trace: List[Variable],
      basic_rvs: Optional[List[Variable]] = None,
      givens_dict: Optional[Dict[Variable, Any]] = None,
      constant_data: Optional[Dict[str, np.ndarray]] = None,
      constant_coords: Optional[Set[str]] = None,
      **kwargs,
  ) -> Tuple[
      Callable[..., Union[np.ndarray, List[np.ndarray]]], Set[Variable]
  ]:
      """Compile a function to draw samples, conditioned on the values of some variables.
  
      The goal of this function is to walk the pytensor computational graph from the list
      of output nodes down to the root nodes, and then compile a function that will produce
      values for these output nodes. The compiled function will take as inputs the subset of
      variables in the ``vars_in_trace`` that are deemed to not be **volatile**.
  
      Volatile variables are variables whose values could change between runs of the
      compiled function or after inference has been run. These variables are:
  
      - Variables in the outputs list
      - ``SharedVariable`` instances that are not ``RandomStateSharedVariable`` or ``RandomGeneratorSharedVariable``, and whose values changed with respect to what they were at inference time
      - Variables that are in the `basic_rvs` list but not in the ``vars_in_trace`` list
      - Variables that are keys in the ``givens_dict``
      - Variables that have volatile inputs
  
      Concretely, this function can be used to compile a function to sample from the
      posterior predictive distribution of a model that has variables that are conditioned
      on ``MutableData`` instances. The variables that depend on the mutable data that have changed
      will be considered volatile, and as such, they wont be included as inputs into the compiled
      function. This means that if they have values stored in the posterior, these values will be
      ignored and new values will be computed (in the case of deterministics and potentials) or
      sampled (in the case of random variables).
  
      This function also enables a way to impute values for any variable in the computational
      graph that produces the desired outputs: the ``givens_dict``. This dictionary can be used
      to set the ``givens`` argument of the pytensor function compilation. This will essentially
      replace a node in the computational graph with any other expression that has the same
      type as the desired node. Passing variables in the givens_dict is considered an intervention
      that might lead to different variable values from those that could have been seen during
      inference, as such, **any variable that is passed in the ``givens_dict`` will be considered
      volatile**.
  
      Parameters
      ----------
      outputs : List[pytensor.graph.basic.Variable]
          The list of variables that will be returned by the compiled function
      vars_in_trace : List[pytensor.graph.basic.Variable]
          The list of variables that are assumed to have values stored in the trace
      basic_rvs : Optional[List[pytensor.graph.basic.Variable]]
          A list of random variables that are defined in the model. This list (which could be the
          output of ``model.basic_RVs``) should have a reference to the variables that should
          be considered as random variable instances. This includes variables that have
          a ``RandomVariable`` owner op, but also unpure random variables like Mixtures, or
          Censored distributions.
      givens_dict : Optional[Dict[pytensor.graph.basic.Variable, Any]]
          A dictionary that maps tensor variables to the values that should be used to replace them
          in the compiled function. The types of the key and value should match or an error will be
          raised during compilation.
      constant_data : Optional[Dict[str, numpy.ndarray]]
          A dictionary that maps the names of ``MutableData`` or ``ConstantData`` instances to their
          corresponding values at inference time. If a model was created with ``MutableData``, these
          are stored as ``SharedVariable`` with the name of the data variable and a value equal to
          the initial data. At inference time, this information is stored in ``InferenceData``
          objects under the ``constant_data`` group, which allows us to check whether a
          ``SharedVariable`` instance changed its values after inference or not. If the values have
          changed, then the ``SharedVariable`` is assumed to be volatile. If it has not changed, then
          the ``SharedVariable`` is assumed to not be volatile. If a ``SharedVariable`` is not found
          in either ``constant_data`` or ``constant_coords``, then it is assumed to be volatile.
          Setting ``constant_data`` to ``None`` is equivalent to passing an empty dictionary.
      constant_coords : Optional[Set[str]]
          A set with the names of the mutable coordinates that have not changed their shape after
          inference. If a model was created with mutable coordinates, these are stored as
          ``SharedVariable`` with the name of the coordinate and a value equal to the length of said
          coordinate. This set let's us check if a ``SharedVariable`` is a mutated coordinate, in
          which case, it is considered volatile. If a ``SharedVariable`` is not found
          in either ``constant_data`` or ``constant_coords``, then it is assumed to be volatile.
          Setting ``constant_coords`` to ``None`` is equivalent to passing an empty set.
  
      Returns
      -------
      function: Callable
          Compiled forward sampling PyTensor function
      volatile_basic_rvs: Set of Variable
          Set of all basic_rvs that were considered volatile and will be resampled when
          the function is evaluated
      """
      if givens_dict is None:
          givens_dict = {}
  
      if basic_rvs is None:
          basic_rvs = []
  
      if constant_data is None:
          constant_data = {}
      if constant_coords is None:
          constant_coords = set()
  
      # We define a helper function to check if shared values match to an array
      def shared_value_matches(var):
          try:
              old_array_value = constant_data[var.name]
          except KeyError:
              return var.name in constant_coords
          current_shared_value = var.get_value(borrow=True)
          return np.array_equal(old_array_value, current_shared_value)
  
      # We need a function graph to walk the clients and propagate the volatile property
      fg = FunctionGraph(outputs=outputs, clone=False)
  
      # Walk the graph from inputs to outputs and tag the volatile variables
      nodes: List[Variable] = general_toposort(
          fg.outputs, deps=lambda x: x.owner.inputs if x.owner else []
      )
      volatile_nodes: Set[Any] = set()
      for node in nodes:
          if (
              node in fg.outputs
              or node in givens_dict
              or (  # SharedVariables, except RandomState/Generators
                  isinstance(node, SharedVariable)
                  and not isinstance(
                      node,
                      (
                          RandomStateSharedVariable,
                          RandomGeneratorSharedVariable,
                      ),
                  )
                  and not shared_value_matches(node)
              )
              or (  # Basic RVs that are not in the trace
                  node in basic_rvs and node not in vars_in_trace
              )
              or (  # Variables that have any volatile input
                  node.owner
                  and any(
                      inp in volatile_nodes for inp in node.owner.inputs
                  )
              )
          ):
              volatile_nodes.add(node)
  
      # Collect the function inputs by walking the graph from the outputs. Inputs will be:
      # 1. Random variables that are not volatile
      # 2. Variables that have no owner and are not constant or shared
      inputs = []
  
      def expand(node):
          if (
              (
                  node.owner is None
                  and not isinstance(node, (Constant, SharedVariable))
              )  # Variables without owners that are not constant or shared
              or node in vars_in_trace  # Variables in the trace
          ) and node not in volatile_nodes:
              # This test will include variables without owners, and that are not constant
              # or shared, because these nodes will never be considered volatile
              inputs.append(node)
          if node.owner:
              return node.owner.inputs
  
      # walk produces a generator, so we have to actually exhaust the generator in a list to walk
      # the entire graph
      list(walk(fg.outputs, expand))
  
      # Populate the givens list
      givens = [
          (
              node,
              value
              if isinstance(value, (Variable, Apply))
              else pt.constant(
                  value,
                  dtype=getattr(node, "dtype", None),
                  name=node.name,
              ),
          )
          for node, value in givens_dict.items()
      ]
  
      return (
          compile_pymc(
              inputs,
              fg.outputs,
              givens=givens,
              on_unused_input="ignore",
              **kwargs,
          ),
          set(basic_rvs)
          & (
              volatile_nodes - set(givens_dict)
          ),  # Basic RVs that will be resampled
      )
  
imports: import logging
  import warnings
  from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Set, Tuple, Union, cast
  import numpy
  import xarray
  from arviz import InferenceData
  from fastprogress.fastprogress import progress_bar
  from pytensor import tensor
  from pytensor.graph.basic import Apply, Constant, Variable, ancestors, general_toposort, walk
  from pytensor.graph.fg import FunctionGraph
  from pytensor.tensor.random.var import RandomGeneratorSharedVariable, RandomStateSharedVariable
  from pytensor.tensor.sharedvar import SharedVariable
  from typing_extensions import TypeAlias
  import pymc
  from pymc.backends.arviz import _DefaultTrace
  from pymc.backends.base import MultiTrace
  from pymc.blocking import PointType
  from pymc.model import Model, modelcontext
  from pymc.pytensorf import compile_pymc
  from pymc.util import RandomState, _get_seeds_per_chain, dataset_to_point_list, get_default_varnames, point_wrapper
label_desc: Use image node_id 2 for calling a global function with example usage: compile_forward_sampling_function(outputs, vars_in_trace, basic_rvs, givens_dict, constant_data, constant_coords) and returns: compile_pymc,  unknown, np, node, unknown