code: def groupby_agg(
      ddf,
      gb_cols,
      aggs_in,
      split_every=None,
      split_out=None,
      dropna=True,
      sep="___",
      sort=False,
      as_index=True,
      shuffle_method=None,
  ):
      """Optimized groupby aggregation for Dask-CuDF.
  
      Parameters
      ----------
      ddf : DataFrame
          DataFrame object to perform grouping on.
      gb_cols : str or list[str]
          Column names to group by.
      aggs_in : str, list, or dict
          Aggregations to perform.
      split_every : int (optional)
          How to group intermediate aggregates.
      dropna : bool
          Drop grouping key values corresponding to NA values.
      as_index : bool
          Currently ignored.
      sort : bool
          Sort the group keys, better performance is obtained when
          not sorting.
      shuffle_method : str (optional)
          Control how shuffling of the DataFrame is performed.
      sep : str
          Internal usage.
  
  
      Notes
      -----
      This "optimized" approach is more performant than the algorithm in
      implemented in :meth:`DataFrame.apply` because it allows the cuDF
      backend to perform multiple aggregations at once.
  
      This aggregation algorithm only supports the following options
  
      * "collect"
      * "count"
      * "first"
      * "last"
      * "max"
      * "mean"
      * "min"
      * "std"
      * "sum"
      * "var"
  
  
      See Also
      --------
      DataFrame.groupby : generic groupby of a DataFrame
      dask.dataframe.apply_concat_apply : for more description of the
          split_every argument.
  
      """
      # Assert that aggregations are supported
      aggs = _redirect_aggs(aggs_in)
      if not _aggs_optimized(aggs, OPTIMIZED_AGGS):
          raise ValueError(
              f"Supported aggs include {OPTIMIZED_AGGS} for groupby_agg API. "
              f"Aggregations must be specified with dict or list syntax."
          )
  
      # If split_every is False, we use an all-to-one reduction
      if split_every is False:
          split_every = max(ddf.npartitions, 2)
  
      # Deal with default split_out and split_every params
      split_every = split_every or 8
      split_out = split_out or 1
  
      # Standardize `gb_cols`, `columns`, and `aggs`
      if isinstance(gb_cols, str):
          gb_cols = [gb_cols]
      columns = [c for c in ddf.columns if c not in gb_cols]
      if not isinstance(aggs, dict):
          aggs = {col: aggs for col in columns}
  
      # Assert if our output will have a MultiIndex; this will be the case if
      # any value in the `aggs` dict is not a string (i.e. multiple/named
      # aggregations per column)
      str_cols_out = True
      aggs_renames = {}
      for col in aggs:
          if isinstance(aggs[col], str) or callable(aggs[col]):
              aggs[col] = [aggs[col]]
          elif isinstance(aggs[col], dict):
              str_cols_out = False
              col_aggs = []
              for k, v in aggs[col].items():
                  aggs_renames[col, v] = k
                  col_aggs.append(v)
              aggs[col] = col_aggs
          else:
              str_cols_out = False
          if col in gb_cols:
              columns.append(col)
  
      # Construct meta
      _aggs = aggs.copy()
      if str_cols_out:
          # Metadata should use `str` for dict values if that is
          # what the user originally specified (column names will
          # be str, rather than tuples).
          for col in aggs:
              _aggs[col] = _aggs[col][0]
      _meta = ddf._meta.groupby(gb_cols, as_index=as_index).agg(_aggs)
      if aggs_renames:
          col_array = []
          agg_array = []
          for col, agg in _meta.columns:
              col_array.append(col)
              agg_array.append(aggs_renames.get((col, agg), agg))
          _meta.columns = pd.MultiIndex.from_arrays(
              [col_array, agg_array]
          )
  
      chunk = _groupby_partition_agg
      chunk_kwargs = {
          "gb_cols": gb_cols,
          "aggs": aggs,
          "columns": columns,
          "dropna": dropna,
          "sort": sort,
          "sep": sep,
      }
  
      combine = _tree_node_agg
      combine_kwargs = {
          "gb_cols": gb_cols,
          "dropna": dropna,
          "sort": sort,
          "sep": sep,
      }
  
      aggregate = _finalize_gb_agg
      aggregate_kwargs = {
          "gb_cols": gb_cols,
          "aggs": aggs,
          "columns": columns,
          "final_columns": _meta.columns,
          "as_index": as_index,
          "dropna": dropna,
          "sort": sort,
          "sep": sep,
          "str_cols_out": str_cols_out,
          "aggs_renames": aggs_renames,
      }
  
      # Use shuffle_method=True for split_out>1
      if sort and split_out > 1 and shuffle_method is None:
          shuffle_method = "tasks"
  
      # Check if we are using the shuffle-based algorithm
      if shuffle_method:
          # Shuffle-based aggregation
          return _shuffle_aggregate(
              ddf,
              gb_cols,
              chunk,
              chunk_kwargs,
              aggregate,
              aggregate_kwargs,
              split_every,
              split_out,
              token="cudf-aggregate",
              sort=sort,
              shuffle_method=shuffle_method
              if isinstance(shuffle_method, str)
              else None,
          )
  
      # Deal with sort/shuffle defaults
      if split_out > 1 and sort:
          raise ValueError(
              "dask-cudf's groupby algorithm does not yet support "
              "`sort=True` when `split_out>1`, unless a shuffle-based "
              "algorithm is used. Please use `split_out=1`, group "
              "with `sort=False`, or set `shuffle_method=True`."
          )
  
      # Determine required columns to enable column projection
      required_columns = list(
          set(gb_cols).union(aggs.keys()).intersection(ddf.columns)
      )
  
      return aca(
          [ddf[required_columns]],
          chunk=chunk,
          chunk_kwargs=chunk_kwargs,
          combine=combine,
          combine_kwargs=combine_kwargs,
          aggregate=aggregate,
          aggregate_kwargs=aggregate_kwargs,
          token="cudf-aggregate",
          split_every=split_every,
          split_out=split_out,
          split_out_setup=split_out_on_cols,
          split_out_setup_kwargs={"cols": gb_cols},
          sort=sort,
          ignore_index=True,
      )
  
imports: from functools import wraps
  from typing import Set
  import numpy
  import pandas
  from dask.dataframe.core import DataFrame, aca, split_out_on_cols
  from dask.dataframe.groupby import DataFrameGroupBy, SeriesGroupBy
  from dask.utils import funcname
  import cudf
  from cudf.utils.nvtx_annotation import _dask_cudf_nvtx_annotate
  from dask_cudf.sorting import _deprecate_shuffle_kwarg
label_desc: Use image node_id 3 for calling a global function with example usage: groupby_agg(ddf, gb_cols, aggs_in, split_every, split_out, dropna, sep, sort, as_index, shuffle_method) and returns: aca, _shuffle_aggregate