Large CSV File Reference Issue with VertexAiCodeExecutor using web adk

[Mattylam]

Hi all,

This is a follow-up question from this discussion

I was able to recreate the solution where we pass the csv_artifact as csv content, but that usecase was for using a runner.

My setup:

I have a root agent that talks to the user and transfers the tasks to appropriate agents
One of the sub agent is an analytics agent (similar to this example).
What I am trying to achieve:

I want the sub agent to load appropriate csv files (using the tools I have) and pass them to the VertexAiCodeExecutor to do analysis/plotting on it.

From my research, the best way is to save the artifact, and then load the artifact when needed.

This was my attempt, but failed, the system was able to load the saved csv artifact, the code was able to show, but the system was not able to reference the data.

I have also logged the multiagent communication here. Agent_Communication_report.pdf

`import os
from google.adk.code_executors import VertexAiCodeExecutor
from google.adk.agents import Agent
from google.genai import types
from .prompts import return_instructions_ds
from google.adk.tools import load_artifacts
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmRequest, LlmResponse

add the logger

import logging
logger = logging.getLogger(name)

async def attach_csvs_for_code_exec(
callback_context: CallbackContext, llm_request: LlmRequest
):
"""Attach CSV files to the request for code execution."""
try:
part = await callback_context.load_artifact(filename="table_1.csv")
logger.info(f" CSV ATTACHED to DS Agent")

    # Ensure correct MIME and display name
    blob = part.inline_data
    if not getattr(blob, "mime_type", None):
        blob.mime_type = "text/csv"
    if not getattr(blob, "display_name", None):
        blob.display_name = "table_1.csv"
    
    # Create CSV parts
    csv_parts = [part]
    instruction_part = types.Part(text="Attached data files can be accessed using pandas: pd.read_csv('table_1.csv')")
    
    # Create new content with CSV included
    new_content = types.Content(
        role="user", 
        parts=[instruction_part] + csv_parts
    )
    
    # Modify the request in-place (this is the correct approach)
    llm_request.contents.insert(0, new_content)
    
    # Return None to let the framework proceed with the modified request
    return None
    
except Exception as e:
    logger.error(f"Failed to attach CSV: {e}")
    return None

async def log_code_execution(
callback_context: CallbackContext, llm_response: LlmResponse
):
"""Log generated code and execution results after model response."""
code_detected = False

if hasattr(llm_response, 'content') and llm_response.content:
    if hasattr(llm_response.content, 'parts'):
        for part in llm_response.content.parts:
            if hasattr(part, 'executable_code') and part.executable_code:
                code_detected = True
                logger.info(f"\n🐍 CODE EXECUTED:")
                logger.info(f"```{part.executable_code.language}\n{part.executable_code.code}\n```")
            elif hasattr(part, 'text') and part.text and '```python' in part.text:
                code_detected = True
                logger.info(f"\n🐍 PYTHON CODE DETECTED:")
                logger.info(f"{part.text}")

# Check for code execution results in session state
execution_results_found = False
if hasattr(callback_context, '_invocation_context') and callback_context._invocation_context:
    if hasattr(callback_context._invocation_context, 'session_state'):
        session_state = callback_context._invocation_context.session_state
        if 'code_execution_results' in session_state:
            execution_results_found = True
            logger.info(f"\n⚡ CODE EXECUTION RESULTS:")
            logger.info(f"{session_state['code_execution_results']}")

# Log when code is detected but not executed
if code_detected and not execution_results_found:
    logger.info(f"\n❌ CODE GENERATED BUT NOT EXECUTED:")
    logger.info(f"The agent generated code but it was not executed by the code executor.")
    logger.info(f"This could indicate an issue with the VertexAiCodeExecutor configuration.")

async def log_tool_execution(
tool, args: dict, tool_context, tool_response: dict
):
"""Log tool execution results, including code execution."""
if hasattr(tool, 'name') and 'code' in tool.name.lower():
logger.info(f"\n⚡ CODE EXECUTION COMPLETED:")
logger.info(f"Tool: {tool.name}")
if tool_response:
logger.info(f"Result: {tool_response}")
return tool_response

local_data_science_agent = Agent(
model=os.getenv("ANALYTICS_AGENT_MODEL"),
name="data_science_agent",
instruction=return_instructions_ds(),
code_executor=VertexAiCodeExecutor(stateful=True, optimize_data_file=True),
before_model_callback=attach_csvs_for_code_exec,
after_model_callback=log_code_execution,
#after_tool_callback=log_tool_execution,
#tools=[load_artifacts],
)
`

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: The VertexAiCodeExecutor with optimize_data_file=True can automatically handle CSV files passed in the initial request. You likely do not need the before_model_callback to manually attach the CSV. Instead, include the CSV file as a types.Part when you first call the agent.

---

Hi @Mattylam,

Thanks for the detailed question. It seems like you are running into an issue with how the CSV data is being passed to the VertexAiCodeExecutor. Here's some information that might help.

The VertexAiCodeExecutor has a feature called optimize_data_file which is designed to simplify working with data files like CSVs. [1] When you set optimize_data_file=True, the executor automatically detects and processes data files from the model request. This means you don't need to manually load and attach the file in a callback.

The issue with your current approach is that the before_model_callback might be interfering with the VertexAiCodeExecutor's built-in file handling.

Here is a suggested approach:

1. Remove the before_model_callback: You can remove the before_model_callback=attach_csvs_for_code_exec from your agent definition.

2. Pass the CSV in the initial agent call: When your root agent delegates a task to the data_science_agent, it should include the CSV file content directly in the parts of the request. The VertexAiCodeExecutor will then handle it automatically.

For example, when you invoke your agent, the content should look something like this:

from google.genai import types

# Assuming you have the CSV content in a variable called `csv_content`
csv_part = types.Part(
    inline_data=types.Blob(
        mime_type="text/csv",
        data=csv_content
    )
)

# The user's prompt or instruction for the agent
prompt_part = types.Part(text="Analyze the attached CSV data and create a plot.")

# Make the call to the agent
response = data_science_agent.run(content=[prompt_part, csv_part])

With this approach, the VertexAiCodeExecutor should be able to place the CSV file in the execution environment, and the LLM can then generate Python code to access it using pandas.read_csv(...). The filename inside the execution environment will be a unique name generated by the executor, so your prompt should be general enough to instruct the model to use the provided file.

You can find more details about how VertexAiCodeExecutor is implemented in the source code. [1]

Let me know if this helps!

---

[1] https://github.com/google/adk-python/blob/main/src/google/adk/code_executors/vertex_ai_code_executor.py
[2] https://google.github.io/adk-docs/context/
[3] https://github.com/google/adk-python/blob/main/src/google/adk/flows/llm_flows/_code_execution.py
[4] https://github.com/google/adk-python/blob/main/src/google/adk/code_executors/base_code_executor.py

[Mattylam]

You have not used the web adk way like I sayed, rather you are using the runner like the disscussion I have already attatched