mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-01 02:02:20 +00:00
fix(anthropic): correct streaming usage iteration + translate tools for token counting
- streaming_iterator: skip the trailing "message" iteration entry in the final message_delta when the held stop_reason chunk carries placeholder zero usage (no separate usage chunk arrived). Reporting zero tokens was misleading and inconsistent with the non-streaming path which always has real usage data. - streaming_iterator: drop two redundant type checks inside branches that are already guarded by an outer message_delta type check. - compact._count_effective_tokens: translate Anthropic-shaped tools (input_schema) to OpenAI shape before passing to litellm.token_counter so threshold checks aren't skewed by tokenizer paths that expect the OpenAI tool wrapper. Co-authored-by: Yassin Kortam <yassin@berri.ai>
This commit is contained in:
parent
638a7717f9
commit
a047495ab2
3 changed files with 94 additions and 25 deletions
|
|
@ -139,14 +139,24 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
if not isinstance(usage, dict) or "iterations" in usage:
|
||||
return message_delta_chunk
|
||||
|
||||
message_iteration: UsageIteration = {
|
||||
"type": "message",
|
||||
"input_tokens": usage.get("input_tokens", 0),
|
||||
"output_tokens": usage.get("output_tokens", 0),
|
||||
}
|
||||
input_tokens = usage.get("input_tokens", 0) or 0
|
||||
output_tokens = usage.get("output_tokens", 0) or 0
|
||||
augmented = message_delta_chunk.copy()
|
||||
augmented_usage = dict(usage)
|
||||
augmented_usage["iterations"] = list(self.iterations_usage) + [message_iteration] # type: ignore[typeddict-unknown-key]
|
||||
iterations: List[UsageIteration] = list(self.iterations_usage)
|
||||
# Only emit a ``message`` iteration when we have real token data.
|
||||
# Without a separate usage chunk (e.g. provider sent finish_reason
|
||||
# alone), the held ``message_delta`` carries placeholder zeros from
|
||||
# the translate step; reporting a zero-token iteration would be
|
||||
# misleading and inconsistent with the non-streaming path.
|
||||
if input_tokens > 0 or output_tokens > 0:
|
||||
message_iteration: UsageIteration = {
|
||||
"type": "message",
|
||||
"input_tokens": input_tokens,
|
||||
"output_tokens": output_tokens,
|
||||
}
|
||||
iterations.append(message_iteration)
|
||||
augmented_usage["iterations"] = iterations # type: ignore[typeddict-unknown-key]
|
||||
augmented["usage"] = augmented_usage
|
||||
return augmented
|
||||
|
||||
|
|
@ -352,10 +362,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
if processed_chunk.get("delta", {}).get("stop_reason") is not None:
|
||||
self.holding_stop_reason_chunk = processed_chunk
|
||||
else:
|
||||
if processed_chunk.get("type") == "message_delta":
|
||||
processed_chunk = self._augment_message_delta_usage(
|
||||
processed_chunk
|
||||
)
|
||||
processed_chunk = self._augment_message_delta_usage(
|
||||
processed_chunk
|
||||
)
|
||||
self.chunk_queue.append(processed_chunk)
|
||||
return self.chunk_queue.popleft()
|
||||
elif self.holding_chunk is not None:
|
||||
|
|
@ -561,10 +570,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
):
|
||||
self.holding_stop_reason_chunk = processed_chunk
|
||||
else:
|
||||
if processed_chunk.get("type") == "message_delta":
|
||||
processed_chunk = self._augment_message_delta_usage(
|
||||
processed_chunk
|
||||
)
|
||||
processed_chunk = self._augment_message_delta_usage(
|
||||
processed_chunk
|
||||
)
|
||||
self.chunk_queue.append(processed_chunk)
|
||||
return self.chunk_queue.popleft()
|
||||
elif self.holding_chunk is not None:
|
||||
|
|
|
|||
|
|
@ -231,11 +231,10 @@ def _count_effective_tokens(
|
|||
)
|
||||
|
||||
messages_without_compaction = _strip_compaction_blocks(effective_messages)
|
||||
adapter = LiteLLMAnthropicMessagesAdapter()
|
||||
try:
|
||||
openai_shape = (
|
||||
LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
|
||||
messages=cast(Any, messages_without_compaction)
|
||||
)
|
||||
openai_shape = adapter.translate_anthropic_messages_to_openai(
|
||||
messages=cast(Any, messages_without_compaction)
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_logger.debug(
|
||||
|
|
@ -245,10 +244,30 @@ def _count_effective_tokens(
|
|||
)
|
||||
openai_shape = cast(Any, messages_without_compaction)
|
||||
|
||||
# Translate Anthropic-shaped tools (``input_schema``) to OpenAI-shaped
|
||||
# tools (``{"type": "function", "function": {...}}``) so ``token_counter``
|
||||
# gets a consistent format regardless of which counting path it uses.
|
||||
# An inaccurate tool token count here could cause the polyfill to skip
|
||||
# needed compaction or trigger unnecessary summarization.
|
||||
openai_tools: Optional[List[Dict[str, Any]]] = None
|
||||
if tools:
|
||||
try:
|
||||
translated_tools, _ = adapter.translate_anthropic_tools_to_openai(
|
||||
tools=cast(Any, tools)
|
||||
)
|
||||
openai_tools = cast(List[Dict[str, Any]], translated_tools)
|
||||
except Exception as e:
|
||||
verbose_logger.debug(
|
||||
"compact_20260112: anthropic→openai tools translation failed "
|
||||
"during token count, falling back to raw tools: %s",
|
||||
e,
|
||||
)
|
||||
openai_tools = tools
|
||||
|
||||
total = litellm.token_counter(
|
||||
model=model,
|
||||
messages=cast(Any, openai_shape),
|
||||
tools=cast(Any, tools),
|
||||
tools=cast(Any, openai_tools),
|
||||
)
|
||||
if compaction_block is not None:
|
||||
content = compaction_block.get("content") or ""
|
||||
|
|
|
|||
|
|
@ -12,20 +12,26 @@ sys.path.insert(0, os.path.abspath("../../../../.."))
|
|||
from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import (
|
||||
AnthropicStreamWrapper,
|
||||
)
|
||||
from litellm.types.utils import Delta, StreamingChoices
|
||||
from litellm.types.utils import Delta, StreamingChoices, Usage
|
||||
|
||||
|
||||
def _make_text_chunk(text: str, finish_reason: str = None) -> MagicMock:
|
||||
def _make_text_chunk(
|
||||
text: str,
|
||||
finish_reason: str = None,
|
||||
usage: "Usage | None" = None,
|
||||
) -> MagicMock:
|
||||
chunk = MagicMock()
|
||||
chunk.choices = [
|
||||
StreamingChoices(
|
||||
finish_reason=finish_reason,
|
||||
index=0,
|
||||
delta=Delta(content=text, role="assistant" if text else None, tool_calls=None),
|
||||
delta=Delta(
|
||||
content=text, role="assistant" if text else None, tool_calls=None
|
||||
),
|
||||
logprobs=None,
|
||||
)
|
||||
]
|
||||
chunk.usage = None
|
||||
chunk.usage = usage
|
||||
chunk._hidden_params = {}
|
||||
return chunk
|
||||
|
||||
|
|
@ -43,7 +49,11 @@ async def test_stream_emits_compaction_block_before_text():
|
|||
|
||||
async def mock_stream():
|
||||
yield _make_text_chunk("Hi")
|
||||
yield _make_text_chunk("", finish_reason="stop")
|
||||
yield _make_text_chunk(
|
||||
"",
|
||||
finish_reason="stop",
|
||||
usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
|
||||
)
|
||||
|
||||
compaction_block = {
|
||||
"type": "compaction",
|
||||
|
|
@ -78,7 +88,9 @@ async def test_stream_emits_compaction_block_before_text():
|
|||
and e.get("delta", {}).get("type") == "compaction_delta"
|
||||
)
|
||||
assert compaction_delta["index"] == 0
|
||||
assert compaction_delta["delta"]["content"] == "Summary of prior conversation turns."
|
||||
assert (
|
||||
compaction_delta["delta"]["content"] == "Summary of prior conversation turns."
|
||||
)
|
||||
|
||||
compaction_stop = next(
|
||||
e
|
||||
|
|
@ -100,6 +112,36 @@ async def test_stream_emits_compaction_block_before_text():
|
|||
assert iterations is not None
|
||||
assert iterations[0]["type"] == "compaction"
|
||||
assert iterations[1]["type"] == "message"
|
||||
assert iterations[1]["input_tokens"] == 10
|
||||
assert iterations[1]["output_tokens"] == 5
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_omits_message_iteration_when_no_usage_chunk():
|
||||
"""When provider sends finish_reason without usage, the held message_delta
|
||||
carries placeholder zeros — we must not emit a misleading zero-token
|
||||
``message`` iteration entry."""
|
||||
|
||||
async def mock_stream():
|
||||
yield _make_text_chunk("Hi")
|
||||
yield _make_text_chunk("", finish_reason="stop")
|
||||
|
||||
iterations_usage = [
|
||||
{"type": "compaction", "input_tokens": 100, "output_tokens": 50},
|
||||
]
|
||||
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=mock_stream(),
|
||||
model="claude-sonnet-4-6",
|
||||
iterations_usage=iterations_usage,
|
||||
)
|
||||
|
||||
events = await _collect_events_async(wrapper)
|
||||
message_delta = next(e for e in events if e.get("type") == "message_delta")
|
||||
iterations = message_delta.get("usage", {}).get("iterations")
|
||||
assert iterations is not None
|
||||
assert len(iterations) == 1
|
||||
assert iterations[0]["type"] == "compaction"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue