mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-10 03:28:53 +00:00
fix(anthropic_messages): key bridged streaming spend rows on the streamed msg_ id
A streaming /v1/messages call against a non-Anthropic model is served an SSE message_start frame carrying a msg_ id the adapter mints locally, since the Responses API upstream only issues a resp_ id. That value never left the adapter, so the spend row was keyed on the bridged response id and GET /spend/logs?request_id=msg_... came back empty. The adapter now hands the id it minted to the logging object, and the /v1/messages logging path keys the row on it.
This commit is contained in:
parent
658f50663d
commit
7d8e1c6a1d
5 changed files with 234 additions and 5 deletions
|
|
@ -429,6 +429,7 @@ class Logging(LiteLLMLoggingBaseClass):
|
|||
custom_pricing: bool = False
|
||||
stream_options = None
|
||||
litellm_request_debug: bool = False
|
||||
streamed_anthropic_message_id: str | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
|
@ -2136,7 +2137,7 @@ class Logging(LiteLLMLoggingBaseClass):
|
|||
self.model_call_details["cache_hit"] = cache_hit
|
||||
|
||||
if self.call_type == CallTypes.anthropic_messages.value:
|
||||
result = self._handle_anthropic_messages_response_logging(result=result)
|
||||
result = self._anthropic_messages_logged_response(result=result)
|
||||
elif (
|
||||
self.call_type == CallTypes.generate_content.value
|
||||
or self.call_type == CallTypes.agenerate_content.value
|
||||
|
|
@ -3806,6 +3807,23 @@ class Logging(LiteLLMLoggingBaseClass):
|
|||
)
|
||||
return None
|
||||
|
||||
def record_streamed_anthropic_message_id(self, message_id: str) -> None:
|
||||
self.streamed_anthropic_message_id = message_id
|
||||
|
||||
def _anthropic_messages_logged_response(self, result: Any) -> ModelResponse:
|
||||
"""
|
||||
The ModelResponse a /v1/messages spend_logs row is built from.
|
||||
|
||||
A streaming call bridged onto the Responses API is the one case where the `msg_` id the
|
||||
caller was served is minted locally rather than issued upstream, so it is absent from the
|
||||
response the row would otherwise be keyed on and has to be carried over here.
|
||||
"""
|
||||
logged: Final = self._handle_anthropic_messages_response_logging(result=result)
|
||||
streamed_message_id: Final = self.streamed_anthropic_message_id
|
||||
if streamed_message_id is None:
|
||||
return logged
|
||||
return logged.model_copy(update={"id": streamed_message_id})
|
||||
|
||||
def _handle_anthropic_messages_response_logging(self, result: Any) -> ModelResponse:
|
||||
"""
|
||||
Handles logging for Anthropic messages responses.
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ Used when the target model is an OpenAI or Azure model.
|
|||
"""
|
||||
|
||||
from collections.abc import AsyncIterator, Coroutine, Mapping
|
||||
from typing import Any, Final, TypeAlias
|
||||
from typing import TYPE_CHECKING, Any, Final, TypeAlias
|
||||
|
||||
import litellm
|
||||
from litellm.types.llms.anthropic import (
|
||||
|
|
@ -24,11 +24,21 @@ from ..utils import local_model_name
|
|||
from .streaming_iterator import AnthropicResponsesStreamWrapper
|
||||
from .transformation import LiteLLMAnthropicToResponsesAPIAdapter
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
|
||||
|
||||
AnthropicRequestMessages: TypeAlias = list[AllAnthropicMessageValues] | list[dict[str, object]]
|
||||
|
||||
_ADAPTER: Final = LiteLLMAnthropicToResponsesAPIAdapter()
|
||||
|
||||
|
||||
def _litellm_logging_obj(responses_kwargs: Mapping[str, object]) -> "LiteLLMLoggingObject | None":
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
|
||||
candidate: Final = responses_kwargs.get("litellm_logging_obj")
|
||||
return candidate if isinstance(candidate, Logging) else None
|
||||
|
||||
|
||||
def _forwarded_kwargs(extra_kwargs: Mapping[str, object] | None) -> Mapping[str, object]:
|
||||
"""The litellm-specific kwargs forwarded verbatim onto the Responses API request."""
|
||||
return extra_kwargs or {}
|
||||
|
|
@ -186,7 +196,9 @@ class LiteLLMMessagesToResponsesAPIHandler:
|
|||
|
||||
if stream:
|
||||
wrapper: Final = AnthropicResponsesStreamWrapper(
|
||||
responses_stream=result, model=local_model_name(model, kwargs.get("custom_llm_provider"))
|
||||
responses_stream=result,
|
||||
model=local_model_name(model, kwargs.get("custom_llm_provider")),
|
||||
litellm_logging_obj=_litellm_logging_obj(responses_kwargs),
|
||||
)
|
||||
return wrapper.async_anthropic_sse_wrapper()
|
||||
|
||||
|
|
@ -266,7 +278,9 @@ class LiteLLMMessagesToResponsesAPIHandler:
|
|||
|
||||
if stream:
|
||||
wrapper: Final = AnthropicResponsesStreamWrapper(
|
||||
responses_stream=result, model=local_model_name(model, kwargs.get("custom_llm_provider"))
|
||||
responses_stream=result,
|
||||
model=local_model_name(model, kwargs.get("custom_llm_provider")),
|
||||
litellm_logging_obj=_litellm_logging_obj(responses_kwargs),
|
||||
)
|
||||
return wrapper.async_anthropic_sse_wrapper()
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import json
|
|||
import traceback
|
||||
from collections import deque
|
||||
from collections.abc import AsyncIterator, Mapping
|
||||
from typing import Any, Final
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from litellm import verbose_logger
|
||||
from litellm._uuid import uuid
|
||||
|
|
@ -12,6 +12,9 @@ from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUs
|
|||
|
||||
from .transformation import LiteLLMAnthropicToResponsesAPIAdapter
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
|
||||
|
||||
|
||||
class AnthropicResponsesStreamWrapper:
|
||||
"""
|
||||
|
|
@ -31,10 +34,13 @@ class AnthropicResponsesStreamWrapper:
|
|||
self,
|
||||
responses_stream: Any,
|
||||
model: str,
|
||||
litellm_logging_obj: "LiteLLMLoggingObject | None" = None,
|
||||
) -> None:
|
||||
self.responses_stream = responses_stream
|
||||
self.model = model
|
||||
self._message_id: str = f"msg_{uuid.uuid4()}"
|
||||
if litellm_logging_obj is not None:
|
||||
litellm_logging_obj.record_streamed_anthropic_message_id(self._message_id)
|
||||
self._current_block_index: int = -1
|
||||
# Map item_id -> content_block_index so we can stop the right block later
|
||||
self._item_id_to_block_index: dict[str, int] = {}
|
||||
|
|
|
|||
|
|
@ -1,9 +1,11 @@
|
|||
import datetime
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
|
||||
|
||||
|
|
@ -15,6 +17,18 @@ from litellm.llms.anthropic.experimental_pass_through.responses_adapters.handler
|
|||
|
||||
MESSAGES = [{"role": "user", "content": "hello"}]
|
||||
|
||||
RESPONSES_SSE_BODY = (
|
||||
b"event: response.created\n"
|
||||
b'data: {"type":"response.created","sequence_number":0,"response":{"id":"resp_lit6825",'
|
||||
b'"object":"response","created_at":1,"status":"in_progress","model":"gpt-5.6-luna","output":[],'
|
||||
b'"parallel_tool_calls":true,"tool_choice":"auto","tools":[]}}\n\n'
|
||||
b"event: response.completed\n"
|
||||
b'data: {"type":"response.completed","sequence_number":1,"response":{"id":"resp_lit6825",'
|
||||
b'"object":"response","created_at":1,"status":"completed","model":"gpt-5.6-luna","output":[],'
|
||||
b'"parallel_tool_calls":true,"tool_choice":"auto","tools":[],'
|
||||
b'"usage":{"input_tokens":3,"output_tokens":4,"total_tokens":7}}}\n\n'
|
||||
)
|
||||
|
||||
|
||||
def test_build_responses_kwargs_derives_prompt_cache_key_from_user_id():
|
||||
responses_kwargs = _build_responses_kwargs(
|
||||
|
|
@ -82,3 +96,48 @@ async def test_streaming_message_start_reports_the_provider_local_model(requeste
|
|||
|
||||
message_start = next(e for e in events if e["type"] == "message_start")
|
||||
assert message_start["message"]["model"] == expected_reported_model
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_hands_the_logging_object_the_message_id_the_caller_is_streamed(
|
||||
respx_mock: respx.MockRouter, monkeypatch
|
||||
):
|
||||
"""
|
||||
The bridge mints the ``msg_`` id itself, and it is the only request id a streaming
|
||||
/v1/messages caller ever sees, so the spend row has to be keyed on that same value.
|
||||
"""
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "sk-lit6825-test")
|
||||
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
litellm.in_memory_llm_clients_cache.flush_cache()
|
||||
respx_mock.post("https://api.openai.com/v1/responses").respond(
|
||||
status_code=200,
|
||||
headers={"Content-Type": "text/event-stream"},
|
||||
content=RESPONSES_SSE_BODY,
|
||||
)
|
||||
|
||||
logging_obj = Logging(
|
||||
model="gpt-5.6-luna",
|
||||
messages=MESSAGES,
|
||||
stream=True,
|
||||
call_type="anthropic_messages",
|
||||
start_time=datetime.datetime.now(datetime.timezone.utc),
|
||||
litellm_call_id="6825beef-0000-4000-8000-000000000003",
|
||||
function_id="1234",
|
||||
)
|
||||
|
||||
sse = await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
|
||||
max_tokens=1024,
|
||||
messages=MESSAGES,
|
||||
model="openai/gpt-5.6-luna",
|
||||
stream=True,
|
||||
custom_llm_provider="openai",
|
||||
litellm_logging_obj=logging_obj,
|
||||
)
|
||||
events = [json.loads(chunk.decode().split("data: ", 1)[1]) async for chunk in sse]
|
||||
|
||||
message_start = next(e for e in events if e["type"] == "message_start")
|
||||
assert message_start["message"]["id"].startswith("msg_")
|
||||
assert logging_obj.streamed_anthropic_message_id == message_start["message"]["id"]
|
||||
|
|
|
|||
|
|
@ -3462,6 +3462,138 @@ def test_get_spend_logs_id_prefers_the_response_id_over_the_standard_logging_id(
|
|||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spend_log_request_id_is_the_message_id_a_bridged_streaming_caller_was_streamed():
|
||||
"""A streaming /v1/messages call against a non-Anthropic model is served a msg_ id the
|
||||
adapter mints itself, and it is the only request id that call ever shows the caller, so
|
||||
GET /spend/logs?request_id=msg_... has to land on the row."""
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.llms.anthropic.experimental_pass_through.responses_adapters.streaming_iterator import (
|
||||
AnthropicResponsesStreamWrapper,
|
||||
)
|
||||
from litellm.types.llms.openai import (
|
||||
ResponseAPIUsage,
|
||||
ResponseCompletedEvent,
|
||||
ResponsesAPIResponse,
|
||||
)
|
||||
|
||||
logging_obj = Logging(
|
||||
model="gpt-5.6",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
stream=True,
|
||||
call_type="anthropic_messages",
|
||||
start_time=datetime.datetime.now(timezone.utc),
|
||||
litellm_call_id="6825cafe-0000-4000-8000-000000000001",
|
||||
function_id="1234",
|
||||
)
|
||||
logging_obj.optional_params = {}
|
||||
|
||||
completed_response = ResponsesAPIResponse(
|
||||
id="resp_01Lit6825Bridged",
|
||||
object="response",
|
||||
created_at=1767225600,
|
||||
model="gpt-5.6",
|
||||
status="completed",
|
||||
output=[
|
||||
{
|
||||
"id": "msg_bridged_output",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"status": "completed",
|
||||
"content": [{"type": "output_text", "text": "epsilon", "annotations": []}],
|
||||
}
|
||||
],
|
||||
usage=ResponseAPIUsage(input_tokens=12, output_tokens=5, total_tokens=17),
|
||||
)
|
||||
|
||||
async def _responses_stream():
|
||||
yield {"type": "response.created"}
|
||||
yield {"type": "response.output_text.delta", "item_id": "msg_bridged_output", "delta": "epsilon"}
|
||||
yield ResponseCompletedEvent(type="response.completed", response=completed_response)
|
||||
|
||||
wrapper = AnthropicResponsesStreamWrapper(
|
||||
responses_stream=_responses_stream(),
|
||||
model="gpt-5.6",
|
||||
litellm_logging_obj=logging_obj,
|
||||
)
|
||||
sse_frames = [frame.decode() async for frame in wrapper.async_anthropic_sse_wrapper()]
|
||||
|
||||
message_start_frames = [f for f in sse_frames if f.startswith("event: message_start\n")]
|
||||
assert len(message_start_frames) == 1
|
||||
streamed_message_id = json.loads(message_start_frames[0].split("data: ", 1)[1])["message"]["id"]
|
||||
assert streamed_message_id.startswith("msg_")
|
||||
|
||||
_, _, logged_response = logging_obj._success_handler_helper_fn(
|
||||
result=ResponseCompletedEvent(type="response.completed", response=completed_response),
|
||||
start_time=datetime.datetime.now(timezone.utc),
|
||||
end_time=datetime.datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
assert logged_response.id == streamed_message_id
|
||||
payload = get_logging_payload(
|
||||
kwargs={
|
||||
"call_type": "anthropic_messages",
|
||||
"model": "gpt-5.6",
|
||||
"litellm_call_id": "6825cafe-0000-4000-8000-000000000001",
|
||||
"litellm_params": {"metadata": {"user_api_key": "test-key"}},
|
||||
},
|
||||
response_obj=logged_response,
|
||||
start_time=datetime.datetime.now(timezone.utc),
|
||||
end_time=datetime.datetime.now(timezone.utc),
|
||||
)
|
||||
assert payload["request_id"] == streamed_message_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spend_log_request_id_is_untouched_when_no_message_id_was_streamed():
|
||||
"""Only the bridged streaming adapter mints a msg_ id of its own, so every other
|
||||
/v1/messages call must keep the id its own response carried."""
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.types.llms.openai import (
|
||||
ResponseAPIUsage,
|
||||
ResponseCompletedEvent,
|
||||
ResponsesAPIResponse,
|
||||
)
|
||||
|
||||
logging_obj = Logging(
|
||||
model="gpt-5.6",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
stream=False,
|
||||
call_type="anthropic_messages",
|
||||
start_time=datetime.datetime.now(timezone.utc),
|
||||
litellm_call_id="6825cafe-0000-4000-8000-000000000002",
|
||||
function_id="1234",
|
||||
)
|
||||
logging_obj.optional_params = {}
|
||||
|
||||
completed_response = ResponsesAPIResponse(
|
||||
id="resp_01Lit6825Unbridged",
|
||||
object="response",
|
||||
created_at=1767225600,
|
||||
model="gpt-5.6",
|
||||
status="completed",
|
||||
output=[
|
||||
{
|
||||
"id": "msg_unbridged_output",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"status": "completed",
|
||||
"content": [{"type": "output_text", "text": "epsilon", "annotations": []}],
|
||||
}
|
||||
],
|
||||
usage=ResponseAPIUsage(input_tokens=12, output_tokens=5, total_tokens=17),
|
||||
)
|
||||
|
||||
_, _, logged_response = logging_obj._success_handler_helper_fn(
|
||||
result=ResponseCompletedEvent(type="response.completed", response=completed_response),
|
||||
start_time=datetime.datetime.now(timezone.utc),
|
||||
end_time=datetime.datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
assert logged_response.id
|
||||
assert not logged_response.id.startswith("msg_")
|
||||
|
||||
|
||||
def test_batch_cost_row_does_not_collide_with_the_batch_creation_row():
|
||||
"""Creating a batch writes a row keyed by the batch's own id, so keying the cost row
|
||||
the same way makes the insert a duplicate of it. request_id is the primary key and the
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue