From 7be0ee56aacc45138799b096e958e69b824ea9cc Mon Sep 17 00:00:00 2001 From: QueryPlanner Date: Thu, 20 Aug 2026 19:54:40 +0530 Subject: [PATCH] feat: add automatic context compaction - Configure ADK token-threshold event compaction at 200000 tokens - Verify session-preserving SQLite compaction persistence - Document retained event context and reset behavior --- docs/architecture.md | 7 +++ src/blacki/agent.py | 14 ++++- tests/test_adk_compaction.py | 114 +++++++++++++++++++++++++++++++++++ tests/test_integration.py | 16 +++++ 4 files changed, 150 insertions(+), 1 deletion(-) create mode 100644 tests/test_adk_compaction.py diff --git a/docs/architecture.md b/docs/architecture.md index 4c9902d..8241ece 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -35,6 +35,13 @@ the ADK application plugins. tool is enabled and selected, it synthesizes a bounded MP3 in memory and sends it directly to the same chat or topic through Telegram `sendAudio`. +Long conversations use Google ADK's native token-based event compaction. After +the latest prompt reaches 200,000 tokens, ADK summarizes older conversation +events after the successful turn and retains the latest eight raw events for +immediate context. This is prompt compaction, not a hard session reset: the +same versioned Telegram session continues, the compaction event is persisted, +and `/reset` remains available when a completely new conversation is desired. + For a private chat with the optional Google Health connector, `/connect_health` creates a short-lived one-time OAuth state and sends a Google authorization URL. The HTTPS callback consumes the state, exchanges the code, resolves Google's diff --git a/src/blacki/agent.py b/src/blacki/agent.py index d86720a..762b7e2 100644 --- a/src/blacki/agent.py +++ b/src/blacki/agent.py @@ -10,6 +10,7 @@ from dotenv import load_dotenv from google.adk.agents import BaseAgent, LlmAgent from google.adk.apps import App +from google.adk.apps.app import EventsCompactionConfig from google.adk.plugins.base_plugin import BasePlugin from google.adk.plugins.global_instruction_plugin import GlobalInstructionPlugin @@ -52,6 +53,12 @@ logging_callbacks = LoggingCallbacks() TASK_WORKER_NAME = "task_worker" TASK_WORKER_ENABLED_VALUES = frozenset({"1", "true", "yes"}) +AUTO_COMPACTION_TOKEN_THRESHOLD = 200_000 +AUTO_COMPACTION_EVENT_RETENTION_SIZE = 8 +# ADK 2.5.0 requires the sliding-window fields even for token-only +# configuration. Keep the interval above any realistic session length so the +# token threshold remains the only practical compaction trigger. +_AUTO_COMPACTION_INTERVAL_SENTINEL = 1_000_000_000 class TelegramModelOverridePlugin(BasePlugin): @@ -294,7 +301,12 @@ def create_app(agent: LlmAgent | None = None) -> App: name="blacki", root_agent=agent, plugins=plugins, - events_compaction_config=None, + events_compaction_config=EventsCompactionConfig( + compaction_interval=_AUTO_COMPACTION_INTERVAL_SENTINEL, + overlap_size=0, + token_threshold=AUTO_COMPACTION_TOKEN_THRESHOLD, + event_retention_size=AUTO_COMPACTION_EVENT_RETENTION_SIZE, + ), context_cache_config=None, resumability_config=None, ) diff --git a/tests/test_adk_compaction.py b/tests/test_adk_compaction.py new file mode 100644 index 0000000..a1ce6eb --- /dev/null +++ b/tests/test_adk_compaction.py @@ -0,0 +1,114 @@ +"""Tests for Blacki's native Google ADK context compaction configuration.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from google.adk.apps.base_events_summarizer import BaseEventsSummarizer +from google.adk.apps.compaction import _run_compaction_for_token_threshold_config +from google.adk.events import Event, EventActions +from google.adk.events.event_actions import EventCompaction +from google.adk.sessions.database_session_service import DatabaseSessionService +from google.genai import types + +from blacki import app + + +class RecordingSummarizer(BaseEventsSummarizer): + """Return a deterministic summary while exercising ADK compaction.""" + + def __init__(self) -> None: + self.batches: list[list[Event]] = [] + + async def maybe_summarize_events(self, *, events: list[Event]) -> Event | None: + self.batches.append(events) + return Event( + author="user", + invocation_id="compaction-invocation", + actions=EventActions( + compaction=EventCompaction( + start_timestamp=events[0].timestamp, + end_timestamp=events[-1].timestamp, + compacted_content=types.Content( + role="model", + parts=[types.Part.from_text(text="Compacted history")], + ), + ) + ), + ) + + +def _conversation_event(index: int) -> Event: + return Event( + author="blacki", + invocation_id=f"invocation-{index}", + timestamp=float(index + 1), + content=types.Content( + role="model", + parts=[types.Part.from_text(text=f"Conversation event {index}")], + ), + usage_metadata=types.GenerateContentResponseUsageMetadata( + prompt_token_count=200_000 + ), + ) + + +@pytest.mark.asyncio +async def test_token_compaction_is_persisted_in_sqlite(tmp_path: Path) -> None: + """Compact older events at the threshold without creating a new session.""" + config = app.events_compaction_config + assert config is not None + retention_size = config.event_retention_size + assert retention_size is not None + summarizer = RecordingSummarizer() + config = config.model_copy(update={"summarizer": summarizer}) + + service = DatabaseSessionService(f"sqlite+aiosqlite:///{tmp_path / 'sessions.db'}") + try: + session = await service.create_session( + app_name=app.name, + user_id="compaction-test-user", + session_id="compaction-session-v1", + ) + for index in range(retention_size + 1): + await service.append_event( + session=session, + event=_conversation_event(index), + ) + + hydrated_session = await service.get_session( + app_name=app.name, + user_id="compaction-test-user", + session_id="compaction-session-v1", + ) + assert hydrated_session is not None + assert app.root_agent is not None + + compacted = await _run_compaction_for_token_threshold_config( + config=config, + session=hydrated_session, + session_service=service, + agent=app.root_agent, + agent_name="blacki", + ) + + assert compacted is True + assert len(summarizer.batches) == 1 + assert len(summarizer.batches[0]) == 1 + + persisted_session = await service.get_session( + app_name=app.name, + user_id="compaction-test-user", + session_id="compaction-session-v1", + ) + assert persisted_session is not None + assert persisted_session.id == "compaction-session-v1" + assert persisted_session.events[-1].actions.compaction is not None + summary_parts = persisted_session.events[ + -1 + ].actions.compaction.compacted_content.parts + assert summary_parts is not None + assert summary_parts[0].text == "Compacted history" + finally: + await service.close() diff --git a/tests/test_integration.py b/tests/test_integration.py index 8f02e9e..578c303 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -11,7 +11,13 @@ from collections.abc import Sequence from typing import Any, Protocol, cast +from google.adk.apps.app import EventsCompactionConfig + from blacki import app +from blacki.agent import ( + AUTO_COMPACTION_EVENT_RETENTION_SIZE, + AUTO_COMPACTION_TOKEN_THRESHOLD, +) class AgentConfigLike(Protocol): @@ -43,6 +49,16 @@ def test_app_has_root_agent(self) -> None: """Verify app is wired to root agent.""" assert app.root_agent is not None + def test_app_uses_token_based_context_compaction(self) -> None: + """Keep long conversations bounded without replacing their sessions.""" + config = app.events_compaction_config + + assert isinstance(config, EventsCompactionConfig) + assert config.token_threshold == AUTO_COMPACTION_TOKEN_THRESHOLD + assert config.event_retention_size == AUTO_COMPACTION_EVENT_RETENTION_SIZE + assert config.overlap_size == 0 + assert config.compaction_interval > config.token_threshold + def test_app_plugins_are_valid_if_configured(self) -> None: """Verify plugins (if any) are properly initialized.""" # Plugins are optional - if configured, they should be a list