scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-15 09:12:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 311

Merge e840831a6a into 570f41daad

Browse source
This commit is contained in:
jesus-berri 2026-11-04 01:06:30 +00:00 • committed by GitHub
commit 2807a09552
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 979 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/proxy/_types.py
View file

@ -2864,6 +2864,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
None,
description="If True, stores request messages and responses in spend logs. Default is False.",
)
spend_logs_prompt_storage: Literal["latest_turn"] | None = Field(
None,
description="With store_prompts_in_spend_logs on, 'latest_turn' stores only the current turn of a stateless request (/chat/completions, /v1/messages, and /v1/responses without previous_response_id): the model and the latest user message, with the system prompt, tool definitions and earlier turns dropped. Requests that continue a stored conversation and the response are stored as they are today. Unset stores the full request.",
)
claude_code_marketplace_auth_required: bool = Field(
False,
description="If True, GET /claude-code/marketplace.json requires a LiteLLM key in the Authorization or x-litellm-api-key header (?key= is not accepted) and answers 401 with WWW-Authenticate: Bearer without one, and GET /public/skill_hub returns an empty list. Default is False, which serves the enabled plugins to anyone.",

5
litellm/proxy/proxy_server.py
View file

@ -7296,6 +7296,11 @@ class ProxyConfig:
)
)
if "spend_logs_prompt_storage" in general_settings:
_ = ConfigGeneralSettings.model_validate(
MappingProxyType({"spend_logs_prompt_storage": typed_general_settings["spend_logs_prompt_storage"]})
)
### PKCE MULTI-INSTANCE PREREQUISITE CHECK ###
# PKCE verifiers are stored in redis_usage_cache when available so they can
# be read back by any instance (not just the one that started the auth flow).

212
litellm/proxy/spend_tracking/spend_tracking_utils.py
View file

@ -63,6 +63,7 @@ from litellm.litellm_core_utils.litellm_logging import (
is_valid_sha256_hash,
request_model_access_groups_from_litellm_params,
)
from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import message_field, parts_of
from litellm.litellm_core_utils.ptu_pricing import azure_spillover
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps, strip_null_bytes
from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
@ -99,6 +100,7 @@ from litellm.types.utils import (
VectorStoreSearchResponse,
batch_chat_usage,
)
from litellm.types.vector_stores import VectorStoreResultContent, VectorStoreSearchResult
from litellm.utils import get_end_user_id_for_cost_tracking
if TYPE_CHECKING:
@ -1260,6 +1262,150 @@ _SENSITIVE_REQUEST_BODY_KEYS: Final = frozenset({"secret_fields"})
_REQUEST_BODY_CREDENTIAL_MASKER: Final = SensitiveDataMasker(extra_sensitive_patterns=frozenset({"apikey"}))
_TOOL_INPUT_BLOCK_TYPES: Final = frozenset({"tool_use", "server_tool_use", "mcp_tool_use"})
_TOOL_OUTPUT_BLOCK_TYPES: Final = frozenset({"tool_result", "mcp_tool_result", "function_call_output"})
_LATEST_TURN_STORAGE: Final = "latest_turn"
_STATEFUL_REQUEST_KEYS: Final = frozenset({"previous_response_id", "conversation"})
_LATEST_TURN_KEPT_KEYS: Final = frozenset({"model"})
_RESPONSES_CALL_TYPES: Final = frozenset({"responses", "aresponses"})
_REQUEST_BODY_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
_TURNS_KEYS: Final = ("messages", "input")
_TEXT_BLOCK_TYPES: Final = frozenset({"text", "input_text"})
_OMITTED_BLOCK_PLACEHOLDERS: Final = MappingProxyType(
{
"image": "[image omitted]",
"image_url": "[image omitted]",
"input_image": "[image omitted]",
"file": "[file omitted]",
"input_file": "[file omitted]",
"document": "[file omitted]",
"input_audio": "[audio omitted]",
}
)
def stores_latest_turn_only() -> bool:
from litellm.proxy.proxy_server import general_settings_view
return general_settings_view().get("spend_logs_prompt_storage") == _LATEST_TURN_STORAGE
def stores_full_prompts_in_spend_logs() -> bool:
return should_store_prompts_and_responses_in_spend_logs() and not stores_latest_turn_only()
def _block_type(block: object) -> str | None:
block_type: Final = message_field(block, "type")
return block_type if isinstance(block_type, str) else None
def _content_block_text(block: object) -> str | None:
if isinstance(block, str):
return block
block_type: Final = _block_type(block)
if block_type in _TEXT_BLOCK_TYPES:
text: Final = message_field(block, "text")
return text if isinstance(text, str) else None
return _OMITTED_BLOCK_PLACEHOLDERS.get(block_type) if block_type is not None else None
def _is_tool_output_block(block: object) -> bool:
return _block_type(block) in _TOOL_OUTPUT_BLOCK_TYPES
def _user_content_text(content: object) -> str | None:
if isinstance(content, str):
return content or None
blocks: Final = parts_of(content)
if any(_is_tool_output_block(block) for block in blocks):
return None
texts: Final = tuple(text for text in map(_content_block_text, blocks) if text)
return "\n".join(texts) or None
def _latest_user_message_text(turns: object) -> str | None:
if isinstance(turns, str):
return turns or None
latest: Final = next(reversed(parts_of(turns)), None)
if message_field(latest, "role") != "user":
return None
return _user_content_text(message_field(latest, "content"))
def _continues_stored_conversation(request_body: Mapping[str, object], call_type: object) -> bool:
"""Only `/v1/responses` resumes a stored conversation, so the same keys on a chat body do not exempt it."""
return call_type in _RESPONSES_CALL_TYPES and any(request_body.get(key) for key in _STATEFUL_REQUEST_KEYS)
def _replayed_by_responses_bridge(standard_payload: Mapping[str, object] | None) -> bool:
"""A `/v1/responses` turn served through the chat completions bridge logs a chat-shaped response.
The bridge rebuilds a `previous_response_id` chain from these rows, so they keep their full request.
"""
if standard_payload is None or standard_payload.get("call_type") not in _RESPONSES_CALL_TYPES:
return False
response: Final = standard_payload.get("response")
return isinstance(response, Mapping) and "choices" in response
def _trimmed_turn(request_body: Mapping[str, object], call_type: object) -> Mapping[str, object]:
if _continues_stored_conversation(request_body, call_type):
return request_body
kept: Final = {key: value for key, value in request_body.items() if key in _LATEST_TURN_KEPT_KEYS}
turns_key: Final = next((key for key in _TURNS_KEYS if key in request_body), None)
text: Final = None if turns_key is None else _latest_user_message_text(request_body[turns_key])
if turns_key is None or text is None:
return MappingProxyType(kept)
return MappingProxyType({**kept, turns_key: [{"role": "user", "content": text}]})
def _latest_turn_body(
request_body: Mapping[str, object], call_type: object, originating: object
) -> Mapping[str, object]:
"""A classifier row stores the trimmed originating request, since the composed classifier message quotes the caller's history.
`originating` comes from the logging payload the classifier wrote, never from the client's body.
"""
if not isinstance(originating, Mapping):
return _trimmed_turn(request_body, call_type)
originating_body: Final = _REQUEST_BODY_ADAPTER.validate_python(originating)
model: Final = {key: value for key, value in request_body.items() if key == "model"}
return MappingProxyType(
{
**_trimmed_turn(originating_body, None),
**model,
"originating_request_masked": _trimmed_turn(originating_body, None),
}
)
def _stored_request_body(
request_body: Mapping[str, object], standard_payload: Mapping[str, object] | None
) -> Mapping[str, object]:
if not stores_latest_turn_only() or _replayed_by_responses_bridge(standard_payload):
return request_body
if standard_payload is None:
return _latest_turn_body(request_body, None, None)
return _latest_turn_body(
request_body, standard_payload.get("call_type"), standard_payload.get("originating_request_masked")
)
def _redacted_when_message_logging_is_off(
request_body: Mapping[str, object], litellm_params: Mapping[str, object], kwargs: Mapping[str, object] | None
) -> Mapping[str, object]:
"""Redaction runs on the body that is about to be stored, after any latest_turn selection."""
if kwargs is None:
return request_body
from litellm.litellm_core_utils.redact_messages import perform_redaction, should_redact_message_logging
model_call_details: Final = {
"litellm_params": litellm_params,
"standard_callback_dynamic_params": kwargs.get("standard_callback_dynamic_params"),
}
if not should_redact_message_logging(model_call_details=model_call_details):
return request_body
redacted: Final = _convert_mapping_to_json_serializable(without_classifier_audit(request_body))
perform_redaction(model_call_details=redacted, result=None)
return redacted
def _is_request_body_credential(key: str, value: object) -> bool:
@ -1522,7 +1668,7 @@ def _sanitize_guardrail_information_for_spend_logs(
here to match OTEL's defensive read pattern; otherwise iteration would
yield the dict's keys and crash the whole spend-log write.
"""
if guardrail_information is None or should_store_prompts_and_responses_in_spend_logs():
if guardrail_information is None or stores_full_prompts_in_spend_logs():
return guardrail_information
entries: Final = [guardrail_information] if isinstance(guardrail_information, dict) else guardrail_information
return [_redact_prompt_fields_in_guardrail_entry(entry) for entry in entries if isinstance(entry, dict)]
@ -1600,7 +1746,7 @@ def sanitize_error_information_for_spend_logs(
)
sanitized = cast(dict, {**persisted})
if not should_store_prompts_and_responses_in_spend_logs():
if not stores_full_prompts_in_spend_logs():
for field in ("error_message", "traceback"):
value = sanitized.get(field)
if isinstance(value, str):
@ -1923,25 +2069,15 @@ def _get_proxy_server_request_for_spend_logs_payload(
_request_body = dict(_request_body)
_request_body["tools"] = realtime_tools
# Apply message redaction if turn_off_message_logging is enabled
if kwargs is not None:
from litellm.litellm_core_utils.redact_messages import (
perform_redaction,
should_redact_message_logging,
)
# Build model_call_details dict to check redaction settings
model_call_details: Final = {
"litellm_params": litellm_params,
"standard_callback_dynamic_params": kwargs.get("standard_callback_dynamic_params"),
}
# If redaction is enabled, convert to serializable dict before redacting
if should_redact_message_logging(model_call_details=model_call_details):
_request_body = _convert_mapping_to_json_serializable(without_classifier_audit(_request_body))
perform_redaction(model_call_details=_request_body, result=None)
stored_request_body: Final = _sanitize_proxy_server_request_body(_request_body)
selected_request_body: Final = _stored_request_body(
_request_body,
_REQUEST_BODY_ADAPTER.validate_python(standard_payload)
if isinstance(standard_payload, Mapping)
else None,
)
stored_request_body: Final = _sanitize_proxy_server_request_body(
_redacted_when_message_logging_is_off(selected_request_body, litellm_params, kwargs)
)
_request_body_json_str: Final = safe_dumps(
_stored_request_without_requester_ip_address(stored_request_body)
if drop_requester_ip_address
@ -1963,22 +2099,32 @@ def _get_vector_store_request_for_spend_logs_payload(
"""
If user does not want to store prompts and responses, then remove the content from the vector store request metadata
"""
if should_store_prompts_and_responses_in_spend_logs():
if stores_full_prompts_in_spend_logs():
return vector_store_request_metadata
# if user does not want to store prompts and responses, then remove the content from the vector store request metadata
if vector_store_request_metadata is None:
return None
for vector_store_request in vector_store_request_metadata:
vector_store_search_response: VectorStoreSearchResponse = (
vector_store_request.get("vector_store_search_response") or VectorStoreSearchResponse()
)
response_data = vector_store_search_response.get("data", []) or []
for response_item in response_data:
for content_item in response_item.get("content", []) or []:
if "text" in content_item:
content_item["text"] = REDACTED_BY_LITELM_STRING
return vector_store_request_metadata
return [_redacted_vector_store_request(request) for request in vector_store_request_metadata]
def _redacted_vector_store_content(content: VectorStoreResultContent) -> VectorStoreResultContent:
return {**content, "text": REDACTED_BY_LITELM_STRING} if "text" in content else content
def _redacted_vector_store_result(result: VectorStoreSearchResult) -> VectorStoreSearchResult:
return {**result, "content": [_redacted_vector_store_content(content) for content in result.get("content") or ()]}
def _redacted_vector_store_request(request: StandardLoggingVectorStoreRequest) -> StandardLoggingVectorStoreRequest:
"""A redacted copy, so the shared logging payload other callbacks read keeps its search results."""
response: Final = request.get("vector_store_search_response") or VectorStoreSearchResponse()
return {
**request,
"vector_store_search_response": {
**response,
"data": [_redacted_vector_store_result(result) for result in response.get("data") or ()],
},
}
def _get_response_for_spend_logs_payload(

372
tests/integration/spend/test_spend_log_latest_turn.py Normal file
View file

@ -0,0 +1,372 @@
import json
from collections.abc import Callable
from pathlib import Path
from typing import Final
from uuid import uuid4
import pytest
from integration._support.client import Gateway, eventually, object_value, string_value
from integration._support.process import owned_proxy
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
from litellm.responses.utils import ResponsesAPIRequestUtils
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
MESSAGES: Final = TypeAdapter(list[dict[str, JsonValue]])
OPENAI_MODEL: Final = "openai/gpt-4o-mini"
ANTHROPIC_MODEL: Final = "anthropic/claude-sonnet-4-5-20250929"
PIXEL: Final = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
ROW_COLUMNS: Final = "request_id, proxy_server_request, response, status, total_tokens, model"
def _latest_turn_config(tmp_path: Path) -> Path:
config: Final = tmp_path / f"latest-turn-{uuid4()}.json"
config.write_text(
json.dumps(
{
"model_list": [],
"general_settings": {
"master_key": "os.environ/LITELLM_MASTER_KEY",
"database_url": "os.environ/DATABASE_URL",
"disable_responses_id_security": True,
"store_model_in_db": True,
"store_prompts_in_spend_logs": True,
"spend_logs_prompt_storage": "latest_turn",
"proxy_batch_write_at": 1,
"proxy_batch_polling_interval": 1,
},
}
)
)
return config
def _answering_model_listing(respond: Callable[[Request], Reply]) -> Callable[[Request], Reply]:
def answer(request: Request) -> Reply:
if request.method == "GET":
return Reply(body=b'{"object":"list","data":[]}')
return respond(request)
return answer
def _chat_completion(text: str) -> bytes:
return json.dumps(
{
"id": f"chatcmpl-latest-{uuid4()}",
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 11, "completion_tokens": 3, "total_tokens": 14},
}
).encode()
def _anthropic_message(text: str) -> bytes:
return json.dumps(
{
"id": f"msg-latest-{uuid4()}",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-5-20250929",
"content": [{"type": "text", "text": text}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 3},
}
).encode()
def _native_response(text: str) -> bytes:
identity: Final = f"resp-latest-{uuid4()}"
return json.dumps(
{
"id": identity,
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-4o-mini",
"output": [
{
"id": f"msg_{identity}",
"type": "message",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": text, "annotations": []}],
}
],
"usage": {"input_tokens": 11, "output_tokens": 3, "total_tokens": 14},
}
).encode()
def _row(proxy: Gateway, where: str, value: str) -> dict[str, JsonValue]:
rows: Final = eventually(
lambda: proxy.rows(f'SELECT {ROW_COLUMNS} FROM "LiteLLM_SpendLogs" WHERE {where}=%s', (value,)),
lambda values: len(values) == 1,
seconds=70,
)
return rows[0]
def _responses_request_id(response_id: str) -> str:
decoded: Final = ResponsesAPIRequestUtils.decode_responses_api_response_id(response_id)
request_id: Final = decoded.get("response_id")
return string_value(request_id) if isinstance(request_id, str) else response_id
def _user_turn(text: str) -> list[JsonValue]:
return [{"role": "user", "content": text}]
def _assert_usage_row(row: dict[str, JsonValue], expected_request: dict[str, JsonValue], answer: str) -> None:
assert object_value(row["proxy_server_request"]) == expected_request, row
assert answer in json.dumps(row["response"]), row
assert row["status"] == "success", row
assert row["total_tokens"] == 14, row
def _last_upstream_body(wire: Wire) -> dict[str, JsonValue]:
posted: Final = [request for request in wire.drain() if request.method == "POST"]
return JSON_OBJECT.validate_json(posted[-1].body)
CHAT_HISTORY: Final = [
{"role": "system", "content": "system prompt must not be stored"},
{"role": "user", "content": "earlier user turn must not be stored"},
{"role": "assistant", "content": "earlier assistant turn must not be stored"},
]
@pytest.mark.parametrize(
("latest", "expected_turns"),
[
pytest.param(
{
"role": "user",
"content": [
{"type": "text", "text": "latest chat question"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{PIXEL}"}},
],
},
{"messages": _user_turn("latest chat question\n[image omitted]")},
id="user-text-and-image",
),
pytest.param(
{"role": "tool", "tool_call_id": "call_1", "content": "tool output must not be stored"},
{},
id="tool-result-turn",
),
],
)
def test_chat_row_keeps_only_the_latest_user_prompt_and_the_response(
gateway: Gateway, tmp_path: Path, latest: dict[str, JsonValue], expected_turns: dict[str, JsonValue]
) -> None:
with (
wire_server(_answering_model_listing(lambda _: Reply(body=_chat_completion("chat answer")))) as wire,
owned_proxy(gateway, tmp_path, {}, config=_latest_turn_config(tmp_path)) as isolated,
isolated.scenario() as scenario,
):
model: Final = scenario.model(model=OPENAI_MODEL, api_base=wire.url, api_key="synthetic-openai-key")
response: Final = isolated.request(
"POST",
"/v1/chat/completions",
{
"model": model,
"messages": [*CHAT_HISTORY, latest],
"tools": [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}],
},
)
assert response.status_code == 200, response.text
row: Final = _row(isolated, "request_id", string_value(JSON_OBJECT.validate_json(response.content)["id"]))
_assert_usage_row(row, {"model": model, **expected_turns}, "chat answer")
@pytest.mark.parametrize(
("latest_content", "expected_turns"),
[
pytest.param(
[
{"type": "text", "text": "latest messages question"},
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": PIXEL}},
],
{"messages": _user_turn("latest messages question\n[image omitted]")},
id="user-text-and-image",
),
pytest.param(
[{"type": "tool_result", "tool_use_id": "toolu_1", "content": "tool output must not be stored"}],
{},
id="tool-result-turn",
),
],
)
def test_messages_row_keeps_only_the_latest_user_prompt_and_the_response(
gateway: Gateway, tmp_path: Path, latest_content: list[JsonValue], expected_turns: dict[str, JsonValue]
) -> None:
with (
wire_server(_answering_model_listing(lambda _: Reply(body=_anthropic_message("messages answer")))) as wire,
owned_proxy(gateway, tmp_path, {}, config=_latest_turn_config(tmp_path)) as isolated,
isolated.scenario() as scenario,
):
model: Final = scenario.model(model=ANTHROPIC_MODEL, api_base=wire.url, api_key="synthetic-anthropic-key")
response: Final = isolated.request(
"POST",
"/v1/messages",
{
"model": model,
"max_tokens": 64,
"system": "system prompt must not be stored",
"messages": [
{"role": "user", "content": "earlier user turn must not be stored"},
{
"role": "assistant",
"content": [{"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {}}],
},
{"role": "user", "content": latest_content},
],
},
)
assert response.status_code == 200, response.text
row: Final = _row(isolated, "request_id", string_value(JSON_OBJECT.validate_json(response.content)["id"]))
_assert_usage_row(row, {"model": model, **expected_turns}, "messages answer")
@pytest.mark.parametrize(
("latest", "expected_turns"),
[
pytest.param(
{
"role": "user",
"content": [
{"type": "input_text", "text": "latest responses question"},
{"type": "input_image", "image_url": f"data:image/png;base64,{PIXEL}"},
],
},
{"input": _user_turn("latest responses question\n[image omitted]")},
id="user-text-and-image",
),
pytest.param(
{"type": "function_call_output", "call_id": "call_1", "output": "tool output must not be stored"},
{},
id="tool-result-turn",
),
],
)
def test_native_responses_row_keeps_only_the_latest_user_prompt_and_the_response(
gateway: Gateway, tmp_path: Path, latest: dict[str, JsonValue], expected_turns: dict[str, JsonValue]
) -> None:
with (
wire_server(_answering_model_listing(lambda _: Reply(body=_native_response("responses answer")))) as wire,
owned_proxy(gateway, tmp_path, {}, config=_latest_turn_config(tmp_path)) as isolated,
isolated.scenario() as scenario,
):
model: Final = scenario.model(model=OPENAI_MODEL, api_base=wire.url, api_key="synthetic-openai-key")
end_user: Final = f"latest-turn-native-{uuid4()}"
response: Final = isolated.request(
"POST",
"/v1/responses",
{
"model": model,
"user": end_user,
"instructions": "system prompt must not be stored",
"input": [
{"role": "user", "content": "earlier user turn must not be stored"},
{"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
latest,
],
},
)
assert response.status_code == 200, response.text
row: Final = _row(isolated, "end_user", end_user)
_assert_usage_row(row, {"model": model, **expected_turns}, "responses answer")
def test_bridged_stateless_responses_row_keeps_the_whole_request(gateway: Gateway, tmp_path: Path) -> None:
with (
wire_server(_answering_model_listing(lambda _: Reply(body=_anthropic_message("bridged answer")))) as wire,
owned_proxy(gateway, tmp_path, {}, config=_latest_turn_config(tmp_path)) as isolated,
isolated.scenario() as scenario,
):
model: Final = scenario.model(model=ANTHROPIC_MODEL, api_base=wire.url, api_key="synthetic-anthropic-key")
body: Final = {
"model": model,
"instructions": "system prompt the bridge replays",
"input": [{"role": "user", "content": "earlier user turn"}, {"role": "user", "content": "latest"}],
}
response: Final = isolated.request("POST", "/v1/responses", body)
assert response.status_code == 200, response.text
response_id: Final = string_value(object_value(response.json())["id"])
row: Final = _row(isolated, "request_id", _responses_request_id(response_id))
stored: Final = object_value(row["proxy_server_request"])
assert {key: stored.get(key) for key in body} == body, row
assert "bridged answer" in json.dumps(row["response"]), row
def test_bridged_responses_chain_replays_its_history_from_latest_turn_rows(gateway: Gateway, tmp_path: Path) -> None:
answers: Final = iter(("first answer", "second answer"))
with (
wire_server(_answering_model_listing(lambda _: Reply(body=_anthropic_message(next(answers))))) as wire,
owned_proxy(gateway, tmp_path, {}, config=_latest_turn_config(tmp_path)) as isolated,
isolated.scenario() as scenario,
):
model: Final = scenario.model(model=ANTHROPIC_MODEL, api_base=wire.url, api_key="synthetic-anthropic-key")
first_body: Final = {
"model": model,
"instructions": "SYS_MARKER answer in one word",
"input": [
{
"role": "user",
"content": [
{"type": "input_text", "text": "first question"},
{"type": "input_image", "image_url": f"data:image/png;base64,{PIXEL}"},
],
}
],
}
first: Final = isolated.post("/v1/responses", first_body)
first_id: Final = string_value(first["id"])
first_row: Final = _row(isolated, "request_id", _responses_request_id(first_id))
first_stored: Final = object_value(first_row["proxy_server_request"])
assert {key: first_stored.get(key) for key in first_body} == first_body, first_row
assert "first answer" in json.dumps(first_row["response"]), first_row
second_body: Final = {"model": model, "previous_response_id": first_id, "input": "second question"}
second: Final = isolated.post("/v1/responses", second_body)
second_row: Final = _row(isolated, "request_id", _responses_request_id(string_value(second["id"])))
second_stored: Final = object_value(second_row["proxy_server_request"])
assert {key: second_stored.get(key) for key in second_body} == second_body, second_row
assert "second answer" in json.dumps(second_row["response"]), second_row
upstream: Final = _last_upstream_body(wire)
assert "SYS_MARKER" in json.dumps(upstream.get("system")), upstream
replayed: Final = [
(message["role"], json.dumps(message["content"]))
for message in MESSAGES.validate_python(upstream["messages"])
]
assert [role for role, _ in replayed] == ["user", "assistant", "user"], replayed
assert "first question" in replayed[0][1] and PIXEL in replayed[0][1], replayed
assert "first answer" in replayed[1][1] and "second question" in replayed[2][1], replayed
def test_failed_chat_row_keeps_only_the_latest_user_prompt(gateway: Gateway, tmp_path: Path) -> None:
end_user: Final = f"latest-turn-failure-{uuid4()}"
failure: Final = Reply(
status=400, body=b'{"error":{"message":"scripted rejection","type":"invalid_request_error"}}'
)
with (
wire_server(_answering_model_listing(lambda _: failure)) as wire,
owned_proxy(gateway, tmp_path, {}, config=_latest_turn_config(tmp_path)) as isolated,
isolated.scenario() as scenario,
):
model: Final = scenario.model(model=OPENAI_MODEL, api_base=wire.url, api_key="synthetic-openai-key")
response: Final = isolated.request(
"POST",
"/v1/chat/completions",
{"model": model, "user": end_user, "messages": [*CHAT_HISTORY, {"role": "user", "content": "failed ask"}]},
)
assert response.status_code == 400, response.text
row: Final = _row(isolated, "end_user", end_user)
assert row["status"] == "failure", row
assert object_value(row["proxy_server_request"]) == {"model": model, "messages": _user_turn("failed ask")}, row

414
tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py
View file

@ -75,6 +75,8 @@ from litellm.types.utils import (
StandardLoggingModelInformation,
StandardLoggingPayload,
StandardLoggingPayloadErrorInformation,
StandardLoggingGuardrailInformation,
StandardLoggingVectorStoreRequest,
)
@ -7149,6 +7151,418 @@ def test_spend_logs_payload_with_prompts_enabled(monkeypatch: pytest.MonkeyPatch
assert payload_disabled["response"] == "{}"
_LATEST_TURN_SETTINGS: Final = {"store_prompts_in_spend_logs": True, "spend_logs_prompt_storage": "latest_turn"}
_PRIOR_CHAT_TURNS: Final = (
{"role": "system", "content": "system prompt"},
{"role": "user", "content": "earlier user turn"},
{"role": "assistant", "content": "earlier answer"},
)
_STORED_JSON: Final = TypeAdapter(object)
_FULL_CHAT_ROW: Final = TypeAdapter(dict[str, list[dict[str, str]]])
def _user_turn(text: str) -> list[dict[str, str]]:
return [{"role": "user", "content": text}]
def _stored_request_json(body: Mapping[str, object]) -> str:
return _get_proxy_server_request_for_spend_logs_payload(
metadata={}, litellm_params={"proxy_server_request": {"body": body}}
)
def _stored_request(body: Mapping[str, object]) -> object:
return _STORED_JSON.validate_json(_stored_request_json(body))
_STATELESS_CASES: Final[tuple[tuple[dict[str, object], dict[str, object]], ...]] = (
(
{"model": "gpt-4o-mini", "messages": [*_PRIOR_CHAT_TURNS, {"role": "user", "content": "latest"}]},
{"model": "gpt-4o-mini", "messages": _user_turn("latest")},
),
(
{
"system": "system prompt",
"tools": [{"name": "lookup"}],
"messages": [
{"role": "user", "content": "earlier user turn"},
{
"role": "user",
"content": [
{"type": "text", "text": "describe"},
{"type": "image", "source": {"type": "base64", "data": "AAAA"}},
{"type": "document", "source": {"type": "base64", "data": "AAAA"}},
],
},
],
},
{"messages": _user_turn("describe\n[image omitted]\n[file omitted]")},
),
(
{"messages": [{"role": "user", "content": [{"type": "image_url", "image_url": {"url": "data:,"}}]}]},
{"messages": _user_turn("[image omitted]")},
),
(
{"instructions": "system prompt", "input": "plain responses input"},
{"input": _user_turn("plain responses input")},
),
(
{
"input": [
{"role": "user", "content": "earlier user turn"},
{
"type": "message",
"role": "user",
"content": [
{"type": "input_text", "text": "summarize"},
{"type": "input_file", "file_id": "file-1"},
],
},
]
},
{"input": _user_turn("summarize\n[file omitted]")},
),
(
{
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": ""},
{"type": "thinking", "thinking": "hidden"},
{"type": "text", "text": "keep"},
],
}
]
},
{"messages": _user_turn("keep")},
),
(
{"model": "m", "messages": [*_PRIOR_CHAT_TURNS, {"role": "tool", "tool_call_id": "c", "content": "out"}]},
{"model": "m"},
),
(
{
"messages": [
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "t", "content": "tool out"},
{"type": "text", "text": "reminder next to a tool result"},
],
}
]
},
{},
),
({"input": [{"role": "user", "content": "q"}, {"type": "function_call_output", "output": "out"}]}, {}),
({"messages": [{"role": "user", "content": "q"}, {"role": "assistant", "content": "prefill"}]}, {}),
(
{"messages": [{"role": "user", "content": ["plain part", {"type": "text", "text": "block part"}]}]},
{"messages": _user_turn("plain part\nblock part")},
),
(
{"messages": [{"role": "user", "content": [{"type": []}, {"type": {}}, {"type": "text", "text": "latest"}]}]},
{"messages": _user_turn("latest")},
),
({"messages": [{"role": "user", "content": ""}]}, {}),
({"messages": []}, {}),
({"model": "gpt-4o-mini"}, {"model": "gpt-4o-mini"}),
)
@pytest.mark.parametrize(("body", "expected"), _STATELESS_CASES)
def test_latest_turn_mode_stores_only_the_model_and_the_latest_user_text_of_a_stateless_request(
monkeypatch: pytest.MonkeyPatch, body: Mapping[str, object], expected: dict[str, object]
) -> None:
from litellm.proxy import proxy_server
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
assert _stored_request(body) == expected
_CONTINUATION_BODIES: Final[tuple[dict[str, object], ...]] = (
{
"model": "gpt-4o-mini",
"previous_response_id": "resp_earlier",
"instructions": "system prompt",
"input": [{"type": "function_call_output", "call_id": "call_1", "output": "tool out"}],
},
{"model": "gpt-4o-mini", "conversation": "conv_1", "input": "next question"},
)
_NATIVE_RESPONSES_PAYLOAD: Final = {"call_type": "aresponses", "response": {"object": "response", "output": []}}
@pytest.mark.parametrize("body", _CONTINUATION_BODIES)
def test_latest_turn_mode_keeps_the_whole_request_when_a_responses_call_continues_a_stored_conversation(
monkeypatch: pytest.MonkeyPatch, body: Mapping[str, object]
) -> None:
from litellm.proxy import proxy_server
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
assert _stored_request_with_payload(body, _NATIVE_RESPONSES_PAYLOAD) == body
@pytest.mark.parametrize(
("body", "call_type", "expected"),
[
(
{
"model": "gpt-4o-mini",
"previous_response_id": "resp_earlier",
"messages": [*_PRIOR_CHAT_TURNS, {"role": "user", "content": "latest"}],
},
"acompletion",
{"model": "gpt-4o-mini", "messages": _user_turn("latest")},
),
(
{
"model": "claude",
"conversation": "conv_1",
"system": "system prompt",
"messages": [*_PRIOR_CHAT_TURNS, {"role": "user", "content": "latest"}],
},
"anthropic_messages",
{"model": "claude", "messages": _user_turn("latest")},
),
(_CONTINUATION_BODIES[0], None, {"model": "gpt-4o-mini"}),
],
)
def test_latest_turn_mode_trims_a_request_whose_continuation_keys_are_not_on_a_responses_call(
monkeypatch: pytest.MonkeyPatch, body: Mapping[str, object], call_type: str | None, expected: dict[str, object]
) -> None:
from litellm.proxy import proxy_server
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
standard_payload: Final = {"call_type": call_type, "response": {"choices": [{"message": {"content": "a"}}]}}
assert _stored_request_with_payload(body, standard_payload) == expected
_BRIDGED_RESPONSES_BODY: Final = {
"model": "claude",
"instructions": "system prompt",
"input": [{"role": "user", "content": "earlier user turn"}, {"role": "user", "content": "latest"}],
}
def _stored_request_with_payload(body: Mapping[str, object], standard_payload: Mapping[str, object]) -> object:
return _STORED_JSON.validate_json(
_get_proxy_server_request_for_spend_logs_payload(
metadata={},
litellm_params={"proxy_server_request": {"body": body}},
kwargs={"standard_logging_object": standard_payload},
)
)
@pytest.mark.parametrize(
("standard_payload", "expected"),
[
(
{"call_type": "aresponses", "response": {"choices": [{"message": {"content": "bridged answer"}}]}},
_BRIDGED_RESPONSES_BODY,
),
(
{"call_type": "aresponses", "response": {"object": "response", "output": []}},
{"model": "claude", "input": _user_turn("latest")},
),
(
{"call_type": "acompletion", "response": {"choices": [{"message": {"content": "chat answer"}}]}},
{"model": "claude", "input": _user_turn("latest")},
),
({"call_type": "aresponses", "response": None}, {"model": "claude", "input": _user_turn("latest")}),
],
)
def test_latest_turn_mode_keeps_the_whole_request_of_a_responses_turn_served_by_the_chat_bridge(
monkeypatch: pytest.MonkeyPatch, standard_payload: Mapping[str, object], expected: dict[str, object]
) -> None:
from litellm.proxy import proxy_server
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
assert _stored_request_with_payload(_BRIDGED_RESPONSES_BODY, standard_payload) == expected
def test_latest_turn_mode_stores_a_classifier_row_as_the_trimmed_originating_request(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from litellm.proxy import proxy_server
originating: Final = {
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "caller system prompt"},
*_PRIOR_CHAT_TURNS,
{"role": "user", "content": "route me"},
],
}
composed_ask: Final = "Caller system prompt, quoted as task context:\ncaller system prompt\nRecent conversation:\n[1] earlier\nCurrent ask:\nroute me"
classifier_input: Final = {
"model": "classifier",
"messages": [{"role": "system", "content": "rubric"}, {"role": "user", "content": composed_ask}],
}
standard_payload: Final = {
"call_type": "acompletion",
"classifier_input": classifier_input,
"originating_request_masked": originating,
"response": {"choices": [{"message": {"content": "simple"}}]},
}
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
stored: Final = _stored_request_with_payload(classifier_input, standard_payload)
assert stored == {
"model": "classifier",
"messages": _user_turn("route me"),
"originating_request_masked": {"model": "gpt-4o-mini", "messages": _user_turn("route me")},
}
assert "caller system prompt" not in json.dumps(stored)
def test_latest_turn_mode_keeps_message_redaction_on_a_classifier_row(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
originating: Final = {
"model": "gpt-4o-mini",
"messages": [*_PRIOR_CHAT_TURNS, {"role": "user", "content": "route me"}],
}
classifier_body: Final = {
"model": "classifier",
"messages": [{"role": "user", "content": "Current ask:\nroute me"}],
}
standard_payload: Final = {
"call_type": "acompletion",
"classifier_input": classifier_body,
"originating_request_masked": originating,
"response": {"choices": [{"message": {"content": "simple"}}]},
}
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
stored: Final = json.loads(
_get_proxy_server_request_for_spend_logs_payload(
metadata={},
litellm_params={"proxy_server_request": {"body": classifier_body}},
kwargs={
"standard_logging_object": standard_payload,
"standard_callback_dynamic_params": {"turn_off_message_logging": True},
},
)
)
assert stored["model"] == "classifier"
assert "route me" not in json.dumps(stored)
assert "originating_request_masked" not in stored
def test_latest_turn_mode_ignores_an_originating_request_supplied_by_the_client(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from litellm.proxy import proxy_server
body: Final = {
"model": "gpt-4o-mini",
"originating_request_masked": {"messages": [{"role": "user", "content": "forged prompt"}]},
"messages": [*_PRIOR_CHAT_TURNS, {"role": "user", "content": "latest"}],
}
standard_payload: Final = {"call_type": "acompletion", "response": {"choices": [{"message": {"content": "a"}}]}}
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
assert _stored_request_with_payload(body, standard_payload) == {
"model": "gpt-4o-mini",
"messages": _user_turn("latest"),
}
def test_latest_turn_mode_redacts_a_copy_of_the_vector_store_search_results(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
from litellm.proxy.spend_tracking.spend_tracking_utils import _get_vector_store_request_for_spend_logs_payload
shared: Final = [
{
"vector_store_id": "vs_1",
"vector_store_search_response": {"data": [{"content": [{"type": "text", "text": "secret passage"}]}]},
}
]
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
stored: Final = _get_vector_store_request_for_spend_logs_payload(shared)
assert stored is not None
assert stored[0]["vector_store_search_response"]["data"][0]["content"][0]["text"] == REDACTED_BY_LITELM_STRING
assert shared[0]["vector_store_search_response"]["data"][0]["content"][0]["text"] == "secret passage"
@pytest.mark.parametrize("value", ["latest_trun", "Latest_Turn", "full", True])
def test_spend_logs_prompt_storage_rejects_values_other_than_latest_turn(value: object) -> None:
from pydantic import ValidationError
from litellm.proxy._types import ConfigGeneralSettings
assert ConfigGeneralSettings.model_validate({"spend_logs_prompt_storage": "latest_turn"}).spend_logs_prompt_storage
with pytest.raises(ValidationError):
ConfigGeneralSettings.model_validate({"spend_logs_prompt_storage": value})
def test_latest_turn_mode_keeps_the_existing_string_cap(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
monkeypatch.delenv("MAX_STRING_LENGTH_PROMPT_IN_DB", raising=False)
chat_body: Final = {"messages": [{"role": "user", "content": "a" * 1500 + "b" * 1500}]}
monkeypatch.setattr(proxy_server, "general_settings", {"store_prompts_in_spend_logs": True})
capped_content: Final = _FULL_CHAT_ROW.validate_json(_stored_request_json(chat_body))["messages"][0]["content"]
assert LITELLM_TRUNCATED_PAYLOAD_FIELD in capped_content, capped_content
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
assert _stored_request(chat_body) == {"messages": _user_turn(capped_content)}
def test_latest_turn_mode_keeps_the_response(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
payload: Final = cast(StandardLoggingPayload, {"response": {"choices": [{"message": {"content": "answer"}}]}})
monkeypatch.setattr(proxy_server, "general_settings", {"store_prompts_in_spend_logs": True})
stored_today: Final = _get_response_for_spend_logs_payload(payload)
assert _STORED_JSON.validate_json(stored_today) == payload["response"]
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
assert _get_response_for_spend_logs_payload(payload) == stored_today
def test_latest_turn_mode_keeps_prompt_text_out_of_spend_log_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
prompt: Final = "earlier turn the row must not keep"
guardrail_information: Final = cast(
list[StandardLoggingGuardrailInformation],
[
{
"guardrail_name": "pii",
"guardrail_request": {"messages": [{"role": "user", "content": prompt}]},
"guardrail_response": prompt,
"match_details": [{"snippet": prompt}],
"classification": {"reason": prompt},
}
],
)
error_information: Final = cast(
StandardLoggingPayloadErrorInformation,
{"error_message": f"BadRequestError: {{'messages': [{{'role': 'user', 'content': '{prompt}'}}]}}"},
)
vector_store_request: Final = cast(
list[StandardLoggingVectorStoreRequest],
[{"vector_store_search_response": {"data": [{"content": [{"type": "text", "text": prompt}]}]}}],
)
monkeypatch.setattr(proxy_server, "general_settings", _LATEST_TURN_SETTINGS)
stored: Final = json.dumps(
[
_sanitize_guardrail_information_for_spend_logs(guardrail_information),
sanitize_error_information_for_spend_logs(error_information),
_get_vector_store_request_for_spend_logs_payload(vector_store_request),
]
)
assert prompt not in stored, stored
def test_latest_turn_mode_unset_stores_the_full_request(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.proxy import proxy_server
body: Final = {"messages": [*_PRIOR_CHAT_TURNS, {"role": "user", "content": "latest"}]}
monkeypatch.setattr(proxy_server, "general_settings", {"store_prompts_in_spend_logs": True})
assert _stored_request(body) == body
@pytest.mark.parametrize(
("configured", "disabled"),
[

5
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -30789,6 +30789,11 @@ export interface components {
spend_capture_rate_check?: components["schemas"]["SpendCaptureRateCheckSettings"] | null;
/** @description Which keys of LiteLLM_SpendLogs.metadata are written to the database. Set exactly one of 'include' (write only these keys) or 'exclude' (drop these keys). 'status' and 'cold_storage_object_key' are always written. Daily spend tables, budgets and logging callbacks still see every key. Unset writes every key */
spend_logs_metadata_fields?: components["schemas"]["SpendLogsMetadataFields"] | null;
/**
* Spend Logs Prompt Storage
* @description With store_prompts_in_spend_logs on, 'latest_turn' stores only the current turn of a stateless request (/chat/completions, /v1/messages, and /v1/responses without previous_response_id): the model and the latest user message, with the system prompt, tool definitions and earlier turns dropped. Requests that continue a stored conversation and the response are stored as they are today. Unset stores the full request.
*/
spend_logs_prompt_storage?: "latest_turn" | null;
/**
* Store Model In Db
* @description If True, models and config are stored in and loaded from the database. Default is False.