[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot] 2026-05-22 14:47:06 +00:00 committed by Daniel Han
commit 75e32f20f7
2 changed files with 52 additions and 42 deletions

View file

@ -117,9 +117,7 @@ class CompactStrategy(ABC):
"""Interface for context-compaction strategies."""
@abstractmethod
def compact(
self, messages: list[dict], budget_tokens: int
) -> list[dict]:
def compact(self, messages: list[dict], budget_tokens: int) -> list[dict]:
"""Return a (possibly shorter) list of messages within
``budget_tokens``. Returns ``messages`` unchanged when no
compaction is needed or possible.
@ -130,9 +128,7 @@ class CompactStrategy(ABC):
class NoCompact(CompactStrategy):
"""Passthrough strategy. Returns ``messages`` unchanged."""
def compact(
self, messages: list[dict], budget_tokens: int
) -> list[dict]:
def compact(self, messages: list[dict], budget_tokens: int) -> list[dict]:
return list(messages)
@ -147,9 +143,7 @@ class SlidingWindowCompact(CompactStrategy):
within ``budget_tokens`` or when there is nothing left to drop.
"""
def __init__(
self, keep_recent: int = 2, compact_threshold: float = 0.85
) -> None:
def __init__(self, keep_recent: int = 2, compact_threshold: float = 0.85) -> None:
if keep_recent < 0:
raise ValueError("keep_recent must be >= 0")
if not (0.0 < compact_threshold <= 1.0):
@ -157,9 +151,7 @@ class SlidingWindowCompact(CompactStrategy):
self.keep_recent = keep_recent
self.compact_threshold = compact_threshold
def compact(
self, messages: list[dict], budget_tokens: int
) -> list[dict]:
def compact(self, messages: list[dict], budget_tokens: int) -> list[dict]:
if budget_tokens <= 0 or not messages:
return list(messages)
@ -224,10 +216,7 @@ class SlidingWindowCompact(CompactStrategy):
# Drop oldest-first until under threshold or nothing left.
dropped: set[int] = set()
for i in droppable:
kept = [
m for j, m in enumerate(messages)
if j not in dropped and j != i
]
kept = [m for j, m in enumerate(messages) if j not in dropped and j != i]
if estimate_tokens(kept) <= threshold:
dropped.add(i)
break

View file

@ -65,11 +65,13 @@ class TestEstimateTokens:
_msg(
"assistant",
content = "",
tool_calls = [{
"id": "c1",
"type": "function",
"function": {"name": "web_search", "arguments": '{"q":"hi"}'},
}],
tool_calls = [
{
"id": "c1",
"type": "function",
"function": {"name": "web_search", "arguments": '{"q":"hi"}'},
}
],
)
]
# 10 char arguments -> 2 tokens.
@ -88,7 +90,11 @@ class TestNoCompact:
class TestSlidingWindowUnderBudget:
def test_no_op_when_already_fits(self):
msgs = [_msg("system", "be concise"), _msg("user", "hi"), _msg("assistant", "bye")]
msgs = [
_msg("system", "be concise"),
_msg("user", "hi"),
_msg("assistant", "bye"),
]
out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 1000)
assert out == msgs
@ -164,16 +170,23 @@ class TestSlidingWindowToolPairs:
_msg(
"assistant",
content = "",
tool_calls = [{
"id": "call_42",
"type": "function",
"function": {
"name": "web_search",
"arguments": '{"q":"x"}',
},
}],
tool_calls = [
{
"id": "call_42",
"type": "function",
"function": {
"name": "web_search",
"arguments": '{"q":"x"}',
},
}
],
),
_msg(
"tool",
content = "result for x",
tool_call_id = "call_42",
name = "web_search",
),
_msg("tool", content = "result for x", tool_call_id = "call_42", name = "web_search"),
_long("assistant", 800),
_long("user", 800),
_long("assistant", 800),
@ -185,7 +198,8 @@ class TestSlidingWindowToolPairs:
# Force aggressive compaction.
out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 50)
kept_assistant_with_calls = [
m for m in out
m
for m in out
if m.get("role") == "assistant" and isinstance(m.get("tool_calls"), list)
]
kept_tool_msgs = [m for m in out if m.get("role") == "tool"]
@ -215,22 +229,27 @@ class TestSlidingWindowToolPairs:
_msg(
"assistant",
content = "",
tool_calls = [{
"id": "call_42",
"type": "function",
"function": {
"name": "web_search",
"arguments": '{"q":"x"}',
},
}],
tool_calls = [
{
"id": "call_42",
"type": "function",
"function": {
"name": "web_search",
"arguments": '{"q":"x"}',
},
}
],
),
_msg("tool", content = "result", tool_call_id = "call_42", name = "web_search"),
_msg("user", "thanks"),
]
out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 50)
ids = [
(m.get("role"), m.get("tool_call_id") or
(m.get("tool_calls") and m["tool_calls"][0].get("id")))
(
m.get("role"),
m.get("tool_call_id")
or (m.get("tool_calls") and m["tool_calls"][0].get("id")),
)
for m in out
]
assert ("assistant", "call_42") in ids
@ -250,11 +269,13 @@ class TestStrategyRegistry:
class TestConstructorValidation:
def test_negative_keep_recent_raises(self):
import pytest
with pytest.raises(ValueError):
SlidingWindowCompact(keep_recent = -1)
def test_invalid_threshold_raises(self):
import pytest
with pytest.raises(ValueError):
SlidingWindowCompact(compact_threshold = 0.0)
with pytest.raises(ValueError):