def execute(
    rt_config: QuantumRuntimeConfiguration,
    executable_group: QuantumExecutableGroup,
    base_data_dir: str = ".",
) -> ExecutableGroupResult:
    """Execute a `cg.QuantumExecutableGroup` according to a `cg.QuantumRuntimeConfiguration`.

    The ExecutableGroupResult's constituent parts will be saved to disk as they become
    available. Within the "{base_data_dir}/{run_id}" directory we save:
        - The `cg.QuantumRuntimeConfiguration` at the start of the execution as a record
          of *how* the executable group was run.
        - A `cg.SharedRuntimeInfo` which is updated throughout the run.
        - An `cg.ExecutableResult` for each `cg.QuantumExecutable` as they become available.
        - A `cg.ExecutableGroupResultFilesystemRecord` which is updated throughout the run.

    Args:
        rt_config: The `cg.QuantumRuntimeConfiguration` specifying how to execute
            `executable_group`.
        executable_group: The `cg.QuantumExecutableGroup` containing the executables to execute.
        base_data_dir: Each data file will be written to the "{base_data_dir}/{run_id}/" directory,
            which must not already exist.

    Returns:
        The `cg.ExecutableGroupResult` containing all data and metadata for an execution.

    Raises:
        NotImplementedError: If an executable uses the `params` field or anything other than
            a BitstringsMeasurement measurement field.
        ValueError: If `base_data_dir` is not a valid directory.
    """
    # run_id defaults logic.
    if rt_config.run_id is None:
        run_id = str(uuid.uuid4())
    else:
        run_id = rt_config.run_id

    # base_data_dir handling.
    if not base_data_dir:  # pragma: no cover
        raise ValueError("Please provide a non-empty `base_data_dir`.")

    sampler = rt_config.processor_record.get_sampler()
    device = rt_config.processor_record.get_device()

    shared_rt_info = SharedRuntimeInfo(
        run_id=run_id, device=device, run_start_time=datetime.datetime.now(tz=datetime.timezone.utc)
    )
    executable_results = []

    saver = _FilesystemSaver(base_data_dir=base_data_dir, run_id=run_id)
    saver.initialize(rt_config, shared_rt_info)

    logger = _PrintLogger(n_total=len(executable_group))
    logger.initialize()

    rs = np.random.RandomState(rt_config.random_seed)
    exe: QuantumExecutable
    for i, exe in enumerate(executable_group):
        runtime_info = RuntimeInfo(execution_index=i)

        if exe.params != tuple():
            raise NotImplementedError("Circuit params are not yet supported.")
        if not hasattr(exe.measurement, 'n_repetitions'):
            raise NotImplementedError("Only `BitstringsMeasurement` are supported.")

        circuit = exe.circuit
        if exe.problem_topology is not None:
            with _time_into_runtime_info(runtime_info, 'placement'):
                circuit, mapping = rt_config.qubit_placer.place_circuit(
                    circuit,
                    problem_topology=exe.problem_topology,
                    shared_rt_info=shared_rt_info,
                    rs=rs,
                )
                runtime_info.qubit_placement = mapping

        if rt_config.target_gateset is not None:
            circuit = cirq.optimize_for_target_gateset(
                circuit, gateset=rt_config.target_gateset
            ).freeze()

        with _time_into_runtime_info(runtime_info, 'run'):
            sampler_run_result = sampler.run(circuit, repetitions=exe.measurement.n_repetitions)

        exe_result = ExecutableResult(
            spec=exe.spec, runtime_info=runtime_info, raw_data=sampler_run_result
        )
        # Do bookkeeping for finished ExecutableResult
        executable_results.append(exe_result)
        saver.consume_result(exe_result, shared_rt_info)
        logger.consume_result(exe_result, shared_rt_info)

    shared_rt_info.run_end_time = datetime.datetime.now(tz=datetime.timezone.utc)
    saver.finalize(shared_rt_info=shared_rt_info)
    logger.finalize()

    return ExecutableGroupResult(
        runtime_configuration=rt_config,
        shared_runtime_info=shared_rt_info,
        executable_results=executable_results,
    )
Execute a `cg.QuantumExecutableGroup` according to a `cg.QuantumRuntimeConfiguration`.

    The ExecutableGroupResult's constituent parts will be saved to disk as they become
    available. Within the "{base_data_dir}/{run_id}" directory we save:
        - The `cg.QuantumRuntimeConfiguration` at the start of the execution as a record
          of *how* the executable group was run.
        - A `cg.SharedRuntimeInfo` which is updated throughout the run.
        - An `cg.ExecutableResult` for each `cg.QuantumExecutable` as they become available.
        - A `cg.ExecutableGroupResultFilesystemRecord` which is updated throughout the run.

    Args:
        rt_config: The `cg.QuantumRuntimeConfiguration` specifying how to execute
            `executable_group`.
        executable_group: The `cg.QuantumExecutableGroup` containing the executables to execute.
        base_data_dir: Each data file will be written to the "{base_data_dir}/{run_id}/" directory,
            which must not already exist.

    Returns:
        The `cg.ExecutableGroupResult` containing all data and metadata for an execution.

    Raises:
        NotImplementedError: If an executable uses the `params` field or anything other than
            a BitstringsMeasurement measurement field.
        ValueError: If `base_data_dir` is not a valid directory.
    
cirq-google/cirq_google/workflow/quantum_runtime.py