code: def conditional_logp(
      rv_values: Dict[TensorVariable, TensorVariable],
      warn_rvs=None,
      ir_rewriter: Optional[GraphRewriter] = None,
      extra_rewrites: Optional[
          Union[GraphRewriter, NodeRewriter]
      ] = None,
      **kwargs,
  ) -> Dict[TensorVariable, TensorVariable]:
      r"""Create a map between variables and conditional log-probabilities
      such that the sum is their joint log-probability.
  
      The `rv_values` dictionary specifies a joint probability graph defined by
      pairs of random variables and respective measure-space input parameters
  
      For example, consider the following
  
      .. code-block:: python
  
          import pytensor.tensor as pt
  
          sigma2_rv = pt.random.invgamma(0.5, 0.5)
          Y_rv = pt.random.normal(0, pt.sqrt(sigma2_rv))
  
      This graph for ``Y_rv`` is equivalent to the following hierarchical model:
  
      .. math::
  
          \sigma^2 \sim& \operatorname{InvGamma}(0.5, 0.5) \\
          Y \sim& \operatorname{N}(0, \sigma^2)
  
      If we create a value variable for ``Y_rv``, i.e. ``y_vv = pt.scalar("y")``,
      the graph of ``conditional_logp({Y_rv: y_vv})`` is equivalent to the
      conditional log-probability :math:`\log p_{Y \mid \sigma^2}(y \mid s^2)`, with a stochastic
      ``sigma2_rv``.
  
      If we specify a value variable for ``sigma2_rv``, i.e.
      ``s2_vv = pt.scalar("s2")``, then ``conditional_logp({Y_rv: y_vv, sigma2_rv: s2_vv})``
      yields the conditional log-probabilities of the two variables.
      The sum of the two terms gives their joint log-probability.
  
      .. math::
  
          \log p_{Y, \sigma^2}(y, s^2) =
              \log p_{Y \mid \sigma^2}(y \mid s^2) + \log p_{\sigma^2}(s^2)
  
  
      Parameters
      ----------
      rv_values: dict
          A ``dict`` of variables that maps stochastic elements
          (e.g. `RandomVariable`\s) to symbolic `Variable`\s representing their
          values in a log-probability.
      warn_rvs : bool, default True
          When ``True``, issue a warning when a `RandomVariable` is found in
          the logp graph and doesn't have a corresponding value variable specified in
          `rv_values`.
      ir_rewriter
          Rewriter that produces the intermediate representation of Measurable Variables.
      extra_rewrites
          Extra rewrites to be applied (e.g. reparameterizations, transforms,
          etc.)
  
      Returns
      -------
      values_to_logps: dict
          A ``dict`` that maps each value variable to the conditional log-probability term derived
          from the respective `RandomVariable`.
  
      """
      warn_rvs, kwargs = _deprecate_warn_missing_rvs(warn_rvs, kwargs)
  
      fgraph, rv_values, _ = construct_ir_fgraph(
          rv_values, ir_rewriter=ir_rewriter
      )
  
      if extra_rewrites is not None:
          extra_rewrites.rewrite(fgraph)
  
      rv_remapper = fgraph.preserve_rv_mappings
  
      # This is the updated random-to-value-vars map with the lifted/rewritten
      # variables.  The rewrites are supposed to produce new
      # `MeasurableVariable`s that are amenable to `_logprob`.
      updated_rv_values = rv_remapper.rv_values
  
      # Some rewrites also transform the original value variables. This is the
      # updated map from the new value variables to the original ones, which
      # we want to use as the keys in the final dictionary output
      original_values = rv_remapper.original_values
  
      # When a `_logprob` has been produced for a `MeasurableVariable` node, all
      # other references to it need to be replaced with its value-variable all
      # throughout the `_logprob`-produced graphs.  The following `dict`
      # cumulatively maintains remappings for all the variables/nodes that needed
      # to be recreated after replacing `MeasurableVariable`s with their
      # value-variables.  Since these replacements work in topological order, all
      # the necessary value-variable replacements should be present for each
      # node.
      replacements = updated_rv_values.copy()
  
      # To avoid cloning the value variables (or ancestors of value variables),
      # we map them to themselves in the `replacements` `dict`
      # (i.e. entries already existing in `replacements` aren't cloned)
      replacements.update(
          {
              v: v
              for v in ancestors(rv_values.values())
              if (not isinstance(v, Constant) and v not in replacements)
          }
      )
  
      # Walk the graph from its inputs to its outputs and construct the
      # log-probability
      q = deque(fgraph.toposort())
      logprob_vars = {}
  
      while q:
          node = q.popleft()
  
          if not isinstance(node.op, MeasurableVariable):
              continue
  
          q_values = [
              replacements[q_rv]
              for q_rv in node.outputs
              if q_rv in updated_rv_values
          ]
  
          if not q_values:
              continue
  
          # Replace `RandomVariable`s in the inputs with value variables.
          remapped_vars = replace_vars_in_graphs(
              graphs=q_values + list(node.inputs),
              replacements=replacements,
          )
          q_values = remapped_vars[: len(q_values)]
          q_rv_inputs = remapped_vars[len(q_values) :]
  
          q_logprob_vars = _logprob(
              node.op,
              q_values,
              *q_rv_inputs,
              **kwargs,
          )
  
          if not isinstance(q_logprob_vars, (list, tuple)):
              q_logprob_vars = [q_logprob_vars]
  
          for q_value_var, q_logprob_var in zip(
              q_values, q_logprob_vars
          ):
              q_value_var = original_values[q_value_var]
  
              if q_value_var.name:
                  q_logprob_var.name = f"{q_value_var.name}_logprob"
  
              if q_value_var in logprob_vars:
                  raise ValueError(
                      f"More than one logprob term was assigned to the value var {q_value_var}"
                  )
  
              logprob_vars[q_value_var] = q_logprob_var
  
          # Recompute test values for the changes introduced by the replacements above.
          if config.compute_test_value != "off":
              for node in io_toposort(
                  graph_inputs(q_logprob_vars), q_logprob_vars
              ):
                  compute_test_value(node)
  
      missing_value_terms = set(original_values.values()) - set(
          logprob_vars.keys()
      )
      if missing_value_terms:
          raise RuntimeError(
              f"The logprob terms of the following value variables could not be derived: {missing_value_terms}"
          )
  
      logprob_expressions = list(logprob_vars.values())
      cleanup_ir(logprob_expressions)
  
      if warn_rvs:
          rvs_in_logp_expressions = _find_unallowed_rvs_in_graph(
              logprob_expressions
          )
          if rvs_in_logp_expressions:
              warnings.warn(
                  RVS_IN_JOINT_LOGP_GRAPH_MSG % rvs_in_logp_expressions,
                  UserWarning,
              )
  
      return logprob_vars
  
imports: import warnings
  from collections import deque
  from typing import Dict, List, Optional, Sequence, Union
  import numpy
  import pytensor.tensor
  from pytensor import config
  from pytensor.graph.basic import Constant, Variable, ancestors, graph_inputs, io_toposort
  from pytensor.graph.op import compute_test_value
  from pytensor.graph.rewriting.basic import GraphRewriter, NodeRewriter
  from pytensor.tensor.variable import TensorVariable
  from typing_extensions import TypeAlias
  from pymc.logprob.abstract import MeasurableVariable, _icdf_helper, _logcdf_helper, _logprob, _logprob_helper
  from pymc.logprob.rewriting import cleanup_ir, construct_ir_fgraph
  from pymc.logprob.transform_value import TransformValuesRewrite
  from pymc.logprob.transforms import Transform
  from pymc.logprob.utils import rvs_in_graph
  from pymc.pytensorf import replace_vars_in_graphs
label_desc: Use image node_id 7 for calling a global function with example usage: conditional_logp(rv_values, warn_rvs, ir_rewriter, extra_rewrites) and returns: logprob_vars