Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 138 additions & 22 deletions src/google/adk/flows/llm_flows/agent_transfer.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@

from __future__ import annotations

import asyncio
import logging
import typing
from typing import AsyncGenerator

Expand All @@ -30,8 +32,13 @@
from ._base_llm_processor import BaseLlmRequestProcessor

if typing.TYPE_CHECKING:
from a2a.types import AgentCard

from ...agents import BaseAgent
from ...agents import LlmAgent
from ...agents.remote_a2a_agent import RemoteA2aAgent

logger = logging.getLogger('google_adk.' + __name__)


class _AgentTransferLlmRequestProcessor(BaseLlmRequestProcessor):
Expand All @@ -50,11 +57,11 @@ async def run_async(
if not transfer_targets:
return

llm_request.append_instructions([
_build_target_agents_instructions(
invocation_context.agent, transfer_targets
)
])
# Build instructions asynchronously to support A2A agent card resolution
instructions = await _build_target_agents_instructions(
invocation_context.agent, transfer_targets
)
llm_request.append_instructions([instructions])

transfer_to_agent_tool = FunctionTool(func=transfer_to_agent)
tool_context = ToolContext(invocation_context)
Expand All @@ -69,19 +76,118 @@ async def run_async(
request_processor = _AgentTransferLlmRequestProcessor()


def _build_target_agent_info_from_card(
target_agent: RemoteA2aAgent, agent_card: AgentCard
) -> str:
"""Build rich agent info from A2A Agent Card.

Args:
target_agent: The RemoteA2aAgent instance
agent_card: The resolved A2A Agent Card

Returns:
Formatted string with detailed agent information from the card,
optimized for LLM consumption when selecting subagents.
"""
info_parts = []

# Start with a clear header for the agent
info_parts.append(f'### Agent: {target_agent.name}')

# Include both RemoteA2aAgent description and agent card description
# This provides both the locally-configured context and the remote agent's self-description
descriptions = []
if target_agent.description:
descriptions.append(f'Description: {target_agent.description}')
if agent_card.description and agent_card.description != target_agent.description:
descriptions.append(f'Agent card description: {agent_card.description}')

if descriptions:
info_parts.append('\n'.join(descriptions))

# Add skills in a structured, LLM-friendly format
if agent_card.skills:
info_parts.append('\nSkills:')
for skill in agent_card.skills:
# Format: "- skill_name: description (tags: tag1, tag2)"
skill_parts = [f' - **{skill.name}**']
if skill.description:
skill_parts.append(f': {skill.description}')
if skill.tags:
skill_parts.append(f' [Tags: {", ".join(skill.tags)}]')
info_parts.append(''.join(skill_parts))

return '\n'.join(info_parts)


async def _build_target_agents_info_async(target_agent: BaseAgent) -> str:
"""Build agent info, using A2A Agent Card if available.

Args:
target_agent: The agent to build info for

Returns:
Formatted string with agent information
"""
from ...agents.remote_a2a_agent import RemoteA2aAgent

# Check if this is a RemoteA2aAgent and ensure it's resolved
if isinstance(target_agent, RemoteA2aAgent):
try:
# Ensure the agent card is resolved
await target_agent._ensure_resolved()

# If we have an agent card, use it to build rich info
if target_agent._agent_card:
return _build_target_agent_info_from_card(
target_agent, target_agent._agent_card
)
except Exception as e:
# If resolution fails, fall through to default behavior
logger.warning(
'Failed to resolve A2A agent card for agent "%s", falling back to' ' basic info. Error: %s',
target_agent.name,
e,
)
pass
# Fallback to original behavior for non-A2A agents or if card unavailable
return _build_target_agents_info(target_agent)


def _build_target_agents_info(target_agent: BaseAgent) -> str:
return f"""
Agent name: {target_agent.name}
Agent description: {target_agent.description}
"""
"""Build basic agent info (fallback for non-A2A agents).

Args:
target_agent: The agent to build info for

Returns:
Formatted string with basic agent information, matching the enhanced format
for consistency with A2A agent cards.
"""
info_parts = [f'### Agent: {target_agent.name}']

if target_agent.description:
info_parts.append(f'Description: {target_agent.description}')

return '\n'.join(info_parts)


line_break = '\n'


def _build_target_agents_instructions(
async def _build_target_agents_instructions(
agent: LlmAgent, target_agents: list[BaseAgent]
) -> str:
"""Build instructions for agent transfer with detailed agent information.

Args:
agent: The current agent
target_agents: List of agents that can be transferred to

Returns:
Formatted instructions string with agent transfer information,
optimized for LLM decision-making about which subagent to use.
"""
# Build list of available agent names for the NOTE
# target_agents already includes parent agent if applicable, so no need to add it again
available_agent_names = [target_agent.name for target_agent in target_agents]
Expand All @@ -94,27 +200,37 @@ def _build_target_agents_instructions(
f'`{name}`' for name in available_agent_names
)

# Build agent info asynchronously and concurrently to support A2A agent card resolution
tasks = [
_build_target_agents_info_async(target_agent)
for target_agent in target_agents
]
agent_info_list = await asyncio.gather(*tasks)

# Create a separator for visual clarity
agents_section = '\n\n'.join(agent_info_list)

si = f"""
You have a list of other agents to transfer to:
## Available Agents for Transfer

You can delegate tasks to the following specialized agents. Carefully review each agent's description and skills to determine the best match for the user's request.

{agents_section}

## Decision Criteria

{line_break.join([
_build_target_agents_info(target_agent) for target_agent in target_agents
])}
1. **Assess your own capability**: If you are the best agent to handle this request based on your own description and capabilities, answer it directly.

If you are the best to answer the question according to your description, you
can answer it.
2. **Consider specialized agents**: If another agent has more relevant skills or expertise for this request, call the `{_TRANSFER_TO_AGENT_FUNCTION_NAME}` function to transfer to that agent. Match the user's needs with the agent's skills and descriptions above.

If another agent is better for answering the question according to its
description, call `{_TRANSFER_TO_AGENT_FUNCTION_NAME}` function to transfer the
question to that agent. When transferring, do not generate any text other than
the function call.
3. **When transferring**: Only call the function - do not generate any additional text.

**NOTE**: the only available agents for `{_TRANSFER_TO_AGENT_FUNCTION_NAME}` function are {formatted_agent_names}.
**IMPORTANT**: The only valid agent names for `{_TRANSFER_TO_AGENT_FUNCTION_NAME}` are: {formatted_agent_names}
"""

if agent.parent_agent and not agent.disallow_transfer_to_parent:
si += f"""
If neither you nor the other agents are best for the question, transfer to your parent agent {agent.parent_agent.name}.
4. **Escalate to parent**: If neither you nor the specialized agents are suitable for this request, transfer to your parent agent `{agent.parent_agent.name}` for broader assistance.
"""
return si

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -102,44 +102,41 @@ async def test_agent_transfer_includes_sorted_agent_names_in_system_instructions

# The NOTE should contain agents in alphabetical order: sub-agents + parent + peers
expected_content = """\
## Available Agents for Transfer

You have a list of other agents to transfer to:
You can delegate tasks to the following specialized agents. Carefully review each agent's description and skills to determine the best match for the user's request.

### Agent: z_agent
Description: Last agent

Agent name: z_agent
Agent description: Last agent
### Agent: a_agent
Description: First agent

### Agent: m_agent
Description: Middle agent

Agent name: a_agent
Agent description: First agent
### Agent: parent_agent
Description: Parent agent

### Agent: peer_agent
Description: Peer agent

Agent name: m_agent
Agent description: Middle agent
## Decision Criteria

1. **Assess your own capability**: If you are the best agent to handle this request based on your own description and capabilities, answer it directly.

Agent name: parent_agent
Agent description: Parent agent
2. **Consider specialized agents**: If another agent has more relevant skills or expertise for this request, call the `transfer_to_agent` function to transfer to that agent. Match the user's needs with the agent's skills and descriptions above.

3. **When transferring**: Only call the function - do not generate any additional text.

Agent name: peer_agent
Agent description: Peer agent


If you are the best to answer the question according to your description, you
can answer it.

If another agent is better for answering the question according to its
description, call `transfer_to_agent` function to transfer the
question to that agent. When transferring, do not generate any text other than
the function call.

**NOTE**: the only available agents for `transfer_to_agent` function are `a_agent`, `m_agent`, `parent_agent`, `peer_agent`, `z_agent`.

If neither you nor the other agents are best for the question, transfer to your parent agent parent_agent."""
**IMPORTANT**: The only valid agent names for `transfer_to_agent` are: `a_agent`, `m_agent`, `parent_agent`, `peer_agent`, `z_agent`
"""

assert expected_content in instructions

# Also verify the parent escalation instruction is present
assert '4. **Escalate to parent**: If neither you nor the specialized agents are suitable for this request, transfer to your parent agent `parent_agent` for broader assistance.' in instructions


@pytest.mark.asyncio
async def test_agent_transfer_system_instructions_without_parent():
Expand Down Expand Up @@ -177,30 +174,32 @@ async def test_agent_transfer_system_instructions_without_parent():

# Direct multiline string assertion showing the exact expected content
expected_content = """\
## Available Agents for Transfer

You have a list of other agents to transfer to:
You can delegate tasks to the following specialized agents. Carefully review each agent's description and skills to determine the best match for the user's request.

### Agent: agent1
Description: First sub-agent

Agent name: agent1
Agent description: First sub-agent
### Agent: agent2
Description: Second sub-agent

## Decision Criteria

Agent name: agent2
Agent description: Second sub-agent
1. **Assess your own capability**: If you are the best agent to handle this request based on your own description and capabilities, answer it directly.

2. **Consider specialized agents**: If another agent has more relevant skills or expertise for this request, call the `transfer_to_agent` function to transfer to that agent. Match the user's needs with the agent's skills and descriptions above.

If you are the best to answer the question according to your description, you
can answer it.
3. **When transferring**: Only call the function - do not generate any additional text.

If another agent is better for answering the question according to its
description, call `transfer_to_agent` function to transfer the
question to that agent. When transferring, do not generate any text other than
the function call.

**NOTE**: the only available agents for `transfer_to_agent` function are `agent1`, `agent2`."""
**IMPORTANT**: The only valid agent names for `transfer_to_agent` are: `agent1`, `agent2`
"""

assert expected_content in instructions

# Verify no parent escalation instruction is present
assert 'Escalate to parent' not in instructions


@pytest.mark.asyncio
async def test_agent_transfer_simplified_parent_instructions():
Expand Down Expand Up @@ -236,32 +235,32 @@ async def test_agent_transfer_simplified_parent_instructions():

# Direct multiline string assertion showing the exact expected content
expected_content = """\
## Available Agents for Transfer

You have a list of other agents to transfer to:

You can delegate tasks to the following specialized agents. Carefully review each agent's description and skills to determine the best match for the user's request.

Agent name: sub_agent
Agent description: Sub agent
### Agent: sub_agent
Description: Sub agent

### Agent: parent_agent
Description: Parent agent

Agent name: parent_agent
Agent description: Parent agent
## Decision Criteria

1. **Assess your own capability**: If you are the best agent to handle this request based on your own description and capabilities, answer it directly.

If you are the best to answer the question according to your description, you
can answer it.
2. **Consider specialized agents**: If another agent has more relevant skills or expertise for this request, call the `transfer_to_agent` function to transfer to that agent. Match the user's needs with the agent's skills and descriptions above.

If another agent is better for answering the question according to its
description, call `transfer_to_agent` function to transfer the
question to that agent. When transferring, do not generate any text other than
the function call.
3. **When transferring**: Only call the function - do not generate any additional text.

**NOTE**: the only available agents for `transfer_to_agent` function are `parent_agent`, `sub_agent`.

If neither you nor the other agents are best for the question, transfer to your parent agent parent_agent."""
**IMPORTANT**: The only valid agent names for `transfer_to_agent` are: `parent_agent`, `sub_agent`
"""

assert expected_content in instructions

# Also verify the parent escalation instruction is present
assert '4. **Escalate to parent**: If neither you nor the specialized agents are suitable for this request, transfer to your parent agent `parent_agent` for broader assistance.' in instructions


@pytest.mark.asyncio
async def test_agent_transfer_no_instructions_when_no_transfer_targets():
Expand Down