scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Merge 45f9c66d09 into 176b2e5eb8

Browse source
This commit is contained in:
Aftab 2026-08-26 14:30:28 -04:00 • committed by GitHub
commit b9c0d397d8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 597 additions and 402 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

62
litellm/litellm_core_utils/litellm_logging.py
View file

@ -5032,27 +5032,53 @@ class StandardLoggingPayloadSetup:
return start_time_float, end_time_float, completion_start_time_float
@staticmethod
def append_system_prompt_messages(kwargs: dict | None = None, messages: Any | None = None):
def get_system_prompt_from_kwargs(kwargs: dict[str, Any] | None) -> str | list[Any] | None:
"""
Append system prompt messages to the messages
Extract the system prompt from kwargs, checking all known sources.
Priority: system_instructions (Vertex Gemini) > instructions > system (Anthropic /messages).
Returns the value as-is — either a string or a list of content blocks.
"""
if kwargs is not None:
if kwargs.get("system") is not None and isinstance(kwargs.get("system"), str):
if messages is None:
return [{"role": "system", "content": kwargs.get("system")}]
elif isinstance(messages, list):
if len(messages) == 0:
return [{"role": "system", "content": kwargs.get("system")}]
# check for duplicates
if messages[0].get("role") == "system" and messages[0].get("content") == kwargs.get("system"):
return messages
messages = [{"role": "system", "content": kwargs.get("system")}] + messages
elif isinstance(messages, str):
messages = [
{"role": "system", "content": kwargs.get("system")},
{"role": "user", "content": messages},
]
if kwargs is None:
return None
return (
kwargs.get("system_instructions")
if kwargs.get("system_instructions") is not None
else (kwargs.get("instructions") if kwargs.get("instructions") is not None else kwargs.get("system"))
)
@staticmethod
def append_system_prompt_messages(kwargs: dict[str, Any] | None = None, messages: Any | None = None):
"""
Append system prompt messages to the messages list.
Handles both string and list-of-content-blocks system prompts so that the
logged ``messages`` field always reflects what was actually sent, regardless
of whether the caller used the Anthropic string form or the block-list form.
"""
if kwargs is None:
return messages
system = kwargs.get("system")
if system is None:
return messages
if not isinstance(system, (str, list)):
return messages
system_message: dict[str, Any] = {"role": "system", "content": system}
if messages is None:
return [system_message]
elif isinstance(messages, list):
if len(messages) == 0:
return [system_message]
# skip prepend if the first message already carries this exact system content
if messages[0].get("role") == "system" and messages[0].get("content") == system:
return messages
return [system_message] + messages
elif isinstance(messages, str):
return [system_message, {"role": "user", "content": messages}]
return messages

1
litellm/types/utils.py
View file

@ -3235,6 +3235,7 @@ class StandardLoggingPayload(TypedDict):
hidden_params: StandardLoggingHiddenParams
guardrail_information: list[StandardLoggingGuardrailInformation] | None
standard_built_in_tools_params: StandardBuiltInToolsParams | None
system_prompt: str | list | None
from collections.abc import AsyncIterator, Iterator

258
tests/local_testing/create_mock_standard_logging_payload.py
View file

@ -1,128 +1,130 @@
import io
import asyncio
import gzip
import json
import logging
import time
from unittest.mock import AsyncMock, patch
import pytest
import litellm
from litellm import completion
from litellm._logging import verbose_logger
from litellm.integrations.datadog.datadog import *
from datetime import datetime, timedelta
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingModelInformation,
StandardLoggingMetadata,
StandardLoggingHiddenParams,
)
verbose_logger.setLevel(logging.DEBUG)
def create_standard_logging_payload() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!"}],
response={"choices": [{"message": {"content": "Hi there!"}}]},
error_str=None,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
)
def create_standard_logging_payload_with_long_content() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!" * 80000}],
response={"choices": [{"message": {"content": "Hi there!" * 80000}}]},
error_str="error_str" * 80000,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
)
import io
import asyncio
import gzip
import json
import logging
import time
from unittest.mock import AsyncMock, patch
import pytest
import litellm
from litellm import completion
from litellm._logging import verbose_logger
from litellm.integrations.datadog.datadog import *
from datetime import datetime, timedelta
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingModelInformation,
StandardLoggingMetadata,
StandardLoggingHiddenParams,
)
verbose_logger.setLevel(logging.DEBUG)
def create_standard_logging_payload() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!"}],
response={"choices": [{"message": {"content": "Hi there!"}}]},
error_str=None,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
system_prompt=None,
)
def create_standard_logging_payload_with_long_content() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!" * 80000}],
response={"choices": [{"message": {"content": "Hi there!" * 80000}}]},
error_str="error_str" * 80000,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
system_prompt=None,
)

258
tests/logging_callback_tests/create_mock_standard_logging_payload.py
View file

