From 80fa6a251ee360e3f12c47d1bb9319ace35ba785 Mon Sep 17 00:00:00 2001 From: Vigilans Date: Fri, 17 Apr 2026 00:01:03 +0800 Subject: [PATCH] fix(logging): create spend logs for native /v1/messages handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The native Anthropic messages handler (AnthropicMessagesConfig) reuses PassThroughStreamingHandler for SSE iteration. Several issues prevented correct spend log entries: 1. GitHub Copilot appends "data: [DONE]" (OpenAI format) at the end of Anthropic SSE streams. The chunk parser raised json.JSONDecodeError on this line, aborting the entire logging flow before the SLP could be built. 2. async_success_handler only built the StandardLoggingPayload for call_type "pass_through_endpoint". The native handler uses call_type "anthropic_messages", so the SLP branch was skipped. 3. Spend log entries used litellm_call_id (UUID) instead of the actual Anthropic response ID (msg_*). Preserve the original ID from message_start SSE events (streaming) and httpx_response (non-streaming). 4. SLP["response"] (the "Response" field in the spend log UI, gated by `store_prompts_in_spend_logs`) recorded the chat.completion-shaped ModelResponse produced by transform_response / stream_chunk_builder instead of the authentic Anthropic /v1/messages JSON. Fixed by: - Non-streaming: capture the raw JSON from httpx_response before transform_response converts it; override SLP["response"] with it. - Streaming: reverse-adapt the aggregated ModelResponse via the existing LiteLLMAnthropicMessagesAdapter (gated on message_start SSE detection) — Anthropic SSE has no terminating event carrying the final message dict. - Preserve citations across streaming by accumulating them in the chat handler (stream_chunk_builder uses last-value-wins for list fields) and carrying them through the reverse adapter into text blocks. --- litellm/litellm_core_utils/litellm_logging.py | 27 +++++++- litellm/llms/anthropic/chat/handler.py | 8 ++- .../adapters/transformation.py | 16 +++-- .../anthropic_passthrough_logging_handler.py | 67 ++++++++++++++++++- 4 files changed, 109 insertions(+), 9 deletions(-) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index fd14f55add3..6aabec8a4c6 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -1778,6 +1778,12 @@ class Logging(LiteLLMLoggingBaseClass): status="success", standard_built_in_tools_params=self.standard_built_in_tools_params, ) + # Native /v1/messages stashes the raw Anthropic JSON here so spend + # logs / UI show the authentic shape instead of the chat.completion + # reconstruction produced by transform_response / stream_chunk_builder. + anthropic_raw = self.model_call_details.get("anthropic_raw_response") + if anthropic_raw is not None and payload is not None: + payload["response"] = anthropic_raw self.callback_duration_ms += (time.time() - _start) * 1000 return payload @@ -2606,7 +2612,10 @@ class Logging(LiteLLMLoggingBaseClass): ) ) is not None: emit_standard_logging_payload(standard_logging_payload) - elif self.call_type == "pass_through_endpoint": + elif self.call_type in ( + "pass_through_endpoint", + CallTypes.anthropic_messages.value, + ): print_verbose( "Async success callbacks: Got a pass-through endpoint response" ) @@ -3431,6 +3440,16 @@ class Logging(LiteLLMLoggingBaseClass): httpx_response = self.model_call_details.get("httpx_response", None) if httpx_response and isinstance(httpx_response, httpx.Response): + # Capture the raw Anthropic JSON before transform_response rewrites + # it into a chat.completion-shaped ModelResponse, so SLP["response"] + # can preserve the native /v1/messages shape in spend logs. + try: + raw_anthropic_json = httpx_response.json() + except Exception: + raw_anthropic_json = None + if isinstance(raw_anthropic_json, dict): + self.model_call_details["anthropic_raw_response"] = raw_anthropic_json + result = litellm.AnthropicConfig().transform_response( raw_response=httpx_response, model_response=litellm.ModelResponse(), @@ -3444,6 +3463,12 @@ class Logging(LiteLLMLoggingBaseClass): json_mode=False, litellm_params={}, ) + # Preserve the original Anthropic response ID (e.g. msg_*) + # which transform_response does not carry over. + if isinstance(raw_anthropic_json, dict): + original_id = raw_anthropic_json.get("id") + if original_id: + result.id = original_id else: from litellm.types.llms.anthropic import AnthropicResponse diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py index f8e61d0166a..0e89c77ab08 100644 --- a/litellm/llms/anthropic/chat/handler.py +++ b/litellm/llms/anthropic/chat/handler.py @@ -554,6 +554,11 @@ class ModelResponseIterator: # See: https://github.com/BerriAI/litellm/issues/17737 self.web_search_results: List[Dict[str, Any]] = [] + # Accumulate citations from citations_delta events. stream_chunk_builder + # uses "last value wins" for list-valued provider_specific_fields keys, + # so each emission must carry every citation seen so far. + self.citations: List[Dict[str, Any]] = [] + # Accumulate compaction blocks for multi-turn reconstruction self.compaction_blocks: List[Dict[str, Any]] = [] @@ -631,7 +636,8 @@ class ModelResponseIterator: }, ) elif "citation" in content_block["delta"]: - provider_specific_fields["citation"] = content_block["delta"]["citation"] + self.citations.append(content_block["delta"]["citation"]) + provider_specific_fields["citations"] = self.citations elif ( "thinking" in content_block["delta"] or "signature" in content_block["delta"] diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py index 072ae7c3bbe..8ceb0ccc1c6 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py @@ -1243,11 +1243,17 @@ class LiteLLMAnthropicMessagesAdapter: # Handle text content if choice.message.content is not None: - new_content.append( - AnthropicResponseContentBlockText( - type="text", text=choice.message.content - ).model_dump() - ) + text_block = AnthropicResponseContentBlockText( + type="text", text=choice.message.content + ).model_dump() + # Preserve citations accumulated by the streaming handler into + # provider_specific_fields["citations"] so the reverse-adapt + # does not silently drop them. + psf = getattr(choice.message, "provider_specific_fields", None) or {} + citations = psf.get("citations") + if citations: + text_block["citations"] = citations + new_content.append(text_block) # Handle tool calls (in parallel to text content) if ( choice.message.tool_calls is not None diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py index 216eb61a9d1..f340da7212e 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py @@ -12,6 +12,9 @@ from litellm.llms.anthropic import get_anthropic_config from litellm.llms.anthropic.chat.handler import ( ModelResponseIterator as AnthropicModelResponseIterator, ) +from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import ( + LiteLLMAnthropicMessagesAdapter, +) from litellm.proxy._types import PassThroughEndpointLoggingTypedDict from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body from litellm.types.passthrough_endpoints.pass_through_endpoints import ( @@ -163,8 +166,10 @@ class AnthropicPassthroughLoggingHandler: json.dumps(kwargs, indent=4, default=str), ) - # set litellm_call_id to logging response object - litellm_model_response.id = logging_obj.litellm_call_id + # Preserve the original Anthropic response ID (msg_*) when + # available; fall back to litellm_call_id otherwise. + if not litellm_model_response.id.startswith("msg"): + litellm_model_response.id = logging_obj.litellm_call_id litellm_model_response.model = model logging_obj.model_call_details["model"] = model if not logging_obj.model_call_details.get("custom_llm_provider"): @@ -221,6 +226,38 @@ class AnthropicPassthroughLoggingHandler: "result": None, "kwargs": {}, } + + # stream_chunk_builder doesn't preserve the response ID — extract + # it from the message_start SSE event so the spend log uses msg_*. + response_id = ( + AnthropicPassthroughLoggingHandler._extract_response_id_from_chunks( + all_chunks + ) + ) + if response_id: + complete_streaming_response.id = response_id + + # Reverse-adapt the aggregated chat.completion ModelResponse back + # to Anthropic messages shape so SLP["response"] mirrors the + # non-streaming path. Gated on response_id — its presence means + # _extract_response_id_from_chunks matched a `message_start` SSE + # event, confirming the upstream stream is Anthropic messages SSE. + try: + anthropic_shaped = dict( + LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic( + complete_streaming_response + ) + ) + if model: + anthropic_shaped["model"] = model + litellm_logging_obj.model_call_details["anthropic_raw_response"] = ( + anthropic_shaped + ) + except Exception as e: + verbose_proxy_logger.debug( + f"Failed to reverse-adapt streaming response to Anthropic shape: {e}" + ) + kwargs = AnthropicPassthroughLoggingHandler._create_anthropic_response_logging_payload( litellm_model_response=complete_streaming_response, model=model, @@ -235,6 +272,29 @@ class AnthropicPassthroughLoggingHandler: "kwargs": kwargs, } + @staticmethod + def _extract_response_id_from_chunks( + all_chunks: Sequence[Union[str, bytes]], + ) -> Optional[str]: + """ + Extract the Anthropic response ID (msg_*) from the message_start SSE event. + + stream_chunk_builder does not preserve the original response ID, so we + scan the raw SSE lines for the first ``data:`` line containing + ``message_start`` and pull the ``message.id`` field from it. + """ + for line in all_chunks: + if isinstance(line, bytes): + line = line.decode("utf-8") + if "message_start" not in line or "data:" not in line: + continue + try: + data_str = line[line.find("data:") + 5 :] + return json.loads(data_str).get("message", {}).get("id") + except (json.JSONDecodeError, AttributeError): + pass + return None + @staticmethod def _split_sse_chunk_into_events(chunk: Union[str, bytes]) -> List[str]: """ @@ -302,6 +362,9 @@ class AnthropicPassthroughLoggingHandler: except (StopIteration, StopAsyncIteration): break + except json.JSONDecodeError: + # Skip non-JSON SSE lines (e.g. "data: [DONE]" from Copilot) + continue complete_streaming_response = litellm.stream_chunk_builder( chunks=all_openai_chunks,