code: def induced_subgraph(
      input_graph,
      vertices: Union[cudf.Series, cudf.DataFrame],
      offsets: Union[list, cudf.Series] = None,
  ) -> Tuple[dask_cudf.DataFrame, dask_cudf.Series]:
      """
      Compute a subgraph of the existing graph including only the specified
      vertices.  This algorithm works with both directed and undirected graphs
      and does not actually traverse the edges, but instead simply pulls out any
      edges that are incident on vertices that are both contained in the vertices
      list.
  
      If no subgraph can be extracted from the vertices provided, a 'None' value
      will be returned.
  
      Parameters
      ----------
      input_graph : cugraph.Graph
          Graph or matrix object, which should contain the connectivity
          information. Edge weights, if present, should be single or double
          precision floating point values.
  
      vertices : cudf.Series or cudf.DataFrame
          Specifies the vertices of the induced subgraph. For multi-column
          vertices, vertices should be provided as a cudf.DataFrame
  
      offsets : list or cudf.Series, optional
          Specifies the subgraph offsets into the subgraph vertices.
          If no offsets array is provided, a default array [0, len(vertices)]
          will be used.
  
      Returns
      -------
      ego_edge_lists : dask_cudf.DataFrame
          Distributed GPU data frame containing all induced sources identifiers,
          destination identifiers, edge weights
      seeds_offsets: dask_cudf.Series
          Distributed Series containing the starting offset in the returned edge list
          for each seed.
  
      """
  
      # Initialize dask client
      client = default_client()
  
      if isinstance(vertices, (int, list)):
          vertices = cudf.Series(vertices)
      elif not isinstance(
          vertices,
          (
              cudf.Series,
              dask_cudf.Series,
              cudf.DataFrame,
              dask_cudf.DataFrame,
          ),
      ):
          raise TypeError(
              f"'vertices' must be either an integer or a list or a "
              f"cudf or dask_cudf Series or DataFrame, got: {type(vertices)}"
          )
  
      if isinstance(offsets, list):
          offsets = cudf.Series(offsets)
  
      if offsets is None:
          offsets = cudf.Series([0, len(vertices)])
  
      if not isinstance(offsets, cudf.Series):
          raise TypeError(
              f"'offsets' must be either 'None', a list or a "
              f"cudf Series, got: {type(offsets)}"
          )
  
      # vertices uses "external" vertex IDs, but since the graph has been
      # renumbered, the node ID must also be renumbered.
      if input_graph.renumbered:
          vertices = input_graph.lookup_internal_vertex_id(vertices)
          vertices_type = input_graph.edgelist.edgelist_df.dtypes[0]
      else:
          vertices_type = input_graph.input_df.dtypes[0]
  
      if isinstance(vertices, (cudf.Series, cudf.DataFrame)):
          vertices = dask_cudf.from_cudf(
              vertices, npartitions=input_graph._npartitions
          )
      vertices = vertices.astype(vertices_type)
  
      vertices = persist_dask_df_equal_parts_per_worker(
          vertices, client, return_type="dict"
      )
  
      do_expensive_check = False
  
      result = [
          client.submit(
              _call_induced_subgraph,
              Comms.get_session_id(),
              input_graph._plc_graph[w],
              vertices_[0]
              if vertices_
              else cudf.Series(dtype=vertices_type),
              offsets,
              do_expensive_check,
              workers=[w],
              allow_other_workers=False,
          )
          for w, vertices_ in vertices.items()
      ]
      wait(result)
  
      cudf_result = [
          client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in result
      ]
  
      wait(cudf_result)
  
      ddf = dask_cudf.from_delayed(cudf_result).persist()
      wait(ddf)
  
      if len(ddf) == 0:
          return None, None
  
      wait(
          [
              (r.release(), c_r.release())
              for r, c_r in zip(result, cudf_result)
          ]
      )
  
      ddf = ddf.sort_values("labels")
  
      # extract offsets from segmented induced_subgraph dataframes
      offsets = ddf["labels"].value_counts().compute().sort_index()
      offsets = cudf.concat([cudf.Series(0), offsets])
      offsets = (
          dask_cudf.from_cudf(
              offsets,
              npartitions=min(input_graph._npartitions, len(vertices)),
          )
          .cumsum()
          .astype(vertices_type)
      )
  
      ddf = ddf.drop(columns="labels")
  
      if input_graph.renumbered:
          ddf = input_graph.unrenumber(ddf, "src")
          ddf = input_graph.unrenumber(ddf, "dst")
  
      return ddf, offsets
  
file_path: cugraph/python/cugraph/cugraph/dask/community/induced_subgraph.py
imports: from dask.distributed import wait, default_client
  import cugraph.dask.comms.comms
  import dask_cudf
  import cudf
  import cupy
  from cugraph.dask.common.part_utils import persist_dask_df_equal_parts_per_worker
  from typing import Union, Tuple
  from pylibcugraph import ResourceHandle, induced_subgraph
label_desc: Use image node_id 4 for calling a global function with example usage: induced_subgraph(input_graph, vertices, offsets) and returns: ddf,  offsets, None,  None