Multi-agent system problems

[emilsFlashpoint]

We have built a Multi-agent system. It works as it should, but we have realized that there are some problems.

This is our root agent:

root_agent = Agent(
    name="pitch_deck_analyser",
    model="gemini-2.5-flash",
    before_agent_callback=save_file_as_artifact,
    instruction="""
        You are a non-conversational, programmatic router. You do not answer questions, provide explanations, or engage in conversation. Your sole purpose is to call the correct tool based on the rules below.

        You have access to the following sub-agents:
        - `pitch_deck_pipeline_agent`: Use this to start the full analysis pipeline ONLY AFTER a file has been uploaded.

        You have these available tools:
        - `general_query_agent`: Use this ONLY for follow-up questions AFTER the main analysis is complete.
        - `general_intro_agent`: Use this for greetings, introductions, or if you are unsure which other tool to use.

        You must use the following state to make your decision:
        - `file_artifact_saved`: {file_artifact_saved} (True if a PDF has been uploaded)
        - `vc_review`: {vc_review} (Not empty if the final analysis is done)

        --- ROUTING RULES ---
        Evaluate these rules in order. The first rule that matches is the only one you execute.

        **RULE 1: Q&A MODE (Analysis is Complete)**
        - **IF** the `vc_review` state is NOT empty:
        - **THEN** any user input is a follow-up question. Your ONLY action is to call the `general_query_agent` tool. Stop.

        **RULE 2: INITIAL PIPELINE MODE (File is Uploaded, Ready for Analysis)**
        - **ELSE IF** the `file_artifact_saved` state is `True` AND the `vc_review` state is EMPTY:
        - **THEN** the user has uploaded a file and the pipeline must be started. Your ONLY action is to call the `pitch_deck_pipeline_agent` tool. Stop.
         - **NOTE**: This rule applies to both initial file uploads AND fresh file replacements (callback resets vc_review to empty for new files).

        **RULE 3: GENERAL INTRO / DEFAULT MODE (No File or Confused)**
        - **ELSE** (this means `file_artifact_saved` is False OR you are unsure):
        - **THEN** the user is asking a general question or greeting you. Your ONLY action is to call the `general_intro_agent` tool. Stop.

        **CRITICAL RULE: Under NO circumstances should you ever answer the user directly. Your only output MUST be a tool call. If the user's request is unclear or does not fit the rules above, your default action is to call the `general_intro_agent`.**
    """,
    sub_agents=[pitch_deck_pipeline_agent],
    tools=[AgentTool(agent=general_query_agent, skip_summarization=True),
           AgentTool(agent=general_intro_agent, skip_summarization=True)]
)

This works almost perfectly - checks the state and, depending on values, selects the necessary sub-agent/tool.
1. The problem is that context is not passed to AgenTool, so agents do not know about previous conversations.

AgentTool as an Encapsulated Unit: The purpose of AgentTool is to wrap a complete agent and expose it as a self-contained, stateless tool. Much like calling an external API, the parent agent provides inputs to the AgentTool, which then initiates a new, separate run of the child agent to get a result. This design ensures that the child agent's execution is fully encapsulated and independent, which is why it generates its own session and invocation context.

Sub-agents for Shared Context: In contrast, the sub_agents paradigm (used with SequentialAgent, ParallelAgent, etc.) is designed for orchestrating a series of agents that work together on a single, continuous task. In this pattern, the agents are intended to operate within the same context, so the framework passes the same InvocationContext, including the session state and ID, from the parent to each sub-agent. [1]

2. Then I tried to move all agents as sub-agents:
sub_agents=[pitch_deck_pipeline_agent, general_query_agent, general_intro_agent],

But then the delegation does weird or expected runs. For example, if the intro agent responds and then the user uploads a file, the previous run has stayed with intro, and the intro agent sometimes transfers to the pipeline agent, sometimes not.
I understand that it is from the https://google.github.io/adk-docs/tools-custom/function-tools/#key-difference-from-sub-agents

Sub-agent: When Agent A calls Agent B as a sub-agent, the responsibility of answering the user is completely transferred to Agent B. Agent A is effectively out of the loop. All subsequent user input will be answered by Agent B.

3. general_query_agent has google_search tool available. Works from AgentTool, but does not work for sub-agent. So if we want context, we can't use google_search in sub-agent. If we want google_search, sub agent/AgentTool does not have the conversation context.

I have built workarounds for all other functionality that was necessary, for example, loading the artifact in the query_agent before running prompt.

Maybe someone could help/guide me in the direction that I could try out?!
I have tried A LOT of solutions - simpler/more complex prompting, disallow_transfer_to_parent.

