From d9a0a936677cbaf296b59d27d341d20806e5575f Mon Sep 17 00:00:00 2001
From: Mark Backman <mark@daily.co>
Date: Wed, 30 Apr 2025 09:55:34 -0400
Subject: [PATCH] Add context_window_compression support to
 GeminiMultimodalLiveLLMService

---
 CHANGELOG.md                                  |  5 ++++
 .../services/gemini_multimodal_live/events.py |  7 +++++
 .../services/gemini_multimodal_live/gemini.py | 28 +++++++++++++++++++
 3 files changed, 40 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1e5bb8849..08e9d30ed 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added a `context_window_compression` InputParam to
+  `GeminiMultimodalLiveLLMService` which allows you to enable a sliding
+  context window for the session as well as set the token limit of the sliding
+  window.
+
 - Updated `SmallWebRTCConnection` to support `ice_servers` with credentials.
 
 - Added `VADUserStartedSpeakingFrame` and `VADUserStoppedSpeakingFrame`,
diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py
index 22cc7b5f4..f621b41fd 100644
--- a/src/pipecat/services/gemini_multimodal_live/events.py
+++ b/src/pipecat/services/gemini_multimodal_live/events.py
@@ -193,3 +193,10 @@ def parse_server_event(str):
     except Exception as e:
         print(f"Error parsing server event: {e}")
         return None
+
+
+class ContextWindowCompressionConfig(BaseModel):
+    """Configuration for context window compression."""
+
+    sliding_window: Optional[bool] = Field(default=True)
+    trigger_tokens: Optional[int] = Field(default=None)
diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py
index d953e5065..3881f7c7e 100644
--- a/src/pipecat/services/gemini_multimodal_live/gemini.py
+++ b/src/pipecat/services/gemini_multimodal_live/gemini.py
@@ -265,6 +265,15 @@ class GeminiVADParams(BaseModel):
     silence_duration_ms: Optional[int] = Field(default=None)
 
 
+class ContextWindowCompressionParams(BaseModel):
+    """Parameters for context window compression."""
+
+    enabled: bool = Field(default=False)
+    trigger_tokens: Optional[int] = Field(
+        default=None
+    )  # None = use default (80% of context window)
+
+
 class InputParams(BaseModel):
     frequency_penalty: Optional[float] = Field(default=None, ge=0.0, le=2.0)
     max_tokens: Optional[int] = Field(default=4096, ge=1)
@@ -280,6 +289,7 @@ class InputParams(BaseModel):
         default=GeminiMediaResolution.UNSPECIFIED
     )
     vad: Optional[GeminiVADParams] = Field(default=None)
+    context_window_compression: Optional[ContextWindowCompressionParams] = Field(default=None)
     extra: Optional[Dict[str, Any]] = Field(default_factory=dict)
 
 
@@ -355,6 +365,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
             "language": self._language_code,
             "media_resolution": params.media_resolution,
             "vad": params.vad,
+            "context_window_compression": params.context_window_compression.model_dump()
+            if params.context_window_compression
+            else None,
             "extra": params.extra if isinstance(params.extra, dict) else {},
         }
 
@@ -561,6 +574,21 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 }
             }
 
+            # Add context window compression if enabled
+            if self._settings.get("context_window_compression", {}).get("enabled", False):
+                compression_config = {}
+                # Add sliding window (always true if compression is enabled)
+                compression_config["sliding_window"] = {}
+
+                # Add trigger_tokens if specified
+                trigger_tokens = self._settings.get("context_window_compression", {}).get(
+                    "trigger_tokens"
+                )
+                if trigger_tokens is not None:
+                    compression_config["trigger_tokens"] = trigger_tokens
+
+                config_data["setup"]["context_window_compression"] = compression_config
+
             # Add VAD configuration if provided
             if self._settings.get("vad"):
                 vad_config = {}