mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-05 08:07:05 +00:00
feat(datadog_llm_obs): cost tag dimensions, router decision fields, reasoning token metric, redaction gating (#39402)
* feat(datadog_llm_obs): cost tag dimensions, router decision fields, reasoning token metric, redaction gating * test(datadog_llm_obs): satisfy test quality gate * fix: forward integer parent_id as its string form * fix(datadog): sanitize redacted message roles * fix(datadog): keep the A2A agent role on redacted spans * fix(datadog): merge current staging budget * style(datadog): format redaction tests * fix(datadog): handle malformed redacted roles * test(datadog): put the test quality suppression on the reported line Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng-berri
GitHub
parent
64e45a069d
commit
291e84e565
5 changed files with 577 additions and 41 deletions
|
|
@ -57,7 +57,7 @@
|
|||
"limit": 5601
|
||||
},
|
||||
"reportMissingTypeArgument": {
|
||||
"limit": 15290
|
||||
"limit": 15288
|
||||
},
|
||||
"reportMissingTypeStubs": {
|
||||
"limit": 40
|
||||
|
|
@ -105,7 +105,7 @@
|
|||
"limit": 109
|
||||
},
|
||||
"reportUnknownMemberType": {
|
||||
"limit": 38332
|
||||
"limit": 38324
|
||||
},
|
||||
"reportUnknownParameterType": {
|
||||
"limit": 19625
|
||||
|
|
@ -123,7 +123,7 @@
|
|||
"limit": 4
|
||||
},
|
||||
"reportUnnecessaryIsInstance": {
|
||||
"limit": 823
|
||||
"limit": 819
|
||||
},
|
||||
"reportUntypedBaseClass": {
|
||||
"limit": 0
|
||||
|
|
|
|||
|
|
@ -19,11 +19,13 @@ import httpx
|
|||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm._uuid import uuid
|
||||
from litellm.constants import REDACTED_BY_LITELLM
|
||||
from litellm.integrations.custom_batch_logger import CustomBatchLogger
|
||||
from litellm.integrations.datadog.datadog_handler import (
|
||||
get_datadog_base_url_from_env,
|
||||
get_datadog_service,
|
||||
get_datadog_tags,
|
||||
normalize_datadog_tag_value,
|
||||
)
|
||||
from litellm.integrations.datadog.datadog_mock_client import (
|
||||
create_mock_datadog_client,
|
||||
|
|
@ -34,6 +36,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
|||
convert_content_list_to_str,
|
||||
handle_any_messages_to_chat_completion_str_messages_conversion,
|
||||
)
|
||||
from litellm.litellm_core_utils.redact_messages import should_redact_message_logging
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
|
||||
from litellm.llms.custom_httpx.http_handler import (
|
||||
|
|
@ -43,6 +46,7 @@ from litellm.llms.custom_httpx.http_handler import (
|
|||
from litellm.proxy.spend_tracking.savings import extract_cache_creation_tokens, extract_cache_read_tokens
|
||||
from litellm.types.integrations.datadog_llm_obs import *
|
||||
from litellm.types.utils import (
|
||||
PROMPT_QUOTING_ROUTING_DECISION_FIELDS,
|
||||
CallTypes,
|
||||
StandardLoggingGuardrailInformation,
|
||||
StandardLoggingPayload,
|
||||
|
|
@ -52,6 +56,120 @@ from litellm.types.utils import (
|
|||
_EMPTY_MAPPING: Final[Mapping[str, Any]] = MappingProxyType({})
|
||||
_EMPTY_MESSAGE: Final[Message] = {"role": "", "content": ""}
|
||||
_MAX_PARSED_TOOL_ARGUMENT_CHARS: Final = 256 * 1024
|
||||
_SAFE_REDACTED_MESSAGE_ROLES: Final = frozenset(
|
||||
{"agent", "assistant", "developer", "function", "model", "system", "tool", "user"}
|
||||
)
|
||||
|
||||
_PROMPT_CARRYING_METADATA_FIELDS: Final = frozenset(
|
||||
{
|
||||
"routing_decision",
|
||||
"requester_metadata",
|
||||
"prompt_management_metadata",
|
||||
"mcp_tool_call_metadata",
|
||||
"vector_store_request_metadata",
|
||||
}
|
||||
)
|
||||
|
||||
_ROUTER_SPAN_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
|
||||
{
|
||||
"tier": "router_tier",
|
||||
"cause": "router_cause",
|
||||
"score": "router_score",
|
||||
"escalated": "router_escalated",
|
||||
"signals": "router_signals",
|
||||
"routed_model": "routed_model",
|
||||
}
|
||||
)
|
||||
_ROUTER_DIMENSIONS: Final[tuple[str, ...]] = ("router_tier", "router_cause", "router_escalated", "routed_model")
|
||||
_COST_DIMENSIONS: Final[tuple[str, ...]] = ("team", "user", "key_alias", "model_group", *_ROUTER_DIMENSIONS)
|
||||
|
||||
|
||||
def _metadata_of(standard_logging_payload: StandardLoggingPayload) -> Mapping[str, Any]:
|
||||
metadata: Final = standard_logging_payload.get("metadata")
|
||||
return metadata or _EMPTY_MAPPING
|
||||
|
||||
|
||||
def _router_span_fields(
|
||||
standard_logging_payload: StandardLoggingPayload, redact_prompt_text: bool
|
||||
) -> Mapping[str, object]:
|
||||
"""Flatten the auto-router decision, omitting prompt-quoting fields when redaction is enabled."""
|
||||
routing_decision: Final = _mapping_field(_metadata_of(standard_logging_payload), "routing_decision")
|
||||
if not routing_decision:
|
||||
return _EMPTY_MAPPING
|
||||
escalated: Final = bool(routing_decision.get("escalated") or routing_decision.get("context_escalated"))
|
||||
return MappingProxyType(
|
||||
{
|
||||
_ROUTER_SPAN_FIELDS[record_field]: value
|
||||
for record_field, value in (*routing_decision.items(), ("escalated", escalated))
|
||||
if record_field in _ROUTER_SPAN_FIELDS
|
||||
and value is not None
|
||||
and not (redact_prompt_text and record_field in PROMPT_QUOTING_ROUTING_DECISION_FIELDS)
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _metadata_without_prompt_carriers(standard_logging_metadata: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||
"""The metadata minus the records that quote prompts, tool arguments, tool results, or retrieved text."""
|
||||
return MappingProxyType(
|
||||
{
|
||||
field: value
|
||||
for field, value in standard_logging_metadata.items()
|
||||
if field not in _PROMPT_CARRYING_METADATA_FIELDS
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _redact_messages(messages: Sequence[Message]) -> tuple[Message, ...]:
|
||||
"""Each message's shape with its content replaced and tool payloads dropped; no message is invented."""
|
||||
return tuple(
|
||||
{
|
||||
"role": role if isinstance(role, str) and role in _SAFE_REDACTED_MESSAGE_ROLES else "",
|
||||
"content": REDACTED_BY_LITELLM,
|
||||
}
|
||||
for message in messages
|
||||
for role in (message.get("role", ""),)
|
||||
)
|
||||
|
||||
|
||||
def _cost_dimension_tags(
|
||||
standard_logging_payload: StandardLoggingPayload, router_fields: Mapping[str, object]
|
||||
) -> tuple[str, ...]:
|
||||
"""The dimensions LLM Obs breaks token and cost metrics down by, as span tags."""
|
||||
metadata: Final = _metadata_of(standard_logging_payload)
|
||||
dimensions: Final = (
|
||||
("user", metadata.get("user_api_key_user_id")),
|
||||
("key_alias", metadata.get("user_api_key_alias")),
|
||||
("model_group", standard_logging_payload.get("model_group")),
|
||||
*((dimension, router_fields.get(dimension)) for dimension in _ROUTER_DIMENSIONS),
|
||||
)
|
||||
return tuple(
|
||||
f"{key}:{normalized}"
|
||||
for key, value in dimensions
|
||||
if value is not None and (normalized := normalize_datadog_tag_value(value)) != ""
|
||||
)
|
||||
|
||||
|
||||
def _declared_cost_tags(span_tags: Sequence[str]) -> tuple[str, ...]:
|
||||
"""Declare only cost dimensions carrying a value on this span."""
|
||||
present: Final = frozenset(key for tag in span_tags if (key := tag.partition(":")[0]) and tag.partition(":")[2])
|
||||
return tuple(dimension for dimension in _COST_DIMENSIONS if dimension in present)
|
||||
|
||||
|
||||
def _reasoning_output_tokens(usage_object: Mapping[str, Any] | None) -> float:
|
||||
"""The provider's reasoning-token count, from either the chat or the responses spelling."""
|
||||
if usage_object is None:
|
||||
return 0.0
|
||||
return next(
|
||||
(
|
||||
float(reasoning_tokens)
|
||||
for details_field in ("completion_tokens_details", "output_tokens_details")
|
||||
if isinstance(
|
||||
reasoning_tokens := _mapping_field(usage_object, details_field).get("reasoning_tokens"), (int, float)
|
||||
)
|
||||
and not isinstance(reasoning_tokens, bool)
|
||||
),
|
||||
0.0,
|
||||
)
|
||||
|
||||
|
||||
def _mapping_field(source: Mapping[str, Any], key: str) -> Mapping[str, Any]:
|
||||
|
|
@ -316,12 +434,12 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
dict_datadog_llm_obs_params: dict = {}
|
||||
if litellm.datadog_llm_observability_params is not None:
|
||||
if isinstance(litellm.datadog_llm_observability_params, DatadogLLMObsInitParams):
|
||||
dict_datadog_llm_obs_params = litellm.datadog_llm_observability_params.model_dump()
|
||||
dict_datadog_llm_obs_params = litellm.datadog_llm_observability_params.model_dump(exclude_unset=True)
|
||||
elif isinstance(litellm.datadog_llm_observability_params, dict):
|
||||
# only allow params that are of DatadogLLMObsInitParams
|
||||
dict_datadog_llm_obs_params = DatadogLLMObsInitParams(
|
||||
**litellm.datadog_llm_observability_params
|
||||
).model_dump()
|
||||
).model_dump(exclude_unset=True)
|
||||
return dict_datadog_llm_obs_params
|
||||
|
||||
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
|
||||
|
|
@ -410,25 +528,40 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
if standard_logging_payload is None:
|
||||
raise Exception("DataDogLLMObs: standard_logging_object is not set")
|
||||
|
||||
metadata: Final = kwargs.get("litellm_params", {}).get("metadata", {})
|
||||
raw_metadata: Final = kwargs.get("litellm_params", {}).get("metadata", {})
|
||||
metadata: Final = raw_metadata if isinstance(raw_metadata, dict) else {}
|
||||
redact_payload: Final = self._payload_logging_is_off(kwargs)
|
||||
|
||||
input_meta: Final = InputMeta(messages=_to_dd_messages(standard_logging_payload["messages"]))
|
||||
input_messages: Final = _to_dd_messages(standard_logging_payload.get("messages"))
|
||||
output_messages: Final = self._get_response_messages(
|
||||
standard_logging_payload=standard_logging_payload,
|
||||
call_type=standard_logging_payload.get("call_type"),
|
||||
)
|
||||
input_meta: Final = InputMeta(messages=_redact_messages(input_messages) if redact_payload else input_messages)
|
||||
output_meta: Final = OutputMeta(
|
||||
messages=self._get_response_messages(
|
||||
standard_logging_payload=standard_logging_payload,
|
||||
call_type=standard_logging_payload.get("call_type"),
|
||||
)
|
||||
messages=_redact_messages(output_messages) if redact_payload else output_messages
|
||||
)
|
||||
|
||||
error_info: Final = self._assemble_error_info(standard_logging_payload)
|
||||
|
||||
metadata_parent_id: str | None = None
|
||||
if isinstance(metadata, dict):
|
||||
metadata_parent_id = metadata.get("parent_id")
|
||||
raw_parent_id: Final = metadata.get("parent_id")
|
||||
metadata_parent_id: Final[str | None] = str(raw_parent_id) if raw_parent_id else None
|
||||
|
||||
tool_definitions: Final = _to_dd_tool_definitions(standard_logging_payload.get("model_parameters"))
|
||||
tool_definitions: Final = (
|
||||
() if redact_payload else _to_dd_tool_definitions(standard_logging_payload.get("model_parameters"))
|
||||
)
|
||||
span_kind: Final = self._get_datadog_span_kind(standard_logging_payload.get("call_type"), metadata_parent_id)
|
||||
payload_metadata: Final = self._get_dd_llm_obs_payload_metadata(standard_logging_payload)
|
||||
router_fields: Final = _router_span_fields(standard_logging_payload, redact_prompt_text=redact_payload)
|
||||
span_tags: Final = [
|
||||
*get_datadog_tags(standard_logging_object=standard_logging_payload),
|
||||
*_cost_dimension_tags(standard_logging_payload, router_fields),
|
||||
]
|
||||
payload_metadata: Final = self._get_dd_llm_obs_payload_metadata(
|
||||
standard_logging_payload,
|
||||
router_fields=router_fields,
|
||||
cost_tags=_declared_cost_tags(span_tags),
|
||||
redact_prompt_text=redact_payload,
|
||||
)
|
||||
|
||||
meta: Final[Meta] = {
|
||||
"kind": span_kind,
|
||||
|
|
@ -451,7 +584,7 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
duration=int((end_time - start_time).total_seconds() * 1e9),
|
||||
metrics=metrics,
|
||||
status="error" if error_info else "ok",
|
||||
tags=get_datadog_tags(standard_logging_object=standard_logging_payload),
|
||||
tags=span_tags,
|
||||
)
|
||||
|
||||
apm_trace_id: Final = self._get_apm_trace_id()
|
||||
|
|
@ -497,6 +630,13 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
)
|
||||
return error_info
|
||||
|
||||
def _payload_logging_is_off(self, kwargs: Mapping[str, Any]) -> bool:
|
||||
return (
|
||||
bool(self.turn_off_message_logging)
|
||||
or self.message_logging is not True
|
||||
or should_redact_message_logging(dict(kwargs))
|
||||
)
|
||||
|
||||
def _assemble_metrics(self, standard_logging_payload: StandardLoggingPayload) -> LLMMetrics:
|
||||
"""
|
||||
Build the span metrics, including the prompt-cache counts LLM Obs charts cache savings from.
|
||||
|
|
@ -513,10 +653,11 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
total_cost: Final = float(standard_logging_payload.get("response_cost", 0))
|
||||
time_to_first_token: Final = self._get_time_to_first_token_seconds(standard_logging_payload)
|
||||
|
||||
raw_usage: Final = (standard_logging_payload.get("metadata") or {}).get("usage_object")
|
||||
raw_usage: Final = _metadata_of(standard_logging_payload).get("usage_object")
|
||||
usage_object: Final = raw_usage if isinstance(raw_usage, dict) else None
|
||||
cache_read: Final = float(extract_cache_read_tokens(usage_object))
|
||||
cache_write: Final = float(extract_cache_creation_tokens(usage_object))
|
||||
reasoning_output_tokens: Final = _reasoning_output_tokens(usage_object)
|
||||
|
||||
metrics: Final[LLMMetrics] = {
|
||||
"input_tokens": prompt_tokens,
|
||||
|
|
@ -533,6 +674,7 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
if cache_read or cache_write
|
||||
else {}
|
||||
),
|
||||
**({"reasoning_output_tokens": reasoning_output_tokens} if reasoning_output_tokens else {}),
|
||||
}
|
||||
return metrics
|
||||
|
||||
|
|
@ -707,11 +849,21 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
# Default fallback for unknown or passthrough operations
|
||||
return "llm"
|
||||
|
||||
def _get_dd_llm_obs_payload_metadata(self, standard_logging_payload: StandardLoggingPayload) -> dict[str, object]:
|
||||
def _get_dd_llm_obs_payload_metadata(
|
||||
self,
|
||||
standard_logging_payload: StandardLoggingPayload,
|
||||
router_fields: Mapping[str, object] | None = None,
|
||||
cost_tags: Sequence[str] = (),
|
||||
redact_prompt_text: bool = False,
|
||||
) -> dict[str, object]:
|
||||
"""
|
||||
Fields to track in DD LLM Observability metadata from litellm standard logging payload
|
||||
"""
|
||||
_metadata: Final[dict[str, object]] = {
|
||||
raw_metadata: Final = _metadata_of(standard_logging_payload)
|
||||
standard_logging_metadata: Final = (
|
||||
_metadata_without_prompt_carriers(raw_metadata) if redact_prompt_text else raw_metadata
|
||||
)
|
||||
return {
|
||||
"model_name": standard_logging_payload.get("model", "unknown"),
|
||||
"model_provider": standard_logging_payload.get("custom_llm_provider", "unknown"),
|
||||
"id": standard_logging_payload.get("id", "unknown"),
|
||||
|
|
@ -719,26 +871,21 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
"cache_hit": standard_logging_payload.get("cache_hit", "unknown"),
|
||||
"cache_key": standard_logging_payload.get("cache_key", "unknown"),
|
||||
"saved_cache_cost": standard_logging_payload.get("saved_cache_cost", 0),
|
||||
"guardrail_information": standard_logging_payload.get("guardrail_information", None),
|
||||
"guardrail_information": (
|
||||
None if redact_prompt_text else standard_logging_payload.get("guardrail_information", None)
|
||||
),
|
||||
"is_streamed_request": self._get_stream_value_from_payload(standard_logging_payload),
|
||||
"latency_metrics": dict(self._get_latency_metrics(standard_logging_payload)),
|
||||
"spend_metrics": dict(self._get_spend_metrics(standard_logging_payload)),
|
||||
**standard_logging_metadata,
|
||||
**(router_fields or _EMPTY_MAPPING),
|
||||
**(
|
||||
{"_dd": {**_mapping_field(standard_logging_metadata, "_dd"), "cost_tags": list(cost_tags)}}
|
||||
if cost_tags
|
||||
else _EMPTY_MAPPING
|
||||
),
|
||||
}
|
||||
|
||||
#########################################################
|
||||
# Add latency metrics to metadata
|
||||
#########################################################
|
||||
latency_metrics: Final = self._get_latency_metrics(standard_logging_payload)
|
||||
_metadata.update({"latency_metrics": dict(latency_metrics)})
|
||||
|
||||
#########################################################
|
||||
# Add spend metrics to metadata
|
||||
#########################################################
|
||||
spend_metrics: Final = self._get_spend_metrics(standard_logging_payload)
|
||||
_metadata.update({"spend_metrics": dict(spend_metrics)})
|
||||
|
||||
_standard_logging_metadata: Final[dict] = dict(standard_logging_payload.get("metadata", {})) or {}
|
||||
_metadata.update(_standard_logging_metadata)
|
||||
return _metadata
|
||||
|
||||
def _get_latency_metrics(self, standard_logging_payload: StandardLoggingPayload) -> DDLLMObsLatencyMetrics:
|
||||
"""
|
||||
Get the latency metrics from the standard logging payload
|
||||
|
|
@ -808,7 +955,7 @@ class DataDogLLMObsLogger(CustomBatchLogger):
|
|||
spend_metrics["response_cost"] = standard_logging_payload.get("response_cost", 0.0)
|
||||
|
||||
# Get budget information from metadata
|
||||
metadata: Final = standard_logging_payload.get("metadata", {})
|
||||
metadata: Final = _metadata_of(standard_logging_payload)
|
||||
|
||||
# API key max budget
|
||||
user_api_key_max_budget: Final = metadata.get("user_api_key_max_budget")
|
||||
|
|
|
|||
|
|
@ -86,6 +86,7 @@ class LLMMetrics(TypedDict, total=False):
|
|||
cache_read_input_tokens: ReadOnly[float]
|
||||
cache_write_input_tokens: ReadOnly[float]
|
||||
non_cached_input_tokens: ReadOnly[float]
|
||||
reasoning_output_tokens: ReadOnly[float]
|
||||
|
||||
|
||||
class LLMObsPayload(TypedDict, total=False):
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ from unittest.mock import patch
|
|||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.datadog.datadog_llm_obs import DataDogLLMObsLogger
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
|
||||
|
|
@ -55,15 +56,22 @@ def build_payload(
|
|||
response_message: dict[str, Any] | None = None,
|
||||
usage_object: dict[str, Any] | None = None,
|
||||
model_parameters: dict[str, Any] | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
model_group: str | None = None,
|
||||
prompt_tokens: int = 4447,
|
||||
) -> dict[str, Any]:
|
||||
standard_logging_metadata: dict[str, Any] = {
|
||||
**(metadata or {}),
|
||||
**({"usage_object": usage_object} if usage_object is not None else {}),
|
||||
}
|
||||
return {
|
||||
"standard_logging_object": {
|
||||
"call_type": "acompletion",
|
||||
"messages": [{"role": "user", "content": "hi"}] if messages is NOT_GIVEN else messages,
|
||||
"response": {"choices": [{"message": response_message or {"role": "assistant", "content": "hello"}}]},
|
||||
"model_parameters": model_parameters or {},
|
||||
"metadata": {"usage_object": usage_object} if usage_object is not None else {},
|
||||
"metadata": standard_logging_metadata,
|
||||
"model_group": model_group,
|
||||
"prompt_tokens": prompt_tokens,
|
||||
"completion_tokens": 507,
|
||||
"total_tokens": prompt_tokens + 507,
|
||||
|
|
@ -244,6 +252,43 @@ def test_no_cache_keys_when_the_provider_reports_no_caching(logger: DataDogLLMOb
|
|||
assert "non_cached_input_tokens" not in payload["metrics"]
|
||||
|
||||
|
||||
def test_reasoning_tokens_are_reported_as_span_metrics(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, usage_object={"completion_tokens_details": {"reasoning_tokens": 128}})
|
||||
|
||||
assert payload["metrics"]["reasoning_output_tokens"] == 128.0
|
||||
|
||||
|
||||
def test_responses_reasoning_tokens_are_reported_as_span_metrics(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, usage_object={"output_tokens_details": {"reasoning_tokens": 64}})
|
||||
|
||||
assert payload["metrics"]["reasoning_output_tokens"] == 64.0
|
||||
|
||||
|
||||
def test_zero_reasoning_tokens_are_not_reported(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, usage_object={"completion_tokens_details": {"reasoning_tokens": 0}})
|
||||
|
||||
assert "reasoning_output_tokens" not in payload["metrics"]
|
||||
|
||||
|
||||
def test_reasoning_tokens_come_from_the_spelling_that_reports_them(logger: DataDogLLMObsLogger) -> None:
|
||||
"""A chat-details mapping without the count must not shadow the responses spelling that has it."""
|
||||
payload = build(
|
||||
logger,
|
||||
usage_object={
|
||||
"completion_tokens_details": {"accepted_prediction_tokens": 5},
|
||||
"output_tokens_details": {"reasoning_tokens": 64},
|
||||
},
|
||||
)
|
||||
|
||||
assert payload["metrics"]["reasoning_output_tokens"] == 64.0
|
||||
|
||||
|
||||
def test_boolean_reasoning_tokens_are_not_a_count(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, usage_object={"completion_tokens_details": {"reasoning_tokens": True}})
|
||||
|
||||
assert "reasoning_output_tokens" not in payload["metrics"]
|
||||
|
||||
|
||||
def test_tool_definitions_are_sent_on_meta(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, model_parameters={"tools": [TOOL_DEFINITION]})
|
||||
|
||||
|
|
@ -256,6 +301,340 @@ def test_tool_definitions_are_sent_on_meta(logger: DataDogLLMObsLogger) -> None:
|
|||
]
|
||||
|
||||
|
||||
def test_cost_tags_include_present_categories_and_dimensions(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(
|
||||
logger,
|
||||
metadata={
|
||||
"user_api_key_user_id": "User 42",
|
||||
"user_api_key_alias": "Primary Key",
|
||||
"team_alias": "Platform",
|
||||
"routing_decision": {
|
||||
"tier": "premium",
|
||||
"cause": "high_complexity",
|
||||
"score": 0.91,
|
||||
"escalated": True,
|
||||
"signals": ["long prompt"],
|
||||
"routed_model": "openai/gpt-5",
|
||||
},
|
||||
},
|
||||
model_group="premium-models",
|
||||
)
|
||||
|
||||
assert payload["tags"][-8:] == [
|
||||
"team:platform",
|
||||
"user:user_42",
|
||||
"key_alias:primary_key",
|
||||
"model_group:premium-models",
|
||||
"router_tier:premium",
|
||||
"router_cause:high_complexity",
|
||||
"router_escalated:true",
|
||||
"routed_model:openai/gpt-5",
|
||||
]
|
||||
assert payload["meta"]["metadata"]["_dd"]["cost_tags"] == [
|
||||
"team",
|
||||
"user",
|
||||
"key_alias",
|
||||
"model_group",
|
||||
"router_tier",
|
||||
"router_cause",
|
||||
"router_escalated",
|
||||
"routed_model",
|
||||
]
|
||||
|
||||
|
||||
def test_missing_cost_tag_values_are_not_declared(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, metadata={"team_alias": "Platform"})
|
||||
|
||||
assert payload["meta"]["metadata"]["_dd"]["cost_tags"] == ["team"]
|
||||
assert not any(tag.startswith(("user:", "key_alias:", "model_group:")) for tag in payload["tags"])
|
||||
|
||||
|
||||
def test_values_that_normalize_to_empty_are_not_tagged_or_declared(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, metadata={"user_api_key_user_id": "___", "user_api_key_alias": "!!!"}, model_group="tier-1")
|
||||
|
||||
assert not any(tag in ("user:", "key_alias:") for tag in payload["tags"])
|
||||
assert payload["meta"]["metadata"]["_dd"]["cost_tags"] == ["model_group"]
|
||||
|
||||
|
||||
def test_a_valueless_tag_from_the_shared_builder_is_not_declared(logger: DataDogLLMObsLogger) -> None:
|
||||
"""The team tag comes from the shared builder, which emits it bare when the alias normalizes away."""
|
||||
payload = build(logger, metadata={"team_alias": "!!!"}, model_group="tier-1")
|
||||
|
||||
assert "team:" in payload["tags"]
|
||||
assert payload["meta"]["metadata"]["_dd"]["cost_tags"] == ["model_group"]
|
||||
|
||||
|
||||
def test_router_fields_are_flattened(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(
|
||||
logger,
|
||||
metadata={
|
||||
"routing_decision": {
|
||||
"tier": "premium",
|
||||
"cause": "high_complexity",
|
||||
"score": 0.91,
|
||||
"escalated": True,
|
||||
"signals": ["secret prompt text"],
|
||||
"routed_model": "openai/gpt-5",
|
||||
}
|
||||
},
|
||||
model_group="premium-models",
|
||||
)
|
||||
|
||||
assert payload["meta"]["metadata"]["router_tier"] == "premium"
|
||||
assert payload["meta"]["metadata"]["router_cause"] == "high_complexity"
|
||||
assert payload["meta"]["metadata"]["router_score"] == 0.91
|
||||
assert payload["meta"]["metadata"]["router_escalated"] is True
|
||||
assert payload["meta"]["metadata"]["router_signals"] == ["secret prompt text"]
|
||||
assert payload["meta"]["metadata"]["routed_model"] == "openai/gpt-5"
|
||||
|
||||
|
||||
def test_a_context_escalated_route_reports_as_escalated(logger: DataDogLLMObsLogger) -> None:
|
||||
"""The router records a size-driven escalation under its own key, and it is still an escalation."""
|
||||
payload = build(logger, metadata={"routing_decision": {"tier": "premium", "context_escalated": True}})
|
||||
|
||||
assert payload["meta"]["metadata"]["router_escalated"] is True
|
||||
assert "router_escalated:true" in payload["tags"]
|
||||
|
||||
|
||||
def test_a_routed_request_that_did_not_escalate_reports_false(logger: DataDogLLMObsLogger) -> None:
|
||||
"""Without this the escalation dimension is absent on ordinary traffic, so nothing can group by it."""
|
||||
payload = build(logger, metadata={"routing_decision": {"tier": "simple", "cause": "heuristic_scorer"}})
|
||||
|
||||
assert payload["meta"]["metadata"]["router_escalated"] is False
|
||||
assert "router_escalated:false" in payload["tags"]
|
||||
assert "router_escalated" in payload["meta"]["metadata"]["_dd"]["cost_tags"]
|
||||
|
||||
|
||||
def test_a_request_that_never_reached_a_router_has_no_router_fields(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build(logger, model_group="premium-models")
|
||||
|
||||
assert "router_escalated" not in payload["meta"]["metadata"]
|
||||
assert not any(tag.startswith("router_") for tag in payload["tags"])
|
||||
|
||||
|
||||
def test_redacted_payload_keeps_metrics_and_removes_sensitive_fields(logger: DataDogLLMObsLogger) -> None:
|
||||
payload = build_payload(
|
||||
messages=[{"role": "user", "content": "secret prompt"}],
|
||||
response_message={"role": "assistant", "content": "secret response"},
|
||||
usage_object={"prompt_tokens_details": {"cached_tokens": 128}},
|
||||
metadata={"routing_decision": {"tier": "premium", "signals": ["secret prompt text"]}},
|
||||
model_parameters={"tools": [TOOL_DEFINITION]},
|
||||
)
|
||||
with patch.dict(os.environ, {"DD_API_KEY": "k", "DD_SITE": "us5.datadoghq.com"}, clear=True):
|
||||
with patch("asyncio.create_task"):
|
||||
redacted_logger = DataDogLLMObsLogger(turn_off_message_logging=True)
|
||||
redacted_payload = redacted_logger.redact_standard_logging_payload_from_model_call_details(payload)
|
||||
result = json.loads(
|
||||
safe_dumps(
|
||||
redacted_logger.create_llm_obs_payload(
|
||||
redacted_payload, datetime(2026, 9, 1, 12, 0, 0), datetime(2026, 9, 1, 12, 0, 2)
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
assert result["meta"]["input"]["messages"][0]["content"] == "redacted-by-litellm"
|
||||
assert result["meta"]["output"]["messages"][0]["content"] == "redacted-by-litellm"
|
||||
assert result["meta"]["metadata"]["router_tier"] == "premium"
|
||||
assert "router_signals" not in result["meta"]["metadata"]
|
||||
assert "routing_decision" not in result["meta"]["metadata"]
|
||||
assert "tool_definitions" not in result["meta"]
|
||||
assert result["metrics"]["cache_read_input_tokens"] == 128.0
|
||||
assert result["metrics"]["total_cost"] == 0.02
|
||||
|
||||
|
||||
def test_redaction_drops_the_routing_record_carried_in_metadata(logger: DataDogLLMObsLogger) -> None:
|
||||
"""The whole routing record rides along in metadata, so dropping the flat copy alone leaks the prompt."""
|
||||
with patch.dict(os.environ, {"DD_API_KEY": "k", "DD_SITE": "us5.datadoghq.com"}, clear=True):
|
||||
with patch("asyncio.create_task"):
|
||||
redacted_logger = DataDogLLMObsLogger(turn_off_message_logging=True)
|
||||
result = json.loads(
|
||||
safe_dumps(
|
||||
redacted_logger.create_llm_obs_payload(
|
||||
build_payload(
|
||||
metadata={
|
||||
"routing_decision": {
|
||||
"tier": "premium",
|
||||
"cause": "keyword_rule",
|
||||
"signals": ["secret prompt text"],
|
||||
"matched_keyword": "secret keyword",
|
||||
"escalation_keyword": "secret escalation",
|
||||
}
|
||||
}
|
||||
),
|
||||
datetime(2026, 9, 1, 12, 0, 0),
|
||||
datetime(2026, 9, 1, 12, 0, 2),
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
assert "routing_decision" not in result["meta"]["metadata"]
|
||||
assert result["meta"]["metadata"]["router_tier"] == "premium"
|
||||
assert result["meta"]["metadata"]["router_cause"] == "keyword_rule"
|
||||
assert "secret" not in safe_dumps(result["meta"]["metadata"])
|
||||
|
||||
|
||||
def test_a_failure_span_redacts_its_messages(logger: DataDogLLMObsLogger) -> None:
|
||||
"""The redaction hook only runs on success, so the failure span has to redact for itself."""
|
||||
failed = build_payload(messages=[{"role": "user", "content": "secret prompt"}])
|
||||
failed["standard_logging_object"]["status"] = "failure"
|
||||
failed["standard_logging_object"]["response"] = None
|
||||
failed["standard_logging_object"]["error_information"] = {"error_message": "boom", "error_class": "BadRequestError"}
|
||||
with patch.dict(os.environ, {"DD_API_KEY": "k", "DD_SITE": "us5.datadoghq.com"}, clear=True):
|
||||
with patch("asyncio.create_task"):
|
||||
redacted_logger = DataDogLLMObsLogger(turn_off_message_logging=True)
|
||||
result = json.loads(
|
||||
safe_dumps(
|
||||
redacted_logger.create_llm_obs_payload(
|
||||
failed, datetime(2026, 9, 1, 12, 0, 0), datetime(2026, 9, 1, 12, 0, 2)
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
assert result["meta"]["input"]["messages"] == [{"role": "user", "content": "redacted-by-litellm"}]
|
||||
assert result["meta"]["output"]["messages"] == []
|
||||
assert result["status"] == "error"
|
||||
|
||||
|
||||
def test_excluding_messages_from_the_logging_payload_still_ships_the_span(logger: DataDogLLMObsLogger) -> None:
|
||||
"""`standard_logging_payload_excluded_fields` deletes the key, and a span with no prompt is still a span."""
|
||||
payload = build_payload()
|
||||
del payload["standard_logging_object"]["messages"]
|
||||
|
||||
span = json.loads(
|
||||
safe_dumps(
|
||||
logger.create_llm_obs_payload(payload, datetime(2026, 9, 1, 12, 0, 0), datetime(2026, 9, 1, 12, 0, 2))
|
||||
)
|
||||
)
|
||||
|
||||
assert span["meta"]["input"]["messages"] == []
|
||||
assert span["metrics"]["total_cost"] == 0.02
|
||||
|
||||
|
||||
def test_an_explicit_redaction_setting_survives_the_global_params(logger: DataDogLLMObsLogger) -> None:
|
||||
"""Global params carry defaults for keys the operator never set, and those must not win."""
|
||||
with patch.dict(os.environ, {"DD_API_KEY": "k", "DD_SITE": "us5.datadoghq.com"}, clear=True):
|
||||
with patch("asyncio.create_task"):
|
||||
with patch.object( # test-quality-ok: the ctor reads this module global with no injection seam
|
||||
litellm, "datadog_llm_observability_params", {}
|
||||
):
|
||||
configured_logger = DataDogLLMObsLogger(
|
||||
turn_off_message_logging=True
|
||||
) # test-quality-ok: verifies ctor setting
|
||||
|
||||
assert configured_logger.turn_off_message_logging is True
|
||||
|
||||
|
||||
def _redacting_logger(
|
||||
**kwargs: Any,
|
||||
) -> DataDogLLMObsLogger: # test-quality-ok: shared test factory accepts init variants
|
||||
with patch.dict(os.environ, {"DD_API_KEY": "k", "DD_SITE": "us5.datadoghq.com"}, clear=True):
|
||||
with patch("asyncio.create_task"):
|
||||
return DataDogLLMObsLogger(**kwargs)
|
||||
|
||||
|
||||
def _span_json(logger_under_test: DataDogLLMObsLogger, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
span = logger_under_test.create_llm_obs_payload(
|
||||
payload, datetime(2026, 9, 1, 12, 0, 0), datetime(2026, 9, 1, 12, 0, 2)
|
||||
)
|
||||
return json.loads(safe_dumps(span))
|
||||
|
||||
|
||||
def test_redaction_keeps_the_conversation_shape_without_its_content() -> None:
|
||||
"""Roles and message count survive so the trace stays legible; contents and tool payloads do not."""
|
||||
result = _span_json(
|
||||
_redacting_logger(turn_off_message_logging=True),
|
||||
build_payload(
|
||||
messages=[
|
||||
{"role": "user", "content": "secret prompt"},
|
||||
{"role": "assistant", "content": None, "tool_calls": [ASSISTANT_TOOL_CALL]},
|
||||
],
|
||||
response_message={"role": "assistant", "content": "secret response"},
|
||||
),
|
||||
)
|
||||
|
||||
assert result["meta"]["input"]["messages"] == [
|
||||
{"role": "user", "content": "redacted-by-litellm"},
|
||||
{"role": "assistant", "content": "redacted-by-litellm"},
|
||||
]
|
||||
assert result["meta"]["output"]["messages"] == [{"role": "assistant", "content": "redacted-by-litellm"}]
|
||||
|
||||
|
||||
def test_redaction_drops_unrecognized_and_malformed_message_roles() -> None:
|
||||
"""Caller-controlled role values must not bypass redaction or crash span creation."""
|
||||
result = _span_json(
|
||||
_redacting_logger(turn_off_message_logging=True),
|
||||
build_payload(
|
||||
messages=[
|
||||
{"role": "SECRET-39402", "content": "hello"},
|
||||
{"role": ["SECRET-39402"], "content": "hello"},
|
||||
{"role": {"secret": "SECRET-39402"}, "content": "hello"},
|
||||
{"role": "agent", "content": "hello"},
|
||||
]
|
||||
),
|
||||
)
|
||||
|
||||
assert result["meta"]["input"]["messages"] == [
|
||||
{"role": "", "content": "redacted-by-litellm"},
|
||||
{"role": "", "content": "redacted-by-litellm"},
|
||||
{"role": "", "content": "redacted-by-litellm"},
|
||||
{"role": "agent", "content": "redacted-by-litellm"},
|
||||
]
|
||||
assert "SECRET-39402" not in safe_dumps(result)
|
||||
|
||||
|
||||
def test_the_deprecated_message_logging_flag_engages_the_same_redaction() -> None:
|
||||
"""The platform redacts for `message_logging is not True`, so this callback's own gate must agree."""
|
||||
result = _span_json(
|
||||
_redacting_logger(message_logging=False),
|
||||
build_payload(
|
||||
messages=[{"role": "user", "content": "secret prompt"}],
|
||||
model_parameters={"tools": [TOOL_DEFINITION]},
|
||||
metadata={"routing_decision": {"tier": "premium", "signals": ["secret prompt text"]}},
|
||||
),
|
||||
)
|
||||
|
||||
assert result["meta"]["input"]["messages"] == [{"role": "user", "content": "redacted-by-litellm"}]
|
||||
assert "tool_definitions" not in result["meta"]
|
||||
assert "routing_decision" not in result["meta"]["metadata"]
|
||||
|
||||
|
||||
def test_a_truthy_redaction_setting_redacts_like_the_shared_hook() -> None:
|
||||
"""The shared hook redacts on truthiness, so a config-provided string must not half-redact the span."""
|
||||
result = _span_json(
|
||||
_redacting_logger(turn_off_message_logging="yes"),
|
||||
build_payload(messages=[{"role": "user", "content": "secret prompt"}]),
|
||||
)
|
||||
|
||||
assert result["meta"]["input"]["messages"] == [{"role": "user", "content": "redacted-by-litellm"}]
|
||||
|
||||
|
||||
def test_redaction_drops_every_prompt_carrying_metadata_record(logger: DataDogLLMObsLogger) -> None:
|
||||
"""Tool arguments, retrieved text, and the guardrail's copy of the request ride in metadata records too."""
|
||||
sensitive_metadata: dict[str, Any] = {
|
||||
"requester_metadata": {"note": "secret prompt text"},
|
||||
"prompt_management_metadata": {"prompt_id": "p1", "prompt_variables": {"topic": "secret"}},
|
||||
"mcp_tool_call_metadata": {"name": "search", "arguments": {"query": "secret"}},
|
||||
"vector_store_request_metadata": [{"query": "secret"}],
|
||||
}
|
||||
|
||||
def sensitive_payload() -> dict[str, Any]:
|
||||
payload = build_payload(metadata=sensitive_metadata)
|
||||
payload["standard_logging_object"]["guardrail_information"] = [
|
||||
{"guardrail_name": "g", "guardrail_request": {"messages": [{"content": "secret prompt"}]}}
|
||||
]
|
||||
return payload
|
||||
|
||||
redacted = _span_json(_redacting_logger(turn_off_message_logging=True), sensitive_payload())
|
||||
unredacted = _span_json(logger, sensitive_payload())
|
||||
|
||||
assert "secret" not in safe_dumps(redacted["meta"]["metadata"])
|
||||
for record in sensitive_metadata:
|
||||
assert record not in redacted["meta"]["metadata"]
|
||||
assert record in unredacted["meta"]["metadata"]
|
||||
assert redacted["meta"]["metadata"]["guardrail_information"] is None
|
||||
assert unredacted["meta"]["metadata"]["guardrail_information"] is not None
|
||||
|
||||
|
||||
def test_tool_definitions_accept_the_bare_anthropic_shape(logger: DataDogLLMObsLogger) -> None:
|
||||
"""The Anthropic surface declares tools unwrapped, with input_schema instead of parameters."""
|
||||
payload = build(
|
||||
|
|
@ -272,6 +651,15 @@ def test_meta_omits_tool_definitions_when_no_tools_were_offered(logger: DataDogL
|
|||
assert "tool_definitions" not in build(logger)["meta"]
|
||||
|
||||
|
||||
def test_a_ddtrace_integer_parent_id_is_forwarded_as_its_string(logger: DataDogLLMObsLogger) -> None:
|
||||
"""ddtrace hands span ids as ints; dropping them detaches the span from its APM trace."""
|
||||
kwargs = build_payload()
|
||||
kwargs["litellm_params"]["metadata"]["parent_id"] = 8675309
|
||||
start = datetime(2026, 9, 1, 12, 0, 0)
|
||||
span = json.loads(safe_dumps(logger.create_llm_obs_payload(kwargs, start, start + timedelta(seconds=2))))
|
||||
assert span["parent_id"] == "8675309"
|
||||
|
||||
|
||||
def test_unparseable_tool_arguments_are_preserved_rather_than_dropped(logger: DataDogLLMObsLogger) -> None:
|
||||
"""A truncated argument string is still the only record of what the model tried to call."""
|
||||
payload = build(
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"LIT001": {
|
||||
"limit": 22334
|
||||
"limit": 22330
|
||||
},
|
||||
"LIT002": {
|
||||
"limit": 26763
|
||||
|
|
@ -27,7 +27,7 @@
|
|||
"limit": 0
|
||||
},
|
||||
"LIT010": {
|
||||
"limit": 16480
|
||||
"limit": 16478
|
||||
},
|
||||
"LIT011": {
|
||||
"limit": 5520
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue