wehub-resource-sync
ec2b666284
Continuous Integration / Pre-commit Linter (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.10) (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.11) (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.12) (push) Has been cancelled
Continuous Integration / Mypy Check (Python 3.13) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.10) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.11) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.12) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.13) (push) Has been cancelled
Continuous Integration / Unit Tests (Python 3.14) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.10) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.11) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.12) (push) Has been cancelled
Copybara PR Handler / close-imported-pr (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.13) (push) Has been cancelled
Continuous Integration / A2A v0.3 Tests (Python 3.14) (push) Has been cancelled
495 lines
16 KiB
Python
495 lines
16 KiB
Python
# Copyright 2026 Google LLC
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Behavioral tests for other agent message processing in contents module."""
|
|
|
|
from google.adk.agents.llm_agent import Agent
|
|
from google.adk.agents.run_config import RunConfig
|
|
from google.adk.events.event import Event
|
|
from google.adk.flows.llm_flows.contents import request_processor
|
|
from google.adk.models.llm_request import LlmRequest
|
|
from google.genai import types
|
|
import pytest
|
|
|
|
from ... import testing_utils
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_message_appears_as_user_context():
|
|
"""Test that messages from other agents appear as user context."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add event from another agent
|
|
other_agent_event = Event(
|
|
invocation_id="test_inv",
|
|
author="other_agent",
|
|
content=types.ModelContent("Hello from other agent"),
|
|
)
|
|
invocation_context.session.events = [other_agent_event]
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify the other agent's message is presented as user context
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] said: Hello from other agent"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_thoughts_are_excluded():
|
|
"""Test that thoughts from other agents are excluded from context."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add event from other agent with both regular text and thoughts
|
|
other_agent_event = Event(
|
|
invocation_id="test_inv",
|
|
author="other_agent",
|
|
content=types.ModelContent([
|
|
types.Part(text="Public message", thought=False),
|
|
types.Part(text="Private thought", thought=True),
|
|
types.Part(text="Another public message"),
|
|
]),
|
|
)
|
|
invocation_context.session.events = [other_agent_event]
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify only non-thought parts are included (thoughts excluded)
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] said: Public message"),
|
|
types.Part(text="[other_agent] said: Another public message"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_thoughts_can_be_included_as_context():
|
|
"""Test opt-in inclusion of thoughts from other agents."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent,
|
|
run_config=RunConfig(include_thoughts_from_other_agents=True),
|
|
)
|
|
other_agent_event = Event(
|
|
invocation_id="test_inv",
|
|
author="other_agent",
|
|
content=types.ModelContent([
|
|
types.Part(text="Public message", thought=False),
|
|
types.Part(text="Private thought", thought=True),
|
|
types.Part(text="Another public message"),
|
|
]),
|
|
)
|
|
invocation_context.session.events = [other_agent_event]
|
|
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] said: Public message"),
|
|
types.Part(text="[other_agent] thought: Private thought"),
|
|
types.Part(text="[other_agent] said: Another public message"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_thought_only_message_can_be_included_as_context():
|
|
"""Test opt-in inclusion of thought-only messages from other agents."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent,
|
|
run_config=RunConfig(include_thoughts_from_other_agents=True),
|
|
)
|
|
other_agent_event = Event(
|
|
invocation_id="test_inv",
|
|
author="other_agent",
|
|
content=types.ModelContent([
|
|
types.Part(text="First private thought", thought=True),
|
|
types.Part(text="Second private thought", thought=True),
|
|
]),
|
|
)
|
|
invocation_context.session.events = [other_agent_event]
|
|
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] thought: First private thought"),
|
|
types.Part(text="[other_agent] thought: Second private thought"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_thoughts_excluded_from_current_turn_only_context():
|
|
"""Test include_contents='none' does not include other-agent thoughts."""
|
|
agent = Agent(
|
|
model="gemini-2.5-flash",
|
|
name="current_agent",
|
|
include_contents="none",
|
|
)
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent,
|
|
run_config=RunConfig(include_thoughts_from_other_agents=True),
|
|
)
|
|
invocation_context.session.events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("Earlier user message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="other_agent",
|
|
content=types.ModelContent([
|
|
types.Part(text="Private thought", thought=True),
|
|
types.Part(text="Visible handoff"),
|
|
]),
|
|
),
|
|
]
|
|
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
assert llm_request.contents == [
|
|
types.Content(
|
|
role="user",
|
|
parts=[
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] said: Visible handoff"),
|
|
],
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_function_calls():
|
|
"""Test that function calls from other agents are preserved in context."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add event from other agent with function call
|
|
function_call = types.FunctionCall(
|
|
id="func_123", name="search_tool", args={"query": "test query"}
|
|
)
|
|
other_agent_event = Event(
|
|
invocation_id="test_inv",
|
|
author="other_agent",
|
|
content=types.ModelContent([types.Part(function_call=function_call)]),
|
|
)
|
|
invocation_context.session.events = [other_agent_event]
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify function call is presented as context
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(
|
|
text="""\
|
|
[other_agent] called tool `search_tool` with parameters: {'query': 'test query'}"""
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_function_responses():
|
|
"""Test that function responses from other agents are properly formatted."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
# Add event from other agent with function response
|
|
function_response = types.FunctionResponse(
|
|
id="func_123",
|
|
name="search_tool",
|
|
response={"results": ["item1", "item2"]},
|
|
)
|
|
other_agent_event = Event(
|
|
invocation_id="test_inv",
|
|
author="other_agent",
|
|
content=types.Content(
|
|
role="user", parts=[types.Part(function_response=function_response)]
|
|
),
|
|
)
|
|
invocation_context.session.events = [other_agent_event]
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify function response is presented as context
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(
|
|
text=(
|
|
"[other_agent] `search_tool` tool returned result: {'results':"
|
|
" ['item1', 'item2']}"
|
|
)
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_function_call_response():
|
|
"""Test function call and response sequence from other agents."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add function call event from other agent
|
|
function_call = types.FunctionCall(
|
|
id="func_123", name="calc_tool", args={"query": "6x7"}
|
|
)
|
|
call_event = Event(
|
|
invocation_id="test_inv1",
|
|
author="other_agent",
|
|
content=types.ModelContent([
|
|
types.Part(text="Let me calculate this"),
|
|
types.Part(function_call=function_call),
|
|
]),
|
|
)
|
|
# Add function response event
|
|
function_response = types.FunctionResponse(
|
|
id="func_123", name="calc_tool", response={"result": 42}
|
|
)
|
|
response_event = Event(
|
|
invocation_id="test_inv2",
|
|
author="other_agent",
|
|
content=types.UserContent(
|
|
parts=[types.Part(function_response=function_response)]
|
|
),
|
|
)
|
|
invocation_context.session.events = [call_event, response_event]
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify function call and response are properly formatted
|
|
assert len(llm_request.contents) == 2
|
|
|
|
# Function call from other agent
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] said: Let me calculate this"),
|
|
types.Part(
|
|
text=(
|
|
"[other_agent] called tool `calc_tool` with parameters: {'query':"
|
|
" '6x7'}"
|
|
)
|
|
),
|
|
]
|
|
# Function response from other agent
|
|
assert llm_request.contents[1].role == "user"
|
|
assert llm_request.contents[1].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(
|
|
text="[other_agent] `calc_tool` tool returned result: {'result': 42}"
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_agent_empty_content():
|
|
"""Test that other agent messages with only thoughts or empty content are filtered out."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add events: user message, other agents with empty content, user message
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("Hello"),
|
|
),
|
|
# Other agent with only thoughts
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="other_agent1",
|
|
content=types.ModelContent([
|
|
types.Part(text="This is a private thought", thought=True),
|
|
types.Part(text="Another private thought", thought=True),
|
|
]),
|
|
),
|
|
# Other agent with empty text and thoughts
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="other_agent2",
|
|
content=types.ModelContent([
|
|
types.Part(text="", thought=False),
|
|
types.Part(text="Secret thought", thought=True),
|
|
]),
|
|
),
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="user",
|
|
content=types.UserContent("World"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify empty content events are completely filtered out
|
|
assert llm_request.contents == [
|
|
types.UserContent("Hello"),
|
|
types.UserContent("World"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_agents_in_conversation():
|
|
"""Test handling multiple agents in a conversation flow."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
# Create a multi-agent conversation
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("Hello everyone"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="agent1",
|
|
content=types.ModelContent("Hi from agent1"),
|
|
),
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="agent2",
|
|
content=types.ModelContent("Hi from agent2"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify all messages are properly processed
|
|
assert len(llm_request.contents) == 3
|
|
|
|
# User message should remain as user
|
|
assert llm_request.contents[0] == types.UserContent("Hello everyone")
|
|
# Other agents' messages should be converted to user context
|
|
assert llm_request.contents[1].role == "user"
|
|
assert llm_request.contents[1].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[agent1] said: Hi from agent1"),
|
|
]
|
|
assert llm_request.contents[2].role == "user"
|
|
assert llm_request.contents[2].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[agent2] said: Hi from agent2"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_current_agent_messages_not_converted():
|
|
"""Test that the current agent's own messages are not converted."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add events from both current agent and other agent
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="current_agent",
|
|
content=types.ModelContent("My own message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="other_agent",
|
|
content=types.ModelContent("Other agent message"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify current agent's message stays as model role
|
|
# and other agent's message is converted to user context
|
|
assert len(llm_request.contents) == 2
|
|
assert llm_request.contents[0] == types.ModelContent("My own message")
|
|
assert llm_request.contents[1].role == "user"
|
|
assert llm_request.contents[1].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[other_agent] said: Other agent message"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_messages_preserved():
|
|
"""Test that user messages are preserved as-is."""
|
|
agent = Agent(model="gemini-2.5-flash", name="current_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
# Add user message
|
|
user_event = Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("User message"),
|
|
)
|
|
invocation_context.session.events = [user_event]
|
|
|
|
# Process the request
|
|
async for _ in request_processor.run_async(invocation_context, llm_request):
|
|
pass
|
|
|
|
# Verify user message is preserved exactly
|
|
assert len(llm_request.contents) == 1
|
|
assert llm_request.contents[0] == types.UserContent("User message")
|