The perfect scenario: every input goes through the root agent and its delegating the input to the necessary sub-agent/tool.

[miyannishar]

The fundamental issue is that ADK's architecture doesn't support invoking agents from function tools while preserving conversation context. Looking at AgentTool's run_async method, it creates a new Runner with InMemorySessionService() and a new session—while it copies state dictionary values (state_dict = {k: v for k, v in tool_context.state.to_dict().items()}), it does NOT copy conversation history (the actual messages/events). AgentTool is supposed to get input from the base agent, run the tool as a sub agent and pass the context to the base agent. However, I see you have done skip_summarization=True that will NOT pass the result of the agent tool to the base agent or the agent its called by. If the events are handled properly, it will be shown as it is in the FE.

The child agent runs in complete isolation with only the state dict, not the full conversation context. When you try to convert agents to function tools, you can't directly call agent.model.generate_content_async() because that bypasses the agent's instructions, tools, and state management. If you use runner.run_async() from a function tool instead, you face the same issue; new session, no conversation history. AgentTool copies state but not conversation history, sub-agents share context but transfer control completely, and function tools can't properly invoke agents through ADK's framework.

Additionally, if you initialize the artifact service in the Runner (as done in adk_agent.py), you don't need to manually save artifacts in a before_agent_callback; the framework handles artifact persistence automatically when agents use tools that save artifacts via tool_context.save_artifact().

What is the solution you are going to implement using this approach? Maybe I can help finding the workaround. You want agentTool to have the conversation history as well?

[emilsFlashpoint]

Thank you, @miyannishar , your response makes perfect sense 🥇

My agent's idea was simple, but as I mentioned above, there are issues.
This current setup works perfectly, routing-wise and by tool usage, the only problem is conversation context passing (and, from your suggestion, simpler artifact handling)

The architecture consists of a root_agent that receives input and routes it to three specialized sub-agents (implemented as Agents-as-a-Tool):

1. general_intro_agent: Handles onboarding and answers general questions about the agent's capabilities and usage.

2. pitch_deck_pipeline_agent: A sequential agent responsible for the analysis pipeline. It handles file processing (PDF analysis) and data enrichment via APIs.

3. general_query_agent: Becomes active after analysis is complete. It serves as the primary interface for answering user questions based on the processed data.

A key requirement is that both the general_intro_agent and general_query_agent need access to the conversation context. For example, the general_query_agent utilizes a Google Search tool and requires historical context to answer follow-up questions effectively.

However, I encountered an issue where context is not preserved when using AgentTool; answers to previous questions become inaccessible in future turns. I attempted to disable summarization to fix this, but it did not resolve the issue.

Regarding artifacts, despite attempting multiple solutions, generated artifacts do not appear in the "Artifact" section of the ADK web interface. Currently, I rely on the root_agent's before_agent_callback to manage state—for instance, clearing all state values if a new file is uploaded during an active session.

EDIT: after updating ADK, the artifacts now show up in the "adk web" Artifacts interface. Will do some digging on how I can simplify my process

For the runner configuration, I use VertexAiSessionService for session management and GcsArtifactService for file storage:

# Create an INSTANCE of the artifact service
artifact_service = InMemoryArtifactService()
session_service = InMemorySessionService()
vertex_service = VertexAiSessionService(os.getenv("GOOGLE_CLOUD_PROJECT"), os.getenv("GOOGLE_CLOUD_LOCATION"))

gcs_artifact_service = GcsArtifactService(
    bucket_name = os.getenv("GCS_BUCKET_NAME")
)

print("Configuring the Runner with the agent and services...")


runner = Runner(
    agent=root_agent,
    app_name="fp-pitch-deck-analyser",
    session_service=vertex_service,
    artifact_service=gcs_artifact_service  
)

print("Runner is configured and ready.")

To ensure the general_query_agent has the necessary context, the only effective method I found was to manually inject the file data and state before execution. This is handled via a custom callback (for testing, I use inline_data, in production - file_data GCS URI ):

import google.genai.types as types
from google.adk.agents.callback_context import CallbackContext
from google.adk.agents import Agent
from google.adk.tools import google_search

