ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot] 2026-06-16 07:59:29 +00:00
commit cc96dc8aad
3 changed files with 48 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

6
studio/backend/core/inference/context_compaction.py
View file

@ -159,7 +159,11 @@ class SlidingWindowCompact(CompactStrategy):
groups. Asst+tools form one group. No-op when within budget.
"""
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):

38
studio/backend/tests/test_context_compaction.py
View file

@ -21,7 +21,13 @@ from core.inference.context_compaction import (
)
def _msg(role, content = "", tool_calls = None, tool_call_id = None, name = None):
def _msg(
role,
content = "",
tool_calls = None,
tool_call_id = None,
name = None,
):
m = {"role": role}
if content is not None:
m["content"] = content
@ -34,7 +40,12 @@ def _msg(role, content = "", tool_calls = None, tool_call_id = None, name = None
return m
def _long(role, length, *, content_prefix = "x"):
def _long(
role,
length,
*,
content_prefix = "x",
):
return _msg(role, content_prefix * length)
@ -117,7 +128,11 @@ class TestSlidingWindowUnderBudget:
class TestSlidingWindowInvariants:
def _make_long_chat(self, n_turns, length_per_turn = 1000):
def _make_long_chat(
self,
n_turns,
length_per_turn = 1000,
):
msgs = [_msg("system", "system prompt")]
msgs.append(_msg("user", "the original task: " + "x" * length_per_turn))
# Alternating assistant/user follow-ups.
@ -131,10 +146,7 @@ class TestSlidingWindowInvariants:
out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 200)
assert out[0]["role"] == "system"
# First user message must survive.
assert any(
m.get("role") == "user" and "original task" in m.get("content", "")
for m in out
)
assert any(m.get("role") == "user" and "original task" in m.get("content", "") for m in out)
def test_keeps_last_n_turns(self):
msgs = self._make_long_chat(n_turns = 20)
@ -209,9 +221,7 @@ class TestSlidingWindowToolPairs:
# Force aggressive compaction.
out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 50)
kept_assistant_with_calls = [
m
for m in out
if m.get("role") == "assistant" and isinstance(m.get("tool_calls"), list)
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"]
# If the assistant tool-call message survives, every matching
@ -258,8 +268,7 @@ class TestSlidingWindowToolPairs:
ids = [
(
m.get("role"),
m.get("tool_call_id")
or (m.get("tool_calls") and m["tool_calls"][0].get("id")),
m.get("tool_call_id") or (m.get("tool_calls") and m["tool_calls"][0].get("id")),
)
for m in out
]
@ -350,21 +359,18 @@ class TestAnchoredMultimodalPairCleanup:
out = SlidingWindowCompact(keep_recent = 1).compact(msgs, budget_tokens = 50)
# Anchored multimodal assistant must still be there.
assert any(
m.get("role") == "assistant" and isinstance(m.get("content"), list)
for m in out
m.get("role") == "assistant" and isinstance(m.get("content"), list) for m in out
), "multimodal assistant got dropped by pair_map cleanup"
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):

32
studio/backend/tests/test_context_compaction_edge.py
View file

@ -34,7 +34,13 @@ from core.inference.context_compaction import (
)
def _msg(role, content = "", tool_calls = None, tool_call_id = None, name = None):
def _msg(
role,
content = "",
tool_calls = None,
tool_call_id = None,
name = None,
):
m = {"role": role}
if content is not None:
m["content"] = content
@ -47,11 +53,20 @@ def _msg(role, content = "", tool_calls = None, tool_call_id = None, name = None
return m
def _long(role, length, *, content_prefix = "x"):
def _long(
role,
length,
*,
content_prefix = "x",
):
return _msg(role, content_prefix * length)
def _tool_call(tcid, name = "web_search", args = '{"q":"x"}'):
def _tool_call(
tcid,
name = "web_search",
args = '{"q":"x"}',
):
return {
"id": tcid,
"type": "function",
@ -272,9 +287,7 @@ class TestDuplicateToolCallIds:
]
out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 30)
# The recent-window assistant+tool group must stay linked.
kept_asst_tool = [
m for m in out if m.get("role") == "assistant" and m.get("tool_calls")
]
kept_asst_tool = [m for m in out if m.get("role") == "assistant" and m.get("tool_calls")]
kept_tools = [m for m in out if m.get("role") == "tool"]
if kept_tools:
tool_ids_in_out = {m["tool_call_id"] for m in kept_tools}
@ -355,8 +368,7 @@ class TestNoSystemNoFirstUser:
out = SlidingWindowCompact(keep_recent = 1).compact(msgs, budget_tokens = 30)
# The first user must be present.
assert any(
m.get("role") == "user" and "real first task" in m.get("content", "")
for m in out
m.get("role") == "user" and "real first task" in m.get("content", "") for m in out
)
@ -851,9 +863,7 @@ def test_compaction_content_part_is_not_multimodal_anchor():
assert not any(
m.get("role") == "assistant"
and isinstance(m.get("content"), list)
and any(
isinstance(p, dict) and p.get("type") == "compaction" for p in m["content"]
)
and any(isinstance(p, dict) and p.get("type") == "compaction" for p in m["content"])
for m in out
)