@ -1,128 +1,130 @@
import io
import asyncio
import gzip
import json
import logging
import time
from unittest.mock import AsyncMock, patch
import pytest
import litellm
from litellm import completion
from litellm._logging import verbose_logger
from litellm.integrations.datadog.datadog import *
from datetime import datetime, timedelta
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingModelInformation,
StandardLoggingMetadata,
StandardLoggingHiddenParams,
)
verbose_logger.setLevel(logging.DEBUG)
def create_standard_logging_payload() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!"}],
response={"choices": [{"message": {"content": "Hi there!"}}]},
error_str=None,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
)
def create_standard_logging_payload_with_long_content() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!" * 80000}],
response={"choices": [{"message": {"content": "Hi there!" * 80000}}]},
error_str="error_str" * 80000,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
)
import io
import asyncio
import gzip
import json
import logging
import time
from unittest.mock import AsyncMock, patch
import pytest
import litellm
from litellm import completion
from litellm._logging import verbose_logger
from litellm.integrations.datadog.datadog import *
from datetime import datetime, timedelta
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingModelInformation,
StandardLoggingMetadata,
StandardLoggingHiddenParams,
)
verbose_logger.setLevel(logging.DEBUG)
def create_standard_logging_payload() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!"}],
response={"choices": [{"message": {"content": "Hi there!"}}]},
error_str=None,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
system_prompt=None,
)
def create_standard_logging_payload_with_long_content() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!" * 80000}],
response={"choices": [{"message": {"content": "Hi there!" * 80000}}]},
error_str="error_str" * 80000,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
system_prompt=None,
)

120
tests/logging_callback_tests/test_standard_logging_payload.py
View file

@ -1265,3 +1265,123 @@ def test_merge_litellm_metadata_bedrock_passthrough_scenario():
# Verify total number of fields (9 user fields + 4 model fields = 13)
assert len(result) == 13
# ---------------------------------------------------------------------------
# Tests for system_prompt logging (issue: list-form system was silently dropped)
# ---------------------------------------------------------------------------
class TestAppendSystemPromptMessages:
"""Tests for StandardLoggingPayloadSetup.append_system_prompt_messages."""
def test_string_system_prepended_to_messages(self):
"""String system prompt is prepended as a system message."""
kwargs = {"system": "You are a helpful assistant."}
messages = [{"role": "user", "content": "Hello"}]
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=messages
)
assert result is not None
assert result[0] == {"role": "system", "content": "You are a helpful assistant."}
assert result[1] == {"role": "user", "content": "Hello"}
def test_list_system_prepended_to_messages(self):
"""List-of-content-blocks system prompt is now prepended, not silently dropped."""
system_blocks = [
{"type": "text", "text": "You are a helpful assistant."},
{"type": "text", "text": "Be concise.", "cache_control": {"type": "ephemeral"}},
]
kwargs = {"system": system_blocks}
messages = [{"role": "user", "content": "Hello"}]
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=messages
)
assert result is not None
assert result[0] == {"role": "system", "content": system_blocks}
assert result[1] == {"role": "user", "content": "Hello"}
def test_none_system_returns_messages_unchanged(self):
"""No system key in kwargs leaves messages unchanged."""
messages = [{"role": "user", "content": "Hello"}]
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs={}, messages=messages
)
assert result == messages
def test_none_kwargs_returns_messages_unchanged(self):
"""None kwargs leaves messages unchanged."""
messages = [{"role": "user", "content": "Hello"}]
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=None, messages=messages
)
assert result == messages
def test_string_system_with_empty_messages(self):
"""String system prompt is returned as a single-element list when messages is empty."""
kwargs = {"system": "Be helpful."}
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=[]
)
assert result == [{"role": "system", "content": "Be helpful."}]
def test_list_system_with_empty_messages(self):
"""List system prompt is returned as a single-element list when messages is empty."""
system_blocks = [{"type": "text", "text": "You are an expert."}]
kwargs = {"system": system_blocks}
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=[]
)
assert result == [{"role": "system", "content": system_blocks}]
def test_string_system_with_none_messages(self):
"""String system prompt creates a new message list when messages is None."""
kwargs = {"system": "Be helpful."}
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=None
)
assert result == [{"role": "system", "content": "Be helpful."}]
def test_list_system_with_none_messages(self):
"""List system prompt creates a new message list when messages is None."""
system_blocks = [{"type": "text", "text": "You are an expert."}]
kwargs = {"system": system_blocks}
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=None
)
assert result == [{"role": "system", "content": system_blocks}]
def test_duplicate_string_system_not_prepended(self):
"""String system prompt that matches first message is not prepended again."""
system = "You are a helpful assistant."
kwargs = {"system": system}
messages = [
{"role": "system", "content": system},
{"role": "user", "content": "Hello"},
]
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=messages
)
assert result is not None
assert result[0]["role"] == "system"
assert result[0]["content"] == system
assert len(result) == 2
def test_string_system_with_str_messages(self):
"""String messages are wrapped in a list with the system message prepended."""
kwargs = {"system": "You are helpful."}
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages="Hello world"
)
assert result == [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello world"},
]
def test_non_string_non_list_system_returns_messages_unchanged(self):
"""Non-string, non-list system values are ignored."""
kwargs = {"system": 42}
messages = [{"role": "user", "content": "Hello"}]
result = StandardLoggingPayloadSetup.append_system_prompt_messages(
kwargs=kwargs, messages=messages
)
assert result == messages

