mirror of
https://github.com/encounter/adk-python.git
synced 2026-07-09 18:19:28 -07:00
fix: Fixes thought handling in contents.py and refactors its unit tests
Before this change, other agent's reply with thought will still be inserted in the outgoing LlmRequest due to the wrong `else` statement for calling all other type of part. This commit also refactors test_contents.py to be behavior-oriented tests, instead of implementation-oriented, and add more test cases to cover expected scenarios. The tests are divided into the following files with different focus: - test_contents.py: covers the basic logic of event filter; - test_contents_branch.py: covers the behavior related to branch, which takes effect when ParallelAgent is used. - test_contents_other_agent.py: covers the retelling behavior to include other agents' reply as context for the current agent. - test_contents_function.py: covers the function_call/function_response rearrangement logic mainly for `LongRunningFunctionTool`. PiperOrigin-RevId: 802759821
This commit is contained in:
Wei Sun (Jack)
committed by
Copybara-Service
Copybara-Service
parent
fe8b37b0d3
commit
a30851ee16
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,288 @@
|
||||
# Copyright 2025 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.
|
||||
|
||||
"""Tests for branch filtering in contents module.
|
||||
|
||||
Branch format: agent_1.agent_2.agent_3 (parent.child.grandchild)
|
||||
Child agents can see parent agents' events, but not sibling agents' events.
|
||||
"""
|
||||
|
||||
from google.adk.agents.llm_agent import Agent
|
||||
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_branch_filtering_child_sees_parent():
|
||||
"""Test that child agents can see parent agents' events."""
|
||||
agent = Agent(model="gemini-2.5-flash", name="child_agent")
|
||||
llm_request = LlmRequest(model="gemini-2.5-flash")
|
||||
invocation_context = await testing_utils.create_invocation_context(
|
||||
agent=agent
|
||||
)
|
||||
# Set current branch as child of "parent_agent"
|
||||
invocation_context.branch = "parent_agent.child_agent"
|
||||
|
||||
# Add events from parent and child levels
|
||||
events = [
|
||||
Event(
|
||||
invocation_id="inv1",
|
||||
author="user",
|
||||
content=types.UserContent("User message"),
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv2",
|
||||
author="parent_agent",
|
||||
content=types.ModelContent("Parent agent response"),
|
||||
branch="parent_agent", # Parent branch - should be included
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv3",
|
||||
author="child_agent",
|
||||
content=types.ModelContent("Child agent response"),
|
||||
branch="parent_agent.child_agent", # Current branch - should be included
|
||||
),
|
||||
]
|
||||
invocation_context.session.events = events
|
||||
|
||||
# Process the request
|
||||
async for _ in request_processor.run_async(invocation_context, llm_request):
|
||||
pass
|
||||
|
||||
# Verify child can see user message and parent events, but not sibling events
|
||||
assert len(llm_request.contents) == 3
|
||||
assert llm_request.contents[0] == types.UserContent("User message")
|
||||
assert llm_request.contents[1].role == "user"
|
||||
assert llm_request.contents[1].parts == [
|
||||
types.Part(text="For context:"),
|
||||
types.Part(text="[parent_agent] said: Parent agent response"),
|
||||
]
|
||||
assert llm_request.contents[2] == types.ModelContent("Child agent response")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_branch_filtering_excludes_sibling_agents():
|
||||
"""Test that sibling agents cannot see each other's events."""
|
||||
agent = Agent(model="gemini-2.5-flash", name="child_agent1")
|
||||
llm_request = LlmRequest(model="gemini-2.5-flash")
|
||||
invocation_context = await testing_utils.create_invocation_context(
|
||||
agent=agent
|
||||
)
|
||||
# Set current branch as first child
|
||||
invocation_context.branch = "parent_agent.child_agent1"
|
||||
|
||||
# Add events from parent, current child, and sibling child
|
||||
events = [
|
||||
Event(
|
||||
invocation_id="inv1",
|
||||
author="user",
|
||||
content=types.UserContent("User message"),
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv2",
|
||||
author="parent_agent",
|
||||
content=types.ModelContent("Parent response"),
|
||||
branch="parent_agent", # Parent - should be included
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv3",
|
||||
author="child_agent1",
|
||||
content=types.ModelContent("Child1 response"),
|
||||
branch="parent_agent.child_agent1", # Current - should be included
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv4",
|
||||
author="child_agent2",
|
||||
content=types.ModelContent("Sibling response"),
|
||||
branch="parent_agent.child_agent2", # Sibling - should be excluded
|
||||
),
|
||||
]
|
||||
invocation_context.session.events = events
|
||||
|
||||
# Process the request
|
||||
async for _ in request_processor.run_async(invocation_context, llm_request):
|
||||
pass
|
||||
|
||||
# Verify sibling events are excluded, but parent and current agent events included
|
||||
assert len(llm_request.contents) == 3
|
||||
assert llm_request.contents[0] == types.UserContent("User message")
|
||||
assert llm_request.contents[1].role == "user"
|
||||
assert llm_request.contents[1].parts == [
|
||||
types.Part(text="For context:"),
|
||||
types.Part(text="[parent_agent] said: Parent response"),
|
||||
]
|
||||
assert llm_request.contents[2] == types.ModelContent("Child1 response")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_branch_filtering_no_branch_allows_all():
|
||||
"""Test that events are included when no branches are set."""
|
||||
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
|
||||
)
|
||||
# No current branch set (None)
|
||||
invocation_context.branch = None
|
||||
|
||||
# Add events with and without branches
|
||||
events = [
|
||||
Event(
|
||||
invocation_id="inv1",
|
||||
author="user",
|
||||
content=types.UserContent("No branch message"),
|
||||
branch=None,
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv2",
|
||||
author="agent1",
|
||||
content=types.ModelContent("Agent with branch"),
|
||||
branch="agent1",
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv3",
|
||||
author="user",
|
||||
content=types.UserContent("Another no branch"),
|
||||
branch=None,
|
||||
),
|
||||
]
|
||||
invocation_context.session.events = events
|
||||
|
||||
# Process the request
|
||||
async for _ in request_processor.run_async(invocation_context, llm_request):
|
||||
pass
|
||||
|
||||
# Verify all events are included when no current branch
|
||||
assert len(llm_request.contents) == 3
|
||||
assert llm_request.contents[0] == types.UserContent("No branch message")
|
||||
assert llm_request.contents[1].role == "user"
|
||||
assert llm_request.contents[1].parts == [
|
||||
types.Part(text="For context:"),
|
||||
types.Part(text="[agent1] said: Agent with branch"),
|
||||
]
|
||||
assert llm_request.contents[2] == types.UserContent("Another no branch")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_branch_filtering_grandchild_sees_grandparent():
|
||||
"""Test that deeply nested child agents can see all ancestor events."""
|
||||
agent = Agent(model="gemini-2.5-flash", name="grandchild_agent")
|
||||
llm_request = LlmRequest(model="gemini-2.5-flash")
|
||||
invocation_context = await testing_utils.create_invocation_context(
|
||||
agent=agent
|
||||
)
|
||||
# Set deeply nested branch: grandparent.parent.grandchild
|
||||
invocation_context.branch = "grandparent_agent.parent_agent.grandchild_agent"
|
||||
|
||||
# Add events from all levels of hierarchy
|
||||
events = [
|
||||
Event(
|
||||
invocation_id="inv1",
|
||||
author="grandparent_agent",
|
||||
content=types.ModelContent("Grandparent response"),
|
||||
branch="grandparent_agent",
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv2",
|
||||
author="parent_agent",
|
||||
content=types.ModelContent("Parent response"),
|
||||
branch="grandparent_agent.parent_agent",
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv3",
|
||||
author="grandchild_agent",
|
||||
content=types.ModelContent("Grandchild response"),
|
||||
branch="grandparent_agent.parent_agent.grandchild_agent",
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv4",
|
||||
author="sibling_agent",
|
||||
content=types.ModelContent("Sibling response"),
|
||||
branch="grandparent_agent.parent_agent.sibling_agent",
|
||||
),
|
||||
]
|
||||
invocation_context.session.events = events
|
||||
|
||||
# Process the request
|
||||
async for _ in request_processor.run_async(invocation_context, llm_request):
|
||||
pass
|
||||
|
||||
# Verify only ancestors and current level are included
|
||||
assert len(llm_request.contents) == 3
|
||||
assert llm_request.contents[0].role == "user"
|
||||
assert llm_request.contents[0].parts == [
|
||||
types.Part(text="For context:"),
|
||||
types.Part(text="[grandparent_agent] said: Grandparent response"),
|
||||
]
|
||||
assert llm_request.contents[1].role == "user"
|
||||
assert llm_request.contents[1].parts == [
|
||||
types.Part(text="For context:"),
|
||||
types.Part(text="[parent_agent] said: Parent response"),
|
||||
]
|
||||
assert llm_request.contents[2] == types.ModelContent("Grandchild response")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_branch_filtering_parent_cannot_see_child():
|
||||
"""Test that parent agents cannot see child agents' events."""
|
||||
agent = Agent(model="gemini-2.5-flash", name="parent_agent")
|
||||
llm_request = LlmRequest(model="gemini-2.5-flash")
|
||||
invocation_context = await testing_utils.create_invocation_context(
|
||||
agent=agent
|
||||
)
|
||||
# Set current branch as parent
|
||||
invocation_context.branch = "parent_agent"
|
||||
|
||||
# Add events from parent and its children
|
||||
events = [
|
||||
Event(
|
||||
invocation_id="inv1",
|
||||
author="user",
|
||||
content=types.UserContent("User message"),
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv2",
|
||||
author="parent_agent",
|
||||
content=types.ModelContent("Parent response"),
|
||||
branch="parent_agent",
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv3",
|
||||
author="child_agent",
|
||||
content=types.ModelContent("Child response"),
|
||||
branch="parent_agent.child_agent",
|
||||
),
|
||||
Event(
|
||||
invocation_id="inv4",
|
||||
author="grandchild_agent",
|
||||
content=types.ModelContent("Grandchild response"),
|
||||
branch="parent_agent.child_agent.grandchild_agent",
|
||||
),
|
||||
]
|
||||
invocation_context.session.events = events
|
||||
|
||||
# Process the request
|
||||
async for _ in request_processor.run_async(invocation_context, llm_request):
|
||||
pass
|
||||
|
||||
# Verify parent cannot see child or grandchild events
|
||||
assert llm_request.contents == [
|
||||
types.UserContent("User message"),
|
||||
types.ModelContent("Parent response"),
|
||||
]
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,388 @@
|
||||
# Copyright 2025 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.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_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")
|
||||
Reference in New Issue
Block a user