mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-05 07:37:50 +00:00
fix(context_management): count clear_at_least freed tokens over sent blocks only
This commit is contained in:
parent
b089560973
commit
e41b3c8769
2 changed files with 37 additions and 3 deletions
|
|
@ -192,10 +192,12 @@ def _cleared_result(block: Mapping[str, object]) -> dict[str, object]:
|
|||
|
||||
def _freed_tokens(model: str, result: Mapping[str, object], use: Mapping[str, object], clears_input: bool) -> int:
|
||||
"""Tokens that clearing this use frees: its result, and its input when ``clear_tool_inputs`` covers it."""
|
||||
before: Final = [result, use] if clears_input else [result]
|
||||
after: Final = [_cleared_result(result), {**use, "input": {}}] if clears_input else [_cleared_result(result)]
|
||||
before: Final[Sequence[Mapping[str, object]]] = (result, use) if clears_input else (result,)
|
||||
after: Final[Sequence[Mapping[str, object]]] = (
|
||||
(_cleared_result(result), {**use, "input": {}}) if clears_input else (_cleared_result(result),)
|
||||
)
|
||||
counts: Final = tuple(
|
||||
litellm.token_counter(model=model, messages=[{"role": "user", "content": blocks}]) for blocks in (before, after)
|
||||
_count_tokens(model, [{"role": "user", "content": blocks}], None) for blocks in (before, after)
|
||||
)
|
||||
return max(counts[0] - counts[1], 0)
|
||||
|
||||
|
|
|
|||
|
|
@ -288,6 +288,38 @@ def test_an_input_tokens_trigger_skips_blocks_no_adapter_sends_inside_a_tool_res
|
|||
assert _cleared_ids(cleared) == {"toolu_00", "toolu_01"}
|
||||
|
||||
|
||||
def _clear_at_least_one_token(
|
||||
messages: list[dict[str, object]],
|
||||
) -> tuple[list[dict[str, object]], AppliedEdit | None]:
|
||||
return apply_clear_tool_uses_20250919(
|
||||
model=MODEL,
|
||||
messages=messages,
|
||||
tools=None,
|
||||
system=None,
|
||||
edit_spec={
|
||||
"type": "clear_tool_uses_20250919",
|
||||
"trigger": {"type": "tool_uses", "value": 0},
|
||||
"keep": {"type": "tool_uses", "value": 1},
|
||||
"clear_at_least": {"type": "input_tokens", "value": 1},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_clear_at_least_frees_only_blocks_an_adapter_sends_inside_a_tool_result() -> None:
|
||||
history: Final = _make_history(n_pairs=3)
|
||||
result_text: Final = {"type": "text", "text": "Result 0: " + "x" * 200}
|
||||
search_result: Final = {"type": "web_search_result", "url": "https://example.com", "title": "rain " * 300}
|
||||
|
||||
_, sent_only = _clear_at_least_one_token([*history[:2], _first_result_with([result_text]), *history[3:]])
|
||||
cleared, applied = _clear_at_least_one_token(
|
||||
[*history[:2], _first_result_with([result_text, search_result]), *history[3:]]
|
||||
)
|
||||
|
||||
assert applied is not None
|
||||
assert applied == sent_only
|
||||
assert _cleared_ids(cleared) == {"toolu_00", "toolu_01"}
|
||||
|
||||
|
||||
def _blocks_of(messages: list[dict[str, object]], block_type: str) -> tuple[dict[str, object], ...]:
|
||||
contents: Final = (message["content"] for message in messages)
|
||||
blocks: Final = chain.from_iterable(content for content in contents if isinstance(content, list))
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue