async def _run_pre_processor(
    invocation_context: InvocationContext,
    llm_request: LlmRequest,
) -> AsyncGenerator[Event, None]:
  """Pre-process the user message by adding the user message to the Colab notebook."""
  from ...agents.llm_agent import LlmAgent

  if not isinstance(invocation_context.agent, LlmAgent):
    return

  agent = invocation_context.agent
  code_executor = agent.code_executor

  if not code_executor or not isinstance(code_executor, BaseCodeExecutor):
    return

  if isinstance(code_executor, BuiltInCodeExecutor):
    code_executor.process_llm_request(llm_request)
    return

  if not code_executor.optimize_data_file:
    return

  code_executor_context = CodeExecutorContext(invocation_context.session.state)

  # Skip if the error count exceeds the max retry attempts.
  if (
      code_executor_context.get_error_count(invocation_context.invocation_id)
      >= code_executor.error_retry_attempts
  ):
    return

  # [Step 1] Extract data files from the session_history and store them in
  # memory. Meanwhile, mutate the inline data file to text part in session
  # history from all turns.
  all_input_files = _extrac_and_replace_inline_files(
      code_executor_context, llm_request
  )

  # [Step 2] Run Explore_Df code on the data files from the current turn. We
  # only need to explore the new data files because the previous data files
  # should already be explored and cached in the code execution runtime.
  processed_file_names = set(code_executor_context.get_processed_file_names())
  files_to_process = [
      f for f in all_input_files if f.name not in processed_file_names
  ]
  for file in files_to_process:
    code_str = _get_data_file_preprocessing_code(file)
    # Skip for unsupported file or executor types.
    if not code_str:
      return

    # Emit the code to execute, and add it to the LLM request.
    code_content = types.Content(
        role='model',
        parts=[
            types.Part(text=f'Processing input file: `{file.name}`'),
            CodeExecutionUtils.build_executable_code_part(code_str),
        ],
    )
    llm_request.contents.append(copy.deepcopy(code_content))
    yield Event(
        invocation_id=invocation_context.invocation_id,
        author=agent.name,
        branch=invocation_context.branch,
        content=code_content,
    )

    code_execution_result = code_executor.execute_code(
        invocation_context,
        CodeExecutionInput(
            code=code_str,
            input_files=[file],
            execution_id=_get_or_set_execution_id(
                invocation_context, code_executor_context
            ),
        ),
    )
    # Update the processing results to code executor context.
    code_executor_context.update_code_execution_result(
        invocation_context.invocation_id,
        code_str,
        code_execution_result.stdout,
        code_execution_result.stderr,
    )
    code_executor_context.add_processed_file_names([file.name])

    # Emit the execution result, and add it to the LLM request.
    execution_result_event = await _post_process_code_execution_result(
        invocation_context, code_executor_context, code_execution_result
    )
    yield execution_result_event
    llm_request.contents.append(copy.deepcopy(execution_result_event.content))
Pre-process the user message by adding the user message to the Colab notebook.
src/google/adk/flows/llm_flows/_code_execution.py