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,
    )
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

    
src/flash/core/serve/execution.py