feat(agent): add support for history window in current turn contents

Author: Sourabh Patil

src/google/adk/agents/llm_agent.py

@@ -169,6 +169,15 @@ class LlmAgent(BaseAgent):
             instruction and input
   """
 
+  history_window_size: Optional[int] = None
+  """The size of the history window (N) to use.
+
+  This parameter is ONLY used when `include_contents` is set to 'windowed'.
+  It must be a positive integer defining the number of recent conversational
+  turns to include in the model context. 
+  """
+
+
   # Controlled input/output configurations - Start
   input_schema: Optional[type[BaseModel]] = None
   """The input schema when agent is used as a tool."""
@@ -268,6 +277,31 @@ class LlmAgent(BaseAgent):
   """
   # Callbacks - End
 
+  @model_validator(mode='after')
+  def validate_history_window(self) -> 'LlmAgent':
+    """Validate that history_window_size is used correctly."""
+    is_windowed = self.include_contents == 'windowed'
+    window_size_is_set = self.history_window_size is not None
+
+    # Case 1: 'windowed' is set, but the window size is not.
+    if is_windowed and not window_size_is_set:
+      raise ValueError(
+          "When 'include_contents' is 'windowed', 'history_window_size' must be set to a positive integer."
+      )
+
+    # Case 2: 'windowed' is set, but window size is not a positive integer.
+    if is_windowed and window_size_is_set and self.history_window_size <= 0:
+      raise ValueError("'history_window_size' must be a positive integer.")
+
+    # Case 3: 'windowed' is NOT set, but the window size is.
+    if not is_windowed and window_size_is_set:
+      raise ValueError(
+          "'history_window_size' can only be set when 'include_contents' is 'windowed'."
+      )
+
+    return self
+
+
   @override
   async def _run_async_impl(
       self, ctx: InvocationContext

src/google/adk/flows/llm_flows/contents.py

@@ -50,6 +50,14 @@ async def run_async(
           invocation_context.session.events,
           agent.name,
       )
+    elif agent.include_contents == 'windowed':
+      llm_request.contents = _get_windowed_turn_contents(
+      ## Includes windowed turn contexts (windowed conversation history)
+      invocation_context.branch,
+      invocation_context.session.events,
+      agent.name,
+      history_window=agent.history_window_size
+      )
     else:
       # Include current turn context only (no conversation history)
       llm_request.contents = _get_current_turn_contents(
@@ -282,6 +290,39 @@ def _get_current_turn_contents(
 
   return []
 
+def _get_windowed_turn_contents(
+    current_branch: Optional[str],
+    events: list[Event],
+    agent_name: str = '',
+    history_window: int = 5  # New argument
+) -> list[types.Content]:
+    """Get contents for the current turn, optionally with limited conversation history.
+
+    Args:
+      current_branch: The current branch of the agent.
+      events: A list of all session events.
+      agent_name: The name of the agent.
+      history_window: Number of previous contents to include from conversation history.
+
+    Returns:
+      A list of contents for the current turn and optional recent history.
+    """
+    # Find the start of the current turn
+    turn_start_index = 0
+    for i in range(len(events) - 1, -1, -1):
+        event = events[i]
+        if event.author == 'user' or _is_other_agent_reply(agent_name, event):
+            turn_start_index = i
+            break
+
+    # Determine the starting index for history window
+    if history_window > 0:
+        history_start_index = max(0, turn_start_index - history_window)
+    else:
+        history_start_index = turn_start_index
+
+    return _get_contents(current_branch, events[history_start_index:], agent_name)
+
 
 def _is_other_agent_reply(current_agent_name: str, event: Event) -> bool:
   """Whether the event is a reply from another agent."""