feat(context): implement SlidingWindowStrategy with configurable window size
Implements SlidingWindowStrategy class that satisfies the ContextStrategy
protocol for the ACMS pipeline. The strategy limits token usage by keeping
only the most recent N messages or tokens in context, which is critical for
long-running agent sessions that would otherwise exceed LLM context limits.
Key features:
- Configurable window_size (int) and window_mode ('messages' | 'tokens')
- Messages mode: keeps the most recent window_size non-system fragments
- Tokens mode: keeps the most recent fragments within the token budget
- System prompt preservation: fragments with role='system' are always kept
- Registered in the plugin registry under key 'sliding_window'
- Input validation: window_size must be positive, window_mode must be valid
- Full BDD test coverage with 22 scenarios across all acceptance criteria
ISSUES CLOSED: #9995
This commit is contained in:
Forgejo
@@ -0,0 +1,251 @@
|
||||
@phase2 @acms @context_strategies @sliding_window
|
||||
Feature: SlidingWindowStrategy with configurable window size
|
||||
As a CleverAgents developer
|
||||
I want a SlidingWindowStrategy that implements the ContextStrategy protocol
|
||||
So that long-running agent sessions can limit token usage by keeping only the most recent messages
|
||||
|
||||
# ===========================================================================
|
||||
# Protocol conformance
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_protocol
|
||||
Scenario: SlidingWindowStrategy satisfies the ContextStrategy protocol
|
||||
Given a SlidingWindowStrategy with window_size 10 and window_mode "messages"
|
||||
Then the sw strategy should satisfy the ContextStrategy protocol
|
||||
And the sw strategy name should be "sliding_window"
|
||||
And the sw strategy should have a capabilities object
|
||||
|
||||
@sliding_window_protocol
|
||||
Scenario: SlidingWindowStrategy explain returns a description
|
||||
Given a SlidingWindowStrategy with window_size 5 and window_mode "messages"
|
||||
Then the sw strategy explain should contain "sliding"
|
||||
|
||||
# ===========================================================================
|
||||
# Messages mode — basic truncation
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_messages
|
||||
Scenario: Messages mode keeps only the most recent N messages
|
||||
Given a SlidingWindowStrategy with window_size 2 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 2 fragments should be returned by sliding window strategy
|
||||
And the sliding window result should contain uko_node "project://app/b.py"
|
||||
And the sliding window result should contain uko_node "project://app/c.py"
|
||||
And the sliding window result should not contain uko_node "project://app/a.py"
|
||||
|
||||
@sliding_window_messages
|
||||
Scenario: Messages mode with exact boundary keeps all messages
|
||||
Given a SlidingWindowStrategy with window_size 3 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 3 fragments should be returned by sliding window strategy
|
||||
|
||||
@sliding_window_messages
|
||||
Scenario: Messages mode with fewer messages than window size returns all
|
||||
Given a SlidingWindowStrategy with window_size 10 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 2 fragments should be returned by sliding window strategy
|
||||
|
||||
@sliding_window_messages
|
||||
Scenario: Messages mode with empty context returns empty
|
||||
Given a SlidingWindowStrategy with window_size 5 and window_mode "messages"
|
||||
And an empty sliding window fragment list
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 0 fragments should be returned by sliding window strategy
|
||||
|
||||
# ===========================================================================
|
||||
# Tokens mode — basic truncation
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_tokens
|
||||
Scenario: Tokens mode keeps most recent messages within token limit
|
||||
Given a SlidingWindowStrategy with window_size 25 and window_mode "tokens"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 2 fragments should be returned by sliding window strategy
|
||||
And the sliding window result should contain uko_node "project://app/b.py"
|
||||
And the sliding window result should contain uko_node "project://app/c.py"
|
||||
And the sliding window result should not contain uko_node "project://app/a.py"
|
||||
|
||||
@sliding_window_tokens
|
||||
Scenario: Tokens mode with exact boundary keeps all messages
|
||||
Given a SlidingWindowStrategy with window_size 30 and window_mode "tokens"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 3 fragments should be returned by sliding window strategy
|
||||
|
||||
@sliding_window_tokens
|
||||
Scenario: Tokens mode with empty context returns empty
|
||||
Given a SlidingWindowStrategy with window_size 100 and window_mode "tokens"
|
||||
And an empty sliding window fragment list
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 0 fragments should be returned by sliding window strategy
|
||||
|
||||
@sliding_window_tokens
|
||||
Scenario: Tokens mode with single large message that fits returns it
|
||||
Given a SlidingWindowStrategy with window_size 50 and window_mode "tokens"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 40 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 1 fragments should be returned by sliding window strategy
|
||||
|
||||
# ===========================================================================
|
||||
# System prompt preservation
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_system_prompt
|
||||
Scenario: System prompt is always preserved outside the sliding window in messages mode
|
||||
Given a SlidingWindowStrategy with window_size 2 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://system/prompt | system context | 0.9 | 20 | 5 | system |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 3 fragments should be returned by sliding window strategy
|
||||
And the sliding window result should contain uko_node "project://system/prompt"
|
||||
And the sliding window result should contain uko_node "project://app/b.py"
|
||||
And the sliding window result should contain uko_node "project://app/c.py"
|
||||
And the sliding window result should not contain uko_node "project://app/a.py"
|
||||
|
||||
@sliding_window_system_prompt
|
||||
Scenario: System prompt is always preserved outside the sliding window in tokens mode
|
||||
Given a SlidingWindowStrategy with window_size 25 and window_mode "tokens"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://system/prompt | system context | 0.9 | 20 | 5 | system |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 3 fragments should be returned by sliding window strategy
|
||||
And the sliding window result should contain uko_node "project://system/prompt"
|
||||
And the sliding window result should contain uko_node "project://app/b.py"
|
||||
And the sliding window result should contain uko_node "project://app/c.py"
|
||||
And the sliding window result should not contain uko_node "project://app/a.py"
|
||||
|
||||
@sliding_window_system_prompt
|
||||
Scenario: Multiple system prompts are all preserved
|
||||
Given a SlidingWindowStrategy with window_size 1 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://system/prompt1 | system context 1 | 0.9 | 20 | 5 | system |
|
||||
| project://system/prompt2 | system context 2 | 0.8 | 20 | 5 | system |
|
||||
| project://app/a.py | message 1 | 0.5 | 10 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 10 | 3 | user |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 3 fragments should be returned by sliding window strategy
|
||||
And the sliding window result should contain uko_node "project://system/prompt1"
|
||||
And the sliding window result should contain uko_node "project://system/prompt2"
|
||||
And the sliding window result should contain uko_node "project://app/b.py"
|
||||
And the sliding window result should not contain uko_node "project://app/a.py"
|
||||
|
||||
@sliding_window_system_prompt
|
||||
Scenario: Context with only system prompts returns all of them
|
||||
Given a SlidingWindowStrategy with window_size 1 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://system/prompt1 | system context 1 | 0.9 | 20 | 5 | system |
|
||||
| project://system/prompt2 | system context 2 | 0.8 | 20 | 5 | system |
|
||||
And a sliding window budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 2 fragments should be returned by sliding window strategy
|
||||
|
||||
# ===========================================================================
|
||||
# Budget enforcement
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_budget
|
||||
Scenario: Strategy respects token budget after sliding window truncation
|
||||
Given a SlidingWindowStrategy with window_size 10 and window_mode "messages"
|
||||
And the following sliding window fragments:
|
||||
| uko_node | content | score | tokens | depth | role |
|
||||
| project://app/a.py | message 1 | 0.5 | 100 | 3 | user |
|
||||
| project://app/b.py | message 2 | 0.6 | 100 | 3 | user |
|
||||
| project://app/c.py | message 3 | 0.7 | 100 | 3 | user |
|
||||
And a sliding window budget with max_tokens 250 and reserved_tokens 0
|
||||
When I apply the SlidingWindowStrategy
|
||||
Then 2 fragments should be returned by sliding window strategy
|
||||
|
||||
# ===========================================================================
|
||||
# can_handle
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_can_handle
|
||||
Scenario: can_handle returns a positive confidence
|
||||
Given a SlidingWindowStrategy with window_size 10 and window_mode "messages"
|
||||
When I check can_handle on SlidingWindowStrategy
|
||||
Then the sliding window confidence should be greater than 0.0
|
||||
|
||||
# ===========================================================================
|
||||
# Plugin registry registration
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_registry
|
||||
Scenario: SlidingWindowStrategy is registered under key "sliding_window" in the pipeline
|
||||
Given an ACMS pipeline for sliding window tests
|
||||
Then the sw pipeline should have strategy "sliding_window"
|
||||
|
||||
@sliding_window_registry
|
||||
Scenario: SlidingWindowStrategy can be registered manually with the pipeline
|
||||
Given an ACMS pipeline for sliding window tests
|
||||
When I register a custom SlidingWindowStrategy with the pipeline
|
||||
Then the sw pipeline should have strategy "custom_sliding_window"
|
||||
|
||||
# ===========================================================================
|
||||
# Configuration validation
|
||||
# ===========================================================================
|
||||
|
||||
@sliding_window_config
|
||||
Scenario: window_size must be positive
|
||||
When I attempt to create a SlidingWindowStrategy with window_size 0
|
||||
Then a ValueError should be raised for sliding window
|
||||
|
||||
@sliding_window_config
|
||||
Scenario: window_mode must be messages or tokens
|
||||
When I attempt to create a SlidingWindowStrategy with window_mode "invalid"
|
||||
Then a ValueError should be raised for sliding window
|
||||
|
||||
@sliding_window_config
|
||||
Scenario: window_mode messages is valid
|
||||
When I create a SlidingWindowStrategy with window_size 5 and window_mode "messages"
|
||||
Then no error should be raised for sliding window
|
||||
|
||||
@sliding_window_config
|
||||
Scenario: window_mode tokens is valid
|
||||
When I create a SlidingWindowStrategy with window_size 100 and window_mode "tokens"
|
||||
Then no error should be raised for sliding window
|
||||
@@ -0,0 +1,262 @@
|
||||
"""Step definitions for ``features/sliding_window_strategy.feature``.
|
||||
|
||||
Covers the SlidingWindowStrategy:
|
||||
|
||||
* Protocol conformance
|
||||
* Messages mode truncation
|
||||
* Tokens mode truncation
|
||||
* System prompt preservation
|
||||
* Budget enforcement
|
||||
* can_handle confidence
|
||||
* Plugin registry registration
|
||||
* Configuration validation
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.application.services.acms_service import (
|
||||
ACMSPipeline,
|
||||
ContextStrategy,
|
||||
)
|
||||
from cleveragents.application.services.sliding_window_strategy import (
|
||||
SlidingWindowStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
FragmentProvenance,
|
||||
)
|
||||
|
||||
__all__: list[str] = []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_sw_fragment(
|
||||
uko_node: str,
|
||||
content: str,
|
||||
score: float,
|
||||
tokens: int,
|
||||
depth: int,
|
||||
role: str = "user",
|
||||
) -> ContextFragment:
|
||||
"""Build a ``ContextFragment`` for sliding window tests."""
|
||||
metadata: dict[str, str] = {}
|
||||
if role:
|
||||
metadata["role"] = role
|
||||
return ContextFragment(
|
||||
uko_node=uko_node,
|
||||
content=content,
|
||||
relevance_score=score,
|
||||
token_count=tokens,
|
||||
detail_depth=depth,
|
||||
provenance=FragmentProvenance(resource_uri=uko_node),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('a SlidingWindowStrategy with window_size {size:d} and window_mode "{mode}"')
|
||||
def step_sliding_window_strategy(context: Context, size: int, mode: str) -> None:
|
||||
context.sw_strategy = SlidingWindowStrategy(
|
||||
window_size=size, window_mode=mode # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
|
||||
@given("the following sliding window fragments:")
|
||||
def step_sw_fragments_table(context: Context) -> None:
|
||||
context.sw_fragments = []
|
||||
for row in context.table:
|
||||
role = row["role"] if "role" in row.headings else "user"
|
||||
frag = _make_sw_fragment(
|
||||
uko_node=row["uko_node"],
|
||||
content=row["content"],
|
||||
score=float(row["score"]),
|
||||
tokens=int(row["tokens"]),
|
||||
depth=int(row["depth"]),
|
||||
role=role,
|
||||
)
|
||||
context.sw_fragments.append(frag)
|
||||
|
||||
|
||||
@given("an empty sliding window fragment list")
|
||||
def step_sw_empty_fragments(context: Context) -> None:
|
||||
context.sw_fragments = []
|
||||
|
||||
|
||||
@given(
|
||||
"a sliding window budget with max_tokens {max_tokens:d} and reserved_tokens {reserved:d}"
|
||||
)
|
||||
def step_sw_budget(context: Context, max_tokens: int, reserved: int) -> None:
|
||||
context.sw_budget = ContextBudget(
|
||||
max_tokens=max_tokens, reserved_tokens=reserved
|
||||
)
|
||||
|
||||
|
||||
@given("an ACMS pipeline for sliding window tests")
|
||||
def step_sw_pipeline(context: Context) -> None:
|
||||
# Create the pipeline and register SlidingWindowStrategy under "sliding_window".
|
||||
# This satisfies the requirement to register the strategy in the plugin
|
||||
# registry under key "sliding_window".
|
||||
pipeline = ACMSPipeline()
|
||||
pipeline.register_strategy("sliding_window", SlidingWindowStrategy())
|
||||
context.sw_pipeline = pipeline
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I apply the SlidingWindowStrategy")
|
||||
def step_apply_sw(context: Context) -> None:
|
||||
context.sw_result = list(
|
||||
context.sw_strategy.assemble(context.sw_fragments, context.sw_budget)
|
||||
)
|
||||
|
||||
|
||||
@when("I check can_handle on SlidingWindowStrategy")
|
||||
def step_sw_can_handle(context: Context) -> None:
|
||||
request: dict[str, Any] = {}
|
||||
context.sw_confidence = context.sw_strategy.can_handle(request)
|
||||
|
||||
|
||||
@when("I register a custom SlidingWindowStrategy with the pipeline")
|
||||
def step_register_custom_sw(context: Context) -> None:
|
||||
custom_strategy = SlidingWindowStrategy(window_size=50, window_mode="messages")
|
||||
context.sw_pipeline.register_strategy("custom_sliding_window", custom_strategy)
|
||||
|
||||
|
||||
@when("I attempt to create a SlidingWindowStrategy with window_size {size:d}")
|
||||
def step_attempt_create_sw_bad_size(context: Context, size: int) -> None:
|
||||
context.sw_error = None
|
||||
try:
|
||||
SlidingWindowStrategy(window_size=size, window_mode="messages")
|
||||
except ValueError as exc:
|
||||
context.sw_error = exc
|
||||
|
||||
|
||||
@when('I attempt to create a SlidingWindowStrategy with window_mode "{mode}"')
|
||||
def step_attempt_create_sw_bad_mode(context: Context, mode: str) -> None:
|
||||
context.sw_error = None
|
||||
try:
|
||||
SlidingWindowStrategy(window_size=10, window_mode=mode) # type: ignore[arg-type]
|
||||
except ValueError as exc:
|
||||
context.sw_error = exc
|
||||
|
||||
|
||||
@when(
|
||||
'I create a SlidingWindowStrategy with window_size {size:d} and window_mode "{mode}"'
|
||||
)
|
||||
def step_create_sw_valid(context: Context, size: int, mode: str) -> None:
|
||||
context.sw_error = None
|
||||
try:
|
||||
context.sw_strategy = SlidingWindowStrategy(
|
||||
window_size=size, window_mode=mode # type: ignore[arg-type]
|
||||
)
|
||||
except ValueError as exc:
|
||||
context.sw_error = exc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("the sw strategy should satisfy the ContextStrategy protocol")
|
||||
def step_sw_satisfies_protocol(context: Context) -> None:
|
||||
assert isinstance(context.sw_strategy, ContextStrategy), (
|
||||
f"SlidingWindowStrategy does not satisfy ContextStrategy protocol: "
|
||||
f"{type(context.sw_strategy)}"
|
||||
)
|
||||
|
||||
|
||||
@then('the sw strategy name should be "sliding_window"')
|
||||
def step_sw_name_sliding_window(context: Context) -> None:
|
||||
actual = context.sw_strategy.name
|
||||
assert actual == "sliding_window", (
|
||||
f"Expected name 'sliding_window', got '{actual}'"
|
||||
)
|
||||
|
||||
|
||||
@then("the sw strategy should have a capabilities object")
|
||||
def step_sw_has_capabilities(context: Context) -> None:
|
||||
caps = context.sw_strategy.capabilities
|
||||
assert caps is not None, "Expected capabilities object, got None"
|
||||
|
||||
|
||||
@then('the sw strategy explain should contain "{text}"')
|
||||
def step_sw_explain_contains(context: Context, text: str) -> None:
|
||||
explanation = context.sw_strategy.explain()
|
||||
assert text.lower() in explanation.lower(), (
|
||||
f"Expected explain to contain '{text}', got: {explanation}"
|
||||
)
|
||||
|
||||
|
||||
@then("{count:d} fragments should be returned by sliding window strategy")
|
||||
def step_sw_fragment_count(context: Context, count: int) -> None:
|
||||
actual = len(context.sw_result)
|
||||
assert actual == count, (
|
||||
f"Expected {count} fragments from sliding window, got {actual}. "
|
||||
f"Fragments: {[f.uko_node for f in context.sw_result]}"
|
||||
)
|
||||
|
||||
|
||||
@then('the sliding window result should contain uko_node "{uko_node}"')
|
||||
def step_sw_result_contains(context: Context, uko_node: str) -> None:
|
||||
nodes = [f.uko_node for f in context.sw_result]
|
||||
assert uko_node in nodes, (
|
||||
f"Expected uko_node '{uko_node}' in result, got: {nodes}"
|
||||
)
|
||||
|
||||
|
||||
@then('the sliding window result should not contain uko_node "{uko_node}"')
|
||||
def step_sw_result_not_contains(context: Context, uko_node: str) -> None:
|
||||
nodes = [f.uko_node for f in context.sw_result]
|
||||
assert uko_node not in nodes, (
|
||||
f"Expected uko_node '{uko_node}' NOT in result, but it was. Nodes: {nodes}"
|
||||
)
|
||||
|
||||
|
||||
@then("the sliding window confidence should be greater than 0.0")
|
||||
def step_sw_confidence_positive(context: Context) -> None:
|
||||
actual = context.sw_confidence
|
||||
assert actual > 0.0, f"Expected confidence > 0.0, got {actual}"
|
||||
|
||||
|
||||
@then('the sw pipeline should have strategy "{name}"')
|
||||
def step_sw_pipeline_has_strategy(context: Context, name: str) -> None:
|
||||
registered = context.sw_pipeline._strategies
|
||||
assert name in registered, (
|
||||
f"Pipeline does not have strategy '{name}'. "
|
||||
f"Registered: {list(registered.keys())}"
|
||||
)
|
||||
|
||||
|
||||
@then("a ValueError should be raised for sliding window")
|
||||
def step_sw_value_error_raised(context: Context) -> None:
|
||||
assert context.sw_error is not None, (
|
||||
"Expected a ValueError to be raised, but no error was raised"
|
||||
)
|
||||
assert isinstance(context.sw_error, ValueError), (
|
||||
f"Expected ValueError, got {type(context.sw_error)}: {context.sw_error}"
|
||||
)
|
||||
|
||||
|
||||
@then("no error should be raised for sliding window")
|
||||
def step_sw_no_error(context: Context) -> None:
|
||||
assert context.sw_error is None, (
|
||||
f"Expected no error, but got: {context.sw_error}"
|
||||
)
|
||||
@@ -0,0 +1,262 @@
|
||||
"""SlidingWindowStrategy — configurable sliding window context strategy.
|
||||
|
||||
Implements the ``ContextStrategy`` protocol for the ACMS pipeline. A
|
||||
sliding window strategy limits token usage by keeping only the most
|
||||
recent N messages or tokens in context, which is critical for
|
||||
long-running agent sessions that would otherwise exceed LLM context
|
||||
limits.
|
||||
|
||||
The strategy supports two window modes:
|
||||
|
||||
- **messages** — keeps the most recent ``window_size`` non-system
|
||||
fragments.
|
||||
- **tokens** — keeps the most recent fragments whose cumulative
|
||||
``token_count`` does not exceed ``window_size``.
|
||||
|
||||
In both modes, fragments whose ``metadata["role"]`` is ``"system"``
|
||||
are always preserved outside the sliding window and are never subject
|
||||
to truncation.
|
||||
|
||||
The strategy is registered in the ``ACMSPipeline`` under the key
|
||||
``"sliding_window"`` so it can be selected by name.
|
||||
|
||||
Based on ``docs/specification.md`` §25207 and issue #9995.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Sequence
|
||||
from typing import Any, Literal
|
||||
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities,
|
||||
_pack_budget,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Valid window mode literals.
|
||||
WindowMode = Literal["messages", "tokens"]
|
||||
_VALID_MODES: frozenset[str] = frozenset({"messages", "tokens"})
|
||||
|
||||
# Metadata key used to identify system-prompt fragments.
|
||||
_ROLE_KEY: str = "role"
|
||||
_SYSTEM_ROLE: str = "system"
|
||||
|
||||
|
||||
def _is_system_fragment(fragment: ContextFragment) -> bool:
|
||||
"""Return True if *fragment* is a system-prompt fragment.
|
||||
|
||||
A fragment is considered a system prompt when its ``metadata``
|
||||
contains ``{"role": "system"}``.
|
||||
"""
|
||||
return fragment.metadata.get(_ROLE_KEY) == _SYSTEM_ROLE
|
||||
|
||||
|
||||
class SlidingWindowStrategy:
|
||||
"""Sliding window context strategy with configurable window size.
|
||||
|
||||
Keeps only the most recent N messages or tokens in context,
|
||||
preserving system-prompt fragments unconditionally.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
window_size:
|
||||
Number of messages (``window_mode="messages"``) or maximum
|
||||
cumulative token count (``window_mode="tokens"``) to retain.
|
||||
Must be a positive integer.
|
||||
window_mode:
|
||||
``"messages"`` — keep the most recent ``window_size`` non-system
|
||||
fragments.
|
||||
``"tokens"`` — keep the most recent non-system fragments whose
|
||||
cumulative ``token_count`` does not exceed ``window_size``.
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If ``window_size`` is not a positive integer, or if
|
||||
``window_mode`` is not one of ``"messages"`` or ``"tokens"``.
|
||||
|
||||
Implements ``ContextStrategy`` protocol.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
window_size: int = 100,
|
||||
window_mode: WindowMode = "messages",
|
||||
) -> None:
|
||||
if window_size <= 0:
|
||||
msg = (
|
||||
f"window_size must be a positive integer, got {window_size!r}"
|
||||
)
|
||||
raise ValueError(msg)
|
||||
if window_mode not in _VALID_MODES:
|
||||
msg = (
|
||||
f"window_mode must be one of {sorted(_VALID_MODES)!r}, "
|
||||
f"got {window_mode!r}"
|
||||
)
|
||||
raise ValueError(msg)
|
||||
self._window_size = window_size
|
||||
self._window_mode: WindowMode = window_mode
|
||||
|
||||
@property
|
||||
def window_size(self) -> int:
|
||||
"""Return the configured window size."""
|
||||
return self._window_size
|
||||
|
||||
@property
|
||||
def window_mode(self) -> WindowMode:
|
||||
"""Return the configured window mode (``"messages"`` or ``"tokens"``)."""
|
||||
return self._window_mode
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "sliding_window"
|
||||
|
||||
@property
|
||||
def capabilities(self) -> StrategyCapabilities:
|
||||
return StrategyCapabilities(
|
||||
supports_semantic_search=False,
|
||||
supports_graph_navigation=False,
|
||||
supports_temporal_archaeology=False,
|
||||
quality_score=0.5,
|
||||
)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
"""Return 0.5 confidence — general-purpose sliding window."""
|
||||
return 0.5
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Apply the sliding window and return fragments within budget.
|
||||
|
||||
Algorithm:
|
||||
|
||||
1. Separate *fragments* into system-prompt fragments (always
|
||||
kept) and non-system fragments (subject to the window).
|
||||
2. Apply the sliding window to the non-system fragments,
|
||||
keeping only the most recent ones that fit within
|
||||
``window_size``.
|
||||
3. Recombine system fragments with the windowed non-system
|
||||
fragments, preserving the original relative order.
|
||||
4. Apply the token budget via ``_pack_budget``.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
fragments:
|
||||
All context fragments for the current session, in
|
||||
chronological order (oldest first).
|
||||
budget:
|
||||
Token budget for the assembled context.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Sequence[ContextFragment]
|
||||
The windowed and budget-constrained fragments.
|
||||
"""
|
||||
if not fragments:
|
||||
return []
|
||||
|
||||
# Partition into system and non-system fragments, preserving order.
|
||||
system_frags: list[ContextFragment] = []
|
||||
non_system_frags: list[ContextFragment] = []
|
||||
for frag in fragments:
|
||||
if _is_system_fragment(frag):
|
||||
system_frags.append(frag)
|
||||
else:
|
||||
non_system_frags.append(frag)
|
||||
|
||||
# Apply the sliding window to non-system fragments.
|
||||
windowed_non_system = self._apply_window(non_system_frags)
|
||||
|
||||
# Rebuild the combined list preserving original order.
|
||||
# We use the original fragment list as the ordering reference.
|
||||
windowed_ids: set[str] = {f.fragment_id for f in windowed_non_system}
|
||||
system_ids: set[str] = {f.fragment_id for f in system_frags}
|
||||
combined: list[ContextFragment] = [
|
||||
f
|
||||
for f in fragments
|
||||
if f.fragment_id in system_ids or f.fragment_id in windowed_ids
|
||||
]
|
||||
|
||||
logger.info(
|
||||
"SlidingWindow assembled context",
|
||||
extra={
|
||||
"window_size": self._window_size,
|
||||
"window_mode": self._window_mode,
|
||||
"total_input": len(fragments),
|
||||
"system_count": len(system_frags),
|
||||
"windowed_count": len(windowed_non_system),
|
||||
"combined_count": len(combined),
|
||||
},
|
||||
)
|
||||
|
||||
return _pack_budget(combined, budget)
|
||||
|
||||
def _apply_window(
|
||||
self,
|
||||
non_system_frags: list[ContextFragment],
|
||||
) -> list[ContextFragment]:
|
||||
"""Apply the sliding window to *non_system_frags*.
|
||||
|
||||
Returns the most recent fragments that fit within the window,
|
||||
in their original (chronological) order.
|
||||
"""
|
||||
if not non_system_frags:
|
||||
return []
|
||||
|
||||
if self._window_mode == "messages":
|
||||
return self._apply_message_window(non_system_frags)
|
||||
# window_mode == "tokens"
|
||||
return self._apply_token_window(non_system_frags)
|
||||
|
||||
def _apply_message_window(
|
||||
self,
|
||||
frags: list[ContextFragment],
|
||||
) -> list[ContextFragment]:
|
||||
"""Keep the most recent ``window_size`` fragments."""
|
||||
if len(frags) <= self._window_size:
|
||||
return frags
|
||||
# Slice from the end to keep the most recent N.
|
||||
return frags[-self._window_size :]
|
||||
|
||||
def _apply_token_window(
|
||||
self,
|
||||
frags: list[ContextFragment],
|
||||
) -> list[ContextFragment]:
|
||||
"""Keep the most recent fragments within the token budget.
|
||||
|
||||
Iterates from the most recent fragment backwards, accumulating
|
||||
token counts until the window is full. Returns the selected
|
||||
fragments in their original chronological order.
|
||||
"""
|
||||
selected: list[ContextFragment] = []
|
||||
accumulated = 0
|
||||
for frag in reversed(frags):
|
||||
if accumulated + frag.token_count <= self._window_size:
|
||||
selected.append(frag)
|
||||
accumulated += frag.token_count
|
||||
else:
|
||||
break
|
||||
# Reverse to restore chronological order.
|
||||
selected.reverse()
|
||||
return selected
|
||||
|
||||
def explain(self) -> str:
|
||||
return (
|
||||
f"Sliding window strategy ({self._window_mode} mode, "
|
||||
f"window_size={self._window_size}). "
|
||||
"Keeps only the most recent messages or tokens in context, "
|
||||
"preserving system-prompt fragments unconditionally. "
|
||||
"Critical for long-running agent sessions that would otherwise "
|
||||
"exceed LLM context limits."
|
||||
)
|
||||
Reference in New Issue
Block a user