How to handle user uploaded file in multi agent scenario

[subbu4github]

I am building a multi agent app to analyse the sql script.

I am building a Sequential Agent which has the Fileacceptor agent, followed by the parallel agent.

The parallel agent contains sub agents to process the input file and generate various outputs.

I want to stop the execution of subsequent agents when the file is not uploaded. I want to use the Artifact approach as I have to share the same file multiple times. If I create the artifact, whether the file will be removed from the context window or I have to remove it manually.

I am new to ADK. Guidance required.

[adk-bot]

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

TLDR: To stop execution, you can have an agent set ctx.end_invocation = True. Artifacts are the correct way to handle files to avoid filling up the context window, and you do not need to manually remove them.

---

Hello! Welcome to ADK. You've asked some great questions about handling multi-agent workflows. Here’s a more detailed breakdown based on your scenario:

1. How to Stop Execution if a File Isn't Uploaded

You can control the flow of execution by having an agent decide to terminate the process early. The recommended way to do this is to have an agent set ctx.end_invocation = True. This signals to the ADK framework to stop processing any further agents in the sequence.

In your case, you could have an agent that runs after your Fileacceptor agent. This new agent's sole job is to check if the file was successfully uploaded. The Fileacceptor can save the status to the context state, and the "validator" agent can read it.

Here's a conceptual example:

# In your Fileacceptor or a subsequent agent, you can set a status
if not file_uploaded:
    context.state["temp:file_upload_status"] = "failed"
else:
    context.state["temp:file_upload_status"] = "success"
    context.save_artifact("my_sql_script.sql", file_content)

Then, you can create a simple agent to act as a gate:

from adk.api import Agent, Context

class StopExecutionAgent(Agent):
    async def _run_async_impl(self, context: Context):
        if context.state.get("temp:file_upload_status") == "failed":
            print("File upload failed. Stopping execution.")
            context.end_invocation = True

By placing this StopExecutionAgent in your sequence, the workflow will halt if the condition is met. [1]

2. Using Artifacts to Share Files

Your instinct to use artifacts is spot on! This is the designed best practice in ADK for passing files or large data between agents. [2]

• Your Fileacceptor agent should use context.save_artifact() to store the uploaded file.
• Your subsequent sub-agents in the parallel agent can then use context.load_artifact() to access the file's content when they need to process it.

Remember to configure an ArtifactService (like InMemoryArtifactService or GcsArtifactService) when you initialize your Runner.

3. Artifacts and the Context Window

This is a key benefit of using artifacts. When you call context.save_artifact(), the file content is not stored directly in the prompt or the context window. Instead, it is managed by the ArtifactService (e.g., saved in memory or to a cloud bucket), and a reference is stored in the session state. [1]

ADK is designed to manage the context window for you by "lazy-loading" artifact data only when an agent explicitly requests it via load_artifact(). [3] This prevents the context from overflowing with large file content.

Therefore, you do not need to remove the file from the context manually. By using the artifact system, you are already following the correct pattern for efficient context management. [4]

I hope this guidance helps you get started with your project!

---

[1] Context in ADK
[2] Handling Artifacts
[3] ADK FAQ
[4] Context Offloading with Artifacts Example