async def inject_context_for_query_agent(callback_context: CallbackContext, **kwargs):
    """
    This callback runs before the general_query_agent.
    It loads the PDF artifact and state data and APPENDS them to the agent's input.
    It returns None to allow the agent to proceed with the now-enriched input.
    """
    print("CALLBACK (Query Agent): Intercepting call. Injecting full context...")
    
    # --- Step 1: Load PDF artifact ---
    artifact_filename = callback_context.state.get("artifact_filename")
    file_type = callback_context.state.get("file_type", "unknown")
    
    if not artifact_filename:
        print("CALLBACK ERROR: No artifact filename in state.")
        return 
        
    loaded_artifact = await callback_context.load_artifact(filename=artifact_filename)
    if not loaded_artifact:
        print("CALLBACK ERROR: Could not load the artifact.")
        return

    print(f"CALLBACK: Loaded artifact of type: {file_type}")
    print(f"CALLBACK: Loaded artifact details: {loaded_artifact}")
    
    # Handle different file types based on how they were saved
    if file_type == "inline_data":
        # For inline_data, the artifact should contain the original file part with binary data
        print("CALLBACK: Processing inline_data artifact - using directly")
        file_part = loaded_artifact
        
    elif file_type == "file_data":
        # For file_data, the artifact contains the URI as text or bytes
        print("CALLBACK: Processing file_data artifact - reading from GCS URI")

        # Check if URI is stored as text or as inline_data bytes
        file_uri = None
        if loaded_artifact.text:
            file_uri = loaded_artifact.text
        elif loaded_artifact.inline_data and loaded_artifact.inline_data.data:
            # URI is stored as bytes in inline_data
            file_uri = loaded_artifact.inline_data.data.decode('utf-8')
        
        if not file_uri:
            print("CALLBACK ERROR: Could not extract file URI from artifact")
            return
            
        print(f"CALLBACK: Found GCS URI: {file_uri}")
        
        try:
            # Read the actual file content from GCS
            if file_uri.startswith("gs://"):
                # Use GCS client library to read the file
                from google.cloud import storage
                
                # Parse the GCS URI to get bucket and blob name
                # Format: gs://bucket-name/path/to/file
                uri_parts = file_uri.replace("gs://", "").split("/", 1)
                if len(uri_parts) != 2:
                    print(f"CALLBACK ERROR: Invalid GCS URI format: {file_uri}")
                    return
                    
                bucket_name, blob_name = uri_parts
                print(f"CALLBACK: Reading from bucket '{bucket_name}', blob '{blob_name}'")
                
                # Download file content
                client = storage.Client()
                bucket = client.bucket(bucket_name)
                blob = bucket.blob(blob_name)
                
                # Get file content as bytes
                file_content = blob.download_as_bytes()
                mime_type = callback_context.state.get("file_mime_type", "application/pdf")
                
                # Create a Part with inline_data containing the actual file content
                file_part = types.Part(
                    inline_data=types.Blob(
                        data=file_content,
                        mime_type=mime_type
                    )
                )
                print("CALLBACK: Successfully read file from GCS and created inline_data part")
                
            else:
                print(f"CALLBACK ERROR: Unsupported URI scheme: {file_uri}")
                return
                
        except Exception as e:
            print(f"CALLBACK ERROR: Failed to read file from GCS: {e}")
            return
            
    else:
        print(f"CALLBACK: Unknown file type '{file_type}', using artifact as-is")
        file_part = loaded_artifact
    
    # IMPORTANT: Append the file part to the existing input
    callback_context.user_content.parts.append(file_part)
    print("CALLBACK: File part injected into agent input.")

    return None



general_query_agent = Agent(
    name="general_query_agent",
    model="gemini-2.5-flash",  # Use a powerful multi-modal model
    
    # Attach the callback that will inject all the necessary data
    before_agent_callback=inject_context_for_query_agent,
    
    tools=[google_search], # This agent's job is to reason over data, not call other tools
       instruction="""
            You are a resourceful, world-class business analyst. Your goal is to provide accurate, comprehensive answers by intelligently using your available resources.

            **CRITICAL CONSTRAINT: Your entire response MUST be under 3500 characters (including spaces). Be concise. Track your character count.**

            ## Available Context
            * Pitch deck PDF file & `{vc_review}` analysis.
            * Prior conversation history.
            * Data package: `{final_enriched_data}` (structured Coresignal LinkedIn data)
                * `"founders"`: Array of founder profiles (employment, education, skills, etc.).
                * `"linkedin_company_data"`: Company profile (description, employees, HQ, funding, industries, etc.).

            ## Response Instructions

            ### 1. Data Source Priority
            * **LinkedIn/Founder/Company Data:** When asked about profiles, backgrounds, company info, employees, funding, or any data potentially from LinkedIn, you **MUST** search `{final_enriched_data}` *first* and *completely* before using Google Search.
            * **Market/Competition/Product Data:** For queries on **market, competition, product offering, current solutions (how the issue is tackled now), valuations, pricing, etc.**, first check the pitch deck and `{vc_review}`. Then, **actively use your reasoning and integrated Google Search to complement, validate, and provide a more comprehensive answer.**
            * **General Google Search Rule:** For all other topics, use Google Search *if* the answer is not found in your provided context (Deck, `{vc_review}`, `{final_enriched_data}`) OR the information available in the Available Context is not sufficient to give a comprehensive response OR if the user *explicitly* asks for external info (e.g., "latest news," "recent posts").

            ### 2. Formatting & Content
            * Formulate a single, comprehensive response.
            * Use standard business abbreviations (e.g., YoY, ARR, NRR, HQ) to save space.
            * **Source Citation:** Briefly indicate your source (e.g., "Per deck...", "Per LinkedIn data..."). Keep citations minimal.
            * **Google Search Citation:** **IMPORTANT:** If you use Google Search, you **MUST** cite the source URL(s).

            ### 3. Specific Query Types
            * **"How do you know" / "What's your source?":** Explicitly state the source:
                * Pitch deck PDF (specify slide/section if possible).
                * `{vc_review}` analysis.
                * `{final_enriched_data}` (specify `founders` or `linkedin_company_data`).
                * Google Search (with URL).
            * **"Breakdown Clare score":** Provide a concise, bulleted breakdown by the four pillars (Opportunity, Team, PMF, Moat), giving reasoning for each pillar's score and its contribution to the total.

            **FINAL REMINDER: Max 3500 characters. Prioritize critical info if approaching the limit.**
   """,
    description="The final agent that answers questions using the file and enriched data."
)

