fix(logging): add system_prompt to StandardLoggingPayload

Anthropic /v1/messages clients send system as a list of content blocks,
but StandardLoggingPayload only surfaced string system kwargs via
append_system_prompt_messages. List-form system prompts were dropped
silently from the logging payload.

Add get_system_prompt_from_kwargs to coalesce system_instructions,
instructions, and system (matching OTel precedence) and populate a new
system_prompt field on StandardLoggingPayload without mutating messages.

Fixes #36402
This commit is contained in:
Souravrajvi0 2026-08-10 14:05:35 +00:00
parent d9b9676d49
commit 850eada677
3 changed files with 110 additions and 0 deletions

View file

@ -4555,6 +4555,26 @@ class StandardLoggingPayloadSetup:
return start_time_float, end_time_float, completion_start_time_float
@staticmethod
def get_system_prompt_from_kwargs(kwargs: Optional[Dict] = None) -> Optional[Union[str, list, dict]]:
"""
Return the system prompt kwargs as sent by the client, without reshaping.
Coalesces the kwarg names used across call paths (Vertex Gemini, Responses API,
Anthropic Messages). Uses `is not None` checks so falsy values like [] do not
fall through to a different kwarg.
"""
if kwargs is None:
return None
if kwargs.get("system_instructions") is not None:
return kwargs.get("system_instructions")
if kwargs.get("instructions") is not None:
return kwargs.get("instructions")
if kwargs.get("system") is not None:
return kwargs.get("system")
return None
@staticmethod
def append_system_prompt_messages(kwargs: Optional[Dict] = None, messages: Optional[Any] = None):
"""
@ -5451,6 +5471,7 @@ def get_standard_logging_object_payload(
kwargs=kwargs, messages=kwargs.get("messages")
)
),
system_prompt=StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs=kwargs),
response=final_response_obj,
model_parameters=ModelParamHelper.get_standard_logging_model_parameters(
kwargs.get("optional_params", None) or {}

View file

@ -3070,6 +3070,7 @@ class StandardLoggingPayload(TypedDict):
requester_ip_address: Optional[str]
user_agent: Optional[str]
messages: Optional[Union[str, list, dict]]
system_prompt: Optional[Union[str, list, dict]]
response: Optional[Union[str, list, dict]]
error_str: Optional[str]
error_information: Optional[StandardLoggingPayloadErrorInformation]

View file

@ -2163,6 +2163,94 @@ def test_append_system_prompt_messages():
assert result == messages
def test_get_system_prompt_from_kwargs():
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
# Anthropic Messages list-form system blocks
system_blocks = [
{"type": "text", "text": "SHAPE-SECRET", "cache_control": {"type": "ephemeral"}},
]
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={"system": system_blocks, "messages": [{"role": "user", "content": "hi"}]}
)
assert result == system_blocks
# String system kwarg
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs={"system": "Be helpful"})
assert result == "Be helpful"
# Responses API instructions
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs={"instructions": "Follow policy"})
assert result == "Follow policy"
# Vertex Gemini system_instructions
gemini_system = [{"role": "system", "content": "Be concise."}]
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={"system_instructions": gemini_system}
)
assert result == gemini_system
# system_instructions wins over instructions and system
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={
"system_instructions": "From Gemini",
"instructions": "From Responses",
"system": "From Anthropic",
}
)
assert result == "From Gemini"
# Empty list should not fall through to instructions
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={"system_instructions": [], "instructions": "From Responses"}
)
assert result == []
# No system kwargs
assert StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs={}) is None
assert StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs=None) is None
def test_get_standard_logging_object_payload_includes_system_prompt_for_list_system(logging_obj):
"""List-form Anthropic system blocks must appear on the payload without mutating messages."""
import datetime
from litellm.litellm_core_utils.litellm_logging import get_standard_logging_object_payload
system_blocks = [
{"type": "text", "text": "SHAPE-SECRET", "cache_control": {"type": "ephemeral"}},
]
user_messages = [{"role": "user", "content": "hello"}]
kwargs = {
"model": "anthropic/claude-sonnet-4-5",
"system": system_blocks,
"messages": user_messages,
"litellm_params": {},
}
mock_response = {
"id": "msg_123",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "hi"}],
"model": "claude-sonnet-4-5",
"usage": {"input_tokens": 5, "output_tokens": 2},
}
now = datetime.datetime.now()
payload = get_standard_logging_object_payload(
kwargs=kwargs,
init_response_obj=mock_response,
start_time=now,
end_time=now,
logging_obj=logging_obj,
status="success",
)
assert payload is not None
assert payload["system_prompt"] == system_blocks
assert payload["messages"] == user_messages
@pytest.mark.asyncio
async def test_async_success_handler_sets_standard_logging_object_for_pass_through_endpoints():
"""