code: def _process_initial(
      endpoint_protocol: "EndpointProtocol",
      components: Dict[str, "ModelComponent"],
  ) -> UnprocessedTaskDask:
      """Extract task dsk and payload / results keys and return computable form.
  
      Parameters
      ----------
      endpoint_protocol
          endpoint protocol definition for the variation of the DAG which
          is currently being evaluated.
      components
          Mapping of component name -> component class definitions which
          contain independent subgraph task dsks'.
  
      Returns
      -------
      UnprocessedTaskDask
  
      """
  
      # mapping payload input keys -> serialized keys / tasks
      payload_dsk_key_map = {
          payload_key: f"{input_key}.serial"
          for payload_key, input_key in endpoint_protocol.dsk_input_key_map.items()
      }
      payload_input_tasks_dsk = {
          input_dsk_key: (identity, payload_key)
          for payload_key, input_dsk_key in payload_dsk_key_map.items()
      }
  
      # mapping result keys -> serialize keys / tasks
      res_dsk_key_map = {
          result_key: f"{output_key}.serial"
          for result_key, output_key in endpoint_protocol.dsk_output_key_map.items()
      }
      result_output_tasks_dsk = {
          result_key: (identity, output_dsk_key)
          for result_key, output_dsk_key in res_dsk_key_map.items()
      }
      output_keys = list(res_dsk_key_map.keys())
  
      # need check to prevent cycle error
      _payload_keys = set(payload_dsk_key_map.keys())
      _result_keys = set(res_dsk_key_map.keys())
      if not _payload_keys.isdisjoint(_result_keys):
          raise KeyError(
              f"Request payload keys `{_payload_keys}` and response keys `{_result_keys}` "
              f"names cannot intersectt. keys: `{_payload_keys.intersection(_result_keys)}` "
              f"must be renamed in either `inputs` or `outputs`. "
          )
  
      component_dsk = merge(
          valmap(attrgetter("_flashserve_meta_.dsk"), components)
      )
      merged_dsk = merge(*(dsk for dsk in component_dsk.values()))
  
      return UnprocessedTaskDask(
          component_dsk=component_dsk,
          merged_dsk=merged_dsk,
          payload_tasks_dsk=payload_input_tasks_dsk,
          payload_dsk_map=payload_dsk_key_map,
          result_tasks_dsk=result_output_tasks_dsk,
          result_dsk_map=res_dsk_key_map,
          output_keys=output_keys,
      )
  
imports: from collections import defaultdict
  from dataclasses import dataclass
  from operator import attrgetter
  from typing import TYPE_CHECKING, Dict, List, Set, Tuple
  from flash.core.serve.dag.optimization import cull, functions_of, inline_functions
  from flash.core.serve.dag.rewrite import RewriteRule, RuleSet
  from flash.core.serve.dag.task import flatten, get_deps, getcycle, isdag, toposort
  from flash.core.serve.dag.utils import funcname
  from flash.core.utilities.imports import _CYTOOLZ_AVAILABLE, _PYDANTIC_AVAILABLE
label_desc: Use image node_id 1 for calling a global function with example usage: _process_initial(endpoint_protocol, components) and returns: UnprocessedTaskDask