[GWeale]

Hi!

The issue you are raising is core to how ADK uses AgentTool (stateless) vs sub_agents (shared context, control transfer).

Since AgentTool copies state (but not conversation history), you could serialize the context you need into state before the tool runs or use a sub_agent instead

[sarojrout]

Your analysis is 100% correct. You've identified a architectural limitation in ADK's current multi-agent patterns.
Currently in AgentTool Doesn't Share Conversation Context
AgentTool creates a new session (line 155-170 in agent_tool.py): https://github.com/google/adk-python/blob/main/src/google/adk/tools/agent_tool.py#L155-L170

• Uses InMemorySessionService() - creates isolated session
• Only passes state_dict (state variables), not conversation history
• Child agent sees no previous messages/events

Another problem is Sub-agents Transfer Control:
When using sub_agents, control is completely transferred:

• From docs: "Agent A is effectively out of the loop"
• Root agent can't route subsequent inputs
• This is by design for SequentialAgent/ParallelAgent patterns

Third problem is: google_search Limitation
google_search has a known workaround (see google_search_agent_tool.py line 54): https://github.com/google/adk-python/blob/main/src/google/adk/tools/google_search_agent_tool.py#L54

• Works with AgentTool (isolated session)
• Doesn't work with sub-agents (shared context)

The workaround is:
Manual Context Injection:

Create a custom AgentTool wrapper that manually injects conversation history:

from google.adk.tools import AgentTool
from google.adk.tools.tool_context import ToolContext
from typing import Any
from google.genai import types

class ContextAwareAgentTool(AgentTool):
    """AgentTool that manually injects conversation context."""
    
    def __init__(self, agent, context_window: int = 10, skip_summarization: bool = False):
        super().__init__(agent=agent, skip_summarization=skip_summarization)
        self.context_window = context_window
    
    async def run_async(self, *, args: dict[str, Any], tool_context: ToolContext) -> Any:
        # Get conversation history from parent context
        invocation_context = tool_context._invocation_context
        events = invocation_context._get_events(current_branch=True)
        
        # Extract recent conversation (last N events)
        recent_messages = []
        for event in events[-self.context_window:]:
            if event.content and event.content.parts:
                for part in event.content.parts:
                    if part.text:
                        author = event.author or "system"
                        recent_messages.append(f"{author}: {part.text}")
        
        # Build context summary (last 5-6 messages for token efficiency)
        context_summary = "\n".join(recent_messages[-6:]) if recent_messages else "No previous conversation."
        
        # Inject context into the request
        original_request = args.get('request', '')
        args['request'] = f"""Previous conversation context:
{context_summary}

Current user request: {original_request}
"""
        
        # Call parent implementation with modified args
        return await super().run_async(args=args, tool_context=tool_context)

And then invoke as

root_agent = Agent(
    name="pitch_deck_analyser",
    tools=[
        ContextAwareAgentTool(agent=general_query_agent, skip_summarization=True),
        ContextAwareAgentTool(agent=general_intro_agent, skip_summarization=True),
    ],
    sub_agents=[pitch_deck_pipeline_agent],  # Keep pipeline as sub-agent for context
)

For a longterm solution, we need a new pattern to be established in ADK