google--adk-python/tests/unittests/apps/test_llm_event_summarizer.py

# 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.

import unittest
from unittest.mock import AsyncMock

from google.adk.apps.llm_event_summarizer import LlmEventSummarizer
from google.adk.events.event import Event
from google.adk.events.event_actions import EventActions
from google.adk.events.event_actions import EventCompaction
from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.genai import types
from google.genai.types import Content
from google.genai.types import FunctionCall
from google.genai.types import FunctionResponse
from google.genai.types import Part
import pytest


@pytest.mark.parametrize(
    'env_variables', ['GOOGLE_AI', 'VERTEX'], indirect=True
)
class TestLlmEventSummarizer(unittest.IsolatedAsyncioTestCase):

  def setUp(self):
    self.mock_llm = AsyncMock(spec=BaseLlm)
    self.mock_llm.model = 'test-model'
    self.compactor = LlmEventSummarizer(llm=self.mock_llm)

  def _create_event(
      self, timestamp: float, text: str, author: str = 'user'
  ) -> Event:
    return Event(
        timestamp=timestamp,
        author=author,
        content=Content(parts=[Part(text=text)]),
    )

  async def test_maybe_compact_events_success(self):
    events = [
        self._create_event(1.0, 'Hello', 'user'),
        self._create_event(2.0, 'Hi there!', 'model'),
    ]
    expected_conversation_history = 'user: Hello\nmodel: Hi there!'
    expected_prompt = self.compactor._DEFAULT_PROMPT_TEMPLATE.format(
        conversation_history=expected_conversation_history
    )
    llm_response = LlmResponse(
        content=Content(parts=[Part(text='Summary')]),
        usage_metadata=None,
    )

    async def async_gen():
      yield llm_response

    self.mock_llm.generate_content_async.return_value = async_gen()

    compacted_event = await self.compactor.maybe_summarize_events(events=events)

    self.assertIsNotNone(compacted_event)
    self.assertEqual(
        compacted_event.actions.compaction.compacted_content.parts[0].text,
        'Summary',
    )
    self.assertEqual(compacted_event.author, 'user')
    self.assertIsNone(compacted_event.usage_metadata)
    self.assertIsNotNone(compacted_event.actions)
    self.assertIsNotNone(compacted_event.actions.compaction)
    self.assertEqual(compacted_event.actions.compaction.start_timestamp, 1.0)
    self.assertEqual(compacted_event.actions.compaction.end_timestamp, 2.0)
    self.assertEqual(
        compacted_event.actions.compaction.compacted_content.parts[0].text,
        'Summary',
    )

    self.mock_llm.generate_content_async.assert_called_once()
    args, kwargs = self.mock_llm.generate_content_async.call_args
    llm_request = args[0]
    self.assertIsInstance(llm_request, LlmRequest)
    self.assertEqual(llm_request.model, 'test-model')
    self.assertEqual(llm_request.contents[0].role, 'user')
    self.assertEqual(llm_request.contents[0].parts[0].text, expected_prompt)
    self.assertFalse(kwargs['stream'])

  async def test_maybe_compact_events_empty_llm_response(self):
    events = [
        self._create_event(1.0, 'Hello', 'user'),
    ]
    llm_response = LlmResponse(content=None, usage_metadata=None)

    async def async_gen():
      yield llm_response

    self.mock_llm.generate_content_async.return_value = async_gen()

    compacted_event = await self.compactor.maybe_summarize_events(events=events)
    self.assertIsNone(compacted_event)

  async def test_maybe_compact_events_includes_usage_metadata(self):
    events = [
        self._create_event(1.0, 'Hello', 'user'),
        self._create_event(2.0, 'Hi there!', 'model'),
    ]
    usage_metadata = types.GenerateContentResponseUsageMetadata(
        prompt_token_count=10,
        candidates_token_count=5,
    )
    llm_response = LlmResponse(
        content=Content(parts=[Part(text='Summary')]),
        usage_metadata=usage_metadata,
    )

    async def async_gen():
      yield llm_response

    self.mock_llm.generate_content_async.return_value = async_gen()

    compacted_event = await self.compactor.maybe_summarize_events(events=events)

    self.assertIsNotNone(compacted_event)
    self.assertEqual(compacted_event.usage_metadata, usage_metadata)
    self.assertEqual(compacted_event.usage_metadata.prompt_token_count, 10)
    self.assertEqual(compacted_event.usage_metadata.candidates_token_count, 5)

  async def test_maybe_compact_events_empty_input(self):
    compacted_event = await self.compactor.maybe_summarize_events(events=[])
    self.assertIsNone(compacted_event)
    self.mock_llm.generate_content_async.assert_not_called()

  def test_format_events_for_prompt(self):
    events = [
        self._create_event(1.0, 'User says...', 'user'),
        self._create_event(2.0, 'Model replies...', 'model'),
        self._create_event(3.0, 'Another user input', 'user'),
        self._create_event(4.0, 'More model text', 'model'),
        # Event with no content
        Event(timestamp=5.0, author='user'),
        # Event with empty content part
        Event(
            timestamp=6.0,
            author='model',
            content=Content(parts=[Part(text='')]),
        ),
        # Event with function call
        Event(
            timestamp=7.0,
            author='model',
            content=Content(
                parts=[
                    Part(
                        function_call=FunctionCall(
                            id='call_1', name='tool', args={'q': 'x'}
                        )
                    )
                ]
            ),
        ),
        # Event with function response
        Event(
            timestamp=8.0,
            author='model',
            content=Content(
                parts=[
                    Part(
                        function_response=FunctionResponse(
                            id='call_1',
                            name='tool',
                            response={'result': 'done'},
                        )
                    )
                ]
            ),
        ),
    ]
    expected_formatted_history = (
        'user: User says...\nmodel: Model replies...\nuser: Another user'
        ' input\nmodel: More model text\nmodel called tool:'
        " tool({'q': 'x'})\nTool response from tool: {'result': 'done'}"
    )
    formatted_history = self.compactor._format_events_for_prompt(events)
    self.assertEqual(formatted_history, expected_formatted_history)

  def test_format_events_for_prompt_includes_thoughts(self):
    events = [
        self._create_event(1.0, 'What is the weather?', 'user'),
        Event(
            timestamp=2.0,
            author='model',
            content=Content(
                parts=[
                    Part(text='Let me check the tool output.', thought=True),
                    Part(text='It is sunny.'),
                ]
            ),
        ),
    ]
    expected_formatted_history = (
        'user: What is the weather?\nmodel (thought): Let me check the tool'
        ' output.\nmodel: It is sunny.'
    )
    formatted_history = self.compactor._format_events_for_prompt(events)
    self.assertEqual(formatted_history, expected_formatted_history)

  def test_format_events_for_prompt_skips_compaction_event_thought(self):
    events = [
        Event(
            timestamp=1.0,
            author='model',
            content=Content(
                parts=[
                    Part(text='Stale summarizer reasoning.', thought=True),
                    Part(text='Prior summary.'),
                ]
            ),
            actions=EventActions(
                compaction=EventCompaction(
                    start_timestamp=0.0,
                    end_timestamp=1.0,
                    compacted_content=Content(parts=[Part(text='Prior')]),
                )
            ),
        ),
        self._create_event(2.0, 'New user input', 'user'),
    ]
    expected_formatted_history = 'model: Prior summary.\nuser: New user input'
    formatted_history = self.compactor._format_events_for_prompt(events)
    self.assertEqual(formatted_history, expected_formatted_history)

  def test_format_events_for_prompt_truncates_large_tool_response(self):
    limit = self.compactor._MAX_TOOL_CONTENT_CHARS
    large_value = 'x' * (limit + 500)
    events = [
        Event(
            timestamp=1.0,
            author='model',
            content=Content(
                parts=[
                    Part(
                        function_response=FunctionResponse(
                            id='call_1',
                            name='search',
                            response={'data': large_value},
                        )
                    )
                ]
            ),
        ),
    ]
    formatted_history = self.compactor._format_events_for_prompt(events)
    self.assertIn('Tool response from search:', formatted_history)
    self.assertIn('... [truncated', formatted_history)
    self.assertLess(len(formatted_history), len(large_value))