258
tests/router_unit_tests/create_mock_standard_logging_payload.py
View file

@ -1,128 +1,130 @@
import io
import asyncio
import gzip
import json
import logging
import time
from unittest.mock import AsyncMock, patch
import pytest
import litellm
from litellm import completion
from litellm._logging import verbose_logger
from litellm.integrations.datadog.datadog import *
from datetime import datetime, timedelta
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingModelInformation,
StandardLoggingMetadata,
StandardLoggingHiddenParams,
)
verbose_logger.setLevel(logging.DEBUG)
def create_standard_logging_payload() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!"}],
response={"choices": [{"message": {"content": "Hi there!"}}]},
error_str=None,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
)
def create_standard_logging_payload_with_long_content() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!" * 80000}],
response={"choices": [{"message": {"content": "Hi there!" * 80000}}]},
error_str="error_str" * 80000,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
)
import io
import asyncio
import gzip
import json
import logging
import time
from unittest.mock import AsyncMock, patch
import pytest
import litellm
from litellm import completion
from litellm._logging import verbose_logger
from litellm.integrations.datadog.datadog import *
from datetime import datetime, timedelta
from litellm.types.utils import (
StandardLoggingPayload,
StandardLoggingModelInformation,
StandardLoggingMetadata,
StandardLoggingHiddenParams,
)
verbose_logger.setLevel(logging.DEBUG)
def create_standard_logging_payload() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!"}],
response={"choices": [{"message": {"content": "Hi there!"}}]},
error_str=None,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
system_prompt=None,
)
def create_standard_logging_payload_with_long_content() -> StandardLoggingPayload:
return StandardLoggingPayload(
id="test_id",
call_type="completion",
response_cost=0.1,
response_cost_failure_debug_info=None,
status="success",
total_tokens=30,
prompt_tokens=20,
completion_tokens=10,
startTime=1234567890.0,
endTime=1234567891.0,
completionStartTime=1234567890.5,
model_map_information=StandardLoggingModelInformation(
model_map_key="gpt-5-mini", model_map_value=None
),
model="gpt-5-mini",
model_id="model-123",
model_group="openai-gpt",
api_base="https://api.openai.com",
metadata=StandardLoggingMetadata(
user_api_key_hash="test_hash",
user_api_key_org_id=None,
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
spend_logs_metadata=None,
requester_ip_address="127.0.0.1",
requester_metadata=None,
),
cache_hit=False,
cache_key=None,
saved_cache_cost=0.0,
request_tags=[],
end_user=None,
requester_ip_address="127.0.0.1",
messages=[{"role": "user", "content": "Hello, world!" * 80000}],
response={"choices": [{"message": {"content": "Hi there!" * 80000}}]},
error_str="error_str" * 80000,
model_parameters={"stream": True},
hidden_params=StandardLoggingHiddenParams(
model_id="model-123",
cache_key=None,
api_base="https://api.openai.com",
response_cost="0.1",
additional_headers=None,
),
system_prompt=None,
)

42
tests/test_litellm/litellm_core_utils/test_litellm_logging.py
View file

@ -5782,3 +5782,45 @@ def test_failure_handler_helper_fn_builds_payload_once_per_exception():
other_exc = _raise_and_catch(_ClientError(status_code=429, message="rate limited"))
obj._failure_handler_helper_fn(exception=other_exc, traceback_exception="")
assert obj.model_call_details["standard_logging_object"] is not first_payload
def test_get_system_prompt_from_kwargs():
"""Test get_system_prompt_from_kwargs extracts system prompt from all known kwarg sources."""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
# None kwargs → None
assert StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(None) is None
# Empty kwargs → None
assert StandardLoggingPayloadSetup.get_system_prompt_from_kwargs({}) is None
# String system prompt (Anthropic /messages form)
assert (
StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
{"system": "Be helpful"}
)
== "Be helpful"
)
# List-of-content-blocks system prompt (Anthropic prompt caching form)
blocks = [{"type": "text", "text": "Be helpful", "cache_control": {"type": "ephemeral"}}]
assert (
StandardLoggingPayloadSetup.get_system_prompt_from_kwargs({"system": blocks})
== blocks
)
# instructions key (OpenAI Responses API) takes priority over system
assert (
StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
{"instructions": "You are a coder", "system": "ignored"}
)
== "You are a coder"
)
# system_instructions key (Vertex Gemini) takes highest priority
assert (
StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
{"system_instructions": "vertex prompt", "instructions": "ignored", "system": "also ignored"}
)
== "vertex prompt"
)