mirror of
https://github.com/encounter/adk-python.git
synced 2026-07-09 18:19:28 -07:00
Wei Sun (Jack)andCopybara-Service
a30851ee16
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
373 lines
11 KiB
Python
373 lines
11 KiB
Python
# 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.
|
|
|
|
from google.adk.agents.llm_agent import Agent
|
|
from google.adk.events.event import Event
|
|
from google.adk.events.event_actions import EventActions
|
|
from google.adk.flows.llm_flows import contents
|
|
from google.adk.flows.llm_flows.functions import REQUEST_CONFIRMATION_FUNCTION_CALL_NAME
|
|
from google.adk.flows.llm_flows.functions import REQUEST_EUC_FUNCTION_CALL_NAME
|
|
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_include_contents_default_full_history():
|
|
"""Test that include_contents='default' includes full conversation history."""
|
|
agent = Agent(
|
|
model="gemini-2.5-flash", name="test_agent", include_contents="default"
|
|
)
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
# Create a multi-turn conversation
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("First message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="test_agent",
|
|
content=types.ModelContent("First response"),
|
|
),
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="user",
|
|
content=types.UserContent("Second message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="test_agent",
|
|
content=types.ModelContent("Second response"),
|
|
),
|
|
Event(
|
|
invocation_id="inv5",
|
|
author="user",
|
|
content=types.UserContent("Third message"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in contents.request_processor.run_async(
|
|
invocation_context, llm_request
|
|
):
|
|
pass
|
|
|
|
# Verify full conversation history is included
|
|
assert llm_request.contents == [
|
|
types.UserContent("First message"),
|
|
types.ModelContent("First response"),
|
|
types.UserContent("Second message"),
|
|
types.ModelContent("Second response"),
|
|
types.UserContent("Third message"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_include_contents_none_current_turn_only():
|
|
"""Test that include_contents='none' includes only current turn context."""
|
|
agent = Agent(
|
|
model="gemini-2.5-flash", name="test_agent", include_contents="none"
|
|
)
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
# Create a multi-turn conversation
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("First message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="test_agent",
|
|
content=types.ModelContent("First response"),
|
|
),
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="user",
|
|
content=types.UserContent("Second message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="test_agent",
|
|
content=types.ModelContent("Second response"),
|
|
),
|
|
Event(
|
|
invocation_id="inv5",
|
|
author="user",
|
|
content=types.UserContent("Current turn message"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in contents.request_processor.run_async(
|
|
invocation_context, llm_request
|
|
):
|
|
pass
|
|
|
|
# Verify only current turn is included (from last user message)
|
|
assert llm_request.contents == [
|
|
types.UserContent("Current turn message"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_include_contents_none_multi_agent_current_turn():
|
|
"""Test current turn detection in multi-agent scenarios with include_contents='none'."""
|
|
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
|
|
)
|
|
|
|
# Create multi-agent conversation where current turn starts from user
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("First user message"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="other_agent",
|
|
content=types.ModelContent("Other agent response"),
|
|
),
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="current_agent",
|
|
content=types.ModelContent("Current agent first response"),
|
|
),
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="user",
|
|
content=types.UserContent("Current turn request"),
|
|
),
|
|
Event(
|
|
invocation_id="inv5",
|
|
author="another_agent",
|
|
content=types.ModelContent("Another agent responds"),
|
|
),
|
|
Event(
|
|
invocation_id="inv6",
|
|
author="current_agent",
|
|
content=types.ModelContent("Current agent in turn"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in contents.request_processor.run_async(
|
|
invocation_context, llm_request
|
|
):
|
|
pass
|
|
|
|
# Verify current turn starts from the most recent other agent message (inv5)
|
|
assert len(llm_request.contents) == 2
|
|
assert llm_request.contents[0].role == "user"
|
|
assert llm_request.contents[0].parts == [
|
|
types.Part(text="For context:"),
|
|
types.Part(text="[another_agent] said: Another agent responds"),
|
|
]
|
|
assert llm_request.contents[1] == types.ModelContent("Current agent in turn")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authentication_events_are_filtered():
|
|
"""Test that authentication function calls and responses are filtered out."""
|
|
agent = Agent(model="gemini-2.5-flash", name="test_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
# Create authentication function call and response
|
|
auth_function_call = types.FunctionCall(
|
|
id="auth_123",
|
|
name=REQUEST_EUC_FUNCTION_CALL_NAME,
|
|
args={"credential_type": "oauth"},
|
|
)
|
|
auth_response = types.FunctionResponse(
|
|
id="auth_123",
|
|
name=REQUEST_EUC_FUNCTION_CALL_NAME,
|
|
response={
|
|
"auth_config": {"exchanged_auth_credential": {"token": "secret"}}
|
|
},
|
|
)
|
|
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("Please authenticate"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="test_agent",
|
|
content=types.ModelContent(
|
|
[types.Part(function_call=auth_function_call)]
|
|
),
|
|
),
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="user",
|
|
content=types.Content(
|
|
parts=[types.Part(function_response=auth_response)], role="user"
|
|
),
|
|
),
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="user",
|
|
content=types.UserContent("Continue after auth"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in contents.request_processor.run_async(
|
|
invocation_context, llm_request
|
|
):
|
|
pass
|
|
|
|
# Verify both authentication call and response are filtered out
|
|
assert llm_request.contents == [
|
|
types.UserContent("Please authenticate"),
|
|
types.UserContent("Continue after auth"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirmation_events_are_filtered():
|
|
"""Test that confirmation function calls and responses are filtered out."""
|
|
agent = Agent(model="gemini-2.5-flash", name="test_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
# Create confirmation function call and response
|
|
confirmation_function_call = types.FunctionCall(
|
|
id="confirm_123",
|
|
name=REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
|
|
args={"action": "delete_file", "confirmation": True},
|
|
)
|
|
confirmation_response = types.FunctionResponse(
|
|
id="confirm_123",
|
|
name=REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
|
|
response={"response": '{"confirmed": true}'},
|
|
)
|
|
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("Delete the file"),
|
|
),
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="test_agent",
|
|
content=types.ModelContent(
|
|
[types.Part(function_call=confirmation_function_call)]
|
|
),
|
|
),
|
|
Event(
|
|
invocation_id="inv3",
|
|
author="user",
|
|
content=types.Content(
|
|
parts=[types.Part(function_response=confirmation_response)],
|
|
role="user",
|
|
),
|
|
),
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="user",
|
|
content=types.UserContent("File deleted successfully"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in contents.request_processor.run_async(
|
|
invocation_context, llm_request
|
|
):
|
|
pass
|
|
|
|
# Verify both confirmation call and response are filtered out
|
|
assert llm_request.contents == [
|
|
types.UserContent("Delete the file"),
|
|
types.UserContent("File deleted successfully"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_with_empty_content_are_skipped():
|
|
"""Test that events with empty content (state-only changes) are skipped."""
|
|
agent = Agent(model="gemini-2.5-flash", name="test_agent")
|
|
llm_request = LlmRequest(model="gemini-2.5-flash")
|
|
invocation_context = await testing_utils.create_invocation_context(
|
|
agent=agent
|
|
)
|
|
|
|
events = [
|
|
Event(
|
|
invocation_id="inv1",
|
|
author="user",
|
|
content=types.UserContent("Hello"),
|
|
),
|
|
# Event with no content (state-only change)
|
|
Event(
|
|
invocation_id="inv2",
|
|
author="test_agent",
|
|
actions=EventActions(state_delta={"key": "val"}),
|
|
),
|
|
# Event with content that has no meaningful parts
|
|
Event(
|
|
invocation_id="inv4",
|
|
author="test_agent",
|
|
content=types.Content(parts=[], role="model"),
|
|
),
|
|
Event(
|
|
invocation_id="inv5",
|
|
author="user",
|
|
content=types.UserContent("How are you?"),
|
|
),
|
|
]
|
|
invocation_context.session.events = events
|
|
|
|
# Process the request
|
|
async for _ in contents.request_processor.run_async(
|
|
invocation_context, llm_request
|
|
):
|
|
pass
|
|
|
|
# Verify only events with meaningful content are included
|
|
assert llm_request.contents == [
|
|
types.UserContent("Hello"),
|
|
types.UserContent("How are you?"),
|
|
]
|