docs: def create_graph(expression_matrix,
                   gene_names,
                   tf_names,
                   regressor_type,
                   regressor_kwargs,
                   client,
                   target_genes='all',
                   limit=None,
                   include_meta=False,
                   early_stop_window_length=EARLY_STOP_WINDOW_LENGTH,
                   repartition_multiplier=1,
                   seed=DEMON_SEED):
      """
      Main API function. Create a Dask computation graph.
  
      Note: fixing the GC problems was fixed by 2 changes: [1] and [2] !!!
  
      :param expression_matrix: numpy matrix. Rows are observations and columns are genes.
      :param gene_names: list of gene names. Each entry corresponds to the expression_matrix column with same index.
      :param tf_names: list of transcription factor names. Should have a non-empty intersection with gene_names.
      :param regressor_type: regressor type. Case insensitive.
      :param regressor_kwargs: dict of key-value pairs that configures the regressor.
      :param client: a dask.distributed client instance.
                     * Used to scatter-broadcast the tf matrix to the workers instead of simply wrapping in a delayed().
      :param target_genes: either int, 'all' or a collection that is a subset of gene_names.
      :param limit: optional number of top regulatory links to return. Default None.
      :param include_meta: Also return the meta DataFrame. Default False.
      :param early_stop_window_length: window length of the early stopping monitor.
      :param repartition_multiplier: multiplier
      :param seed: (optional) random seed for the regressors. Default 666.
      :return: if include_meta is False, returns a Dask graph that computes the links DataFrame.
               If include_meta is True, returns a tuple: the links DataFrame and the meta DataFrame.
      """
  
      assert expression_matrix.shape[1] == len(gene_names)
      assert client, "client is required"
  
      tf_matrix, tf_matrix_gene_names = to_tf_matrix(expression_matrix, gene_names, tf_names)
  
      future_tf_matrix = client.scatter(tf_matrix, broadcast=True)
      # [1] wrap in a list of 1 -> unsure why but Matt. Rocklin does this often...
      [future_tf_matrix_gene_names] = client.scatter([tf_matrix_gene_names], broadcast=True)
  
      delayed_link_dfs = []  # collection of delayed link DataFrames
      delayed_meta_dfs = []  # collection of delayed meta DataFrame
  
      for target_gene_index in target_gene_indices(gene_names, target_genes):
          target_gene_name = delayed(gene_names[target_gene_index], pure=True)
          target_gene_expression = delayed(expression_matrix[:, target_gene_index], pure=True)
  
          if include_meta:
              delayed_link_df, delayed_meta_df = delayed(infer_partial_network, pure=True, nout=2)(
                  regressor_type, regressor_kwargs,
                  future_tf_matrix, future_tf_matrix_gene_names,
                  target_gene_name, target_gene_expression, include_meta, early_stop_window_length, seed)
  
              if delayed_link_df is not None:
                  delayed_link_dfs.append(delayed_link_df)
                  delayed_meta_dfs.append(delayed_meta_df)
          else:
              delayed_link_df = delayed(infer_partial_network, pure=True)(
                  regressor_type, regressor_kwargs,
                  future_tf_matrix, future_tf_matrix_gene_names,
                  target_gene_name, target_gene_expression, include_meta, early_stop_window_length, seed)
  
              if delayed_link_df is not None:
                  delayed_link_dfs.append(delayed_link_df)
  
      # gather the DataFrames into one distributed DataFrame
      all_links_df = from_delayed(delayed_link_dfs, meta=_GRN_SCHEMA)
      all_meta_df = from_delayed(delayed_meta_dfs, meta=_META_SCHEMA)
  
      # optionally limit the number of resulting regulatory links, descending by top importance
      if limit:
          maybe_limited_links_df = all_links_df.nlargest(limit, columns=['importance'])
      else:
          maybe_limited_links_df = all_links_df
  
      # [2] repartition to nr of workers -> important to avoid GC problems!
      # see: http://dask.pydata.org/en/latest/dataframe-performance.html#repartition-to-reduce-overhead
      n_parts = len(client.ncores()) * repartition_multiplier
  
      if include_meta:
          return maybe_limited_links_df.repartition(npartitions=n_parts), \
                 all_meta_df.repartition(npartitions=n_parts)
      else:
          return maybe_limited_links_df.repartition(npartitions=n_parts)