scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 114

fix(logging): redact assistant tool call arguments in spend logs (#33111)

Browse source 
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-07-16 00:07:54 +03:00 • committed by GitHub
parent f7fc679f27
commit 24a438adfd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 211 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
litellm/litellm_core_utils/redact_messages.py
View file

@ -77,6 +77,22 @@ def _redact_streaming_response(streaming_response):
streaming_response.reasoning = None
def _redact_tool_calls(tool_calls) -> None:
"""Redact tool call arguments (assistant tool calls carry prompt-derived data)."""
if not tool_calls:
return
for tool_call in tool_calls:
function = getattr(tool_call, "function", None)
if function is not None and hasattr(function, "arguments"):
function.arguments = "redacted-by-litellm"
def _redact_function_call(function_call) -> None:
"""Redact legacy assistant function_call arguments."""
if function_call is not None and hasattr(function_call, "arguments"):
function_call.arguments = "redacted-by-litellm"
def _redact_choice_content(choice):
"""Helper to redact content in a choice (message or delta)."""
if isinstance(choice, litellm.Choices):
@ -85,12 +101,16 @@ def _redact_choice_content(choice):
choice.message.reasoning_content = "redacted-by-litellm"
if hasattr(choice.message, "thinking_blocks"):
choice.message.thinking_blocks = None
_redact_tool_calls(getattr(choice.message, "tool_calls", None))
_redact_function_call(getattr(choice.message, "function_call", None))
elif isinstance(choice, litellm.utils.StreamingChoices):
choice.delta.content = "redacted-by-litellm"
if hasattr(choice.delta, "reasoning_content"):
choice.delta.reasoning_content = "redacted-by-litellm"
if hasattr(choice.delta, "thinking_blocks"):
choice.delta.thinking_blocks = None
_redact_tool_calls(getattr(choice.delta, "tool_calls", None))
_redact_function_call(getattr(choice.delta, "function_call", None))
def _redact_responses_api_output(output_items):
@ -111,6 +131,9 @@ def _redact_responses_api_output(output_items):
if hasattr(summary_item, "text"):
summary_item.text = "redacted-by-litellm"
if hasattr(output_item, "type") and output_item.type == "function_call" and hasattr(output_item, "arguments"):
output_item.arguments = "redacted-by-litellm"
def _redact_responses_api_output_dict(output_items, redacted_str: str):
"""Helper to redact ResponsesAPIResponse output items in dict form."""
@ -131,6 +154,9 @@ def _redact_responses_api_output_dict(output_items, redacted_str: str):
if isinstance(summary_item, dict) and "text" in summary_item:
summary_item["text"] = redacted_str
if output_item.get("type") == "function_call" and "arguments" in output_item:
output_item["arguments"] = redacted_str
def _redact_standard_logging_object(model_call_details: dict):
"""Redact messages and response inside standard_logging_object if present."""
@ -162,6 +188,19 @@ def _redact_standard_logging_object(model_call_details: dict):
standard_logging_object["response"] = {"text": redacted_str}
def _redact_tool_calls_dict(message: dict, redacted_str: str) -> None:
"""Redact tool call / function_call arguments in a dict-form message or delta."""
tool_calls = message.get("tool_calls")
if isinstance(tool_calls, list):
for tool_call in tool_calls:
if isinstance(tool_call, dict) and isinstance(tool_call.get("function"), dict):
tool_call["function"]["arguments"] = redacted_str
function_call = message.get("function_call")
if isinstance(function_call, dict) and "arguments" in function_call:
function_call["arguments"] = redacted_str
def _redact_model_response_dict_choices(choices, redacted_str: str):
for choice in choices:
if isinstance(choice, dict):
@ -173,6 +212,7 @@ def _redact_model_response_dict_choices(choices, redacted_str: str):
choice["message"]["thinking_blocks"] = None
if "audio" in choice["message"]:
choice["message"]["audio"] = None
_redact_tool_calls_dict(choice["message"], redacted_str)
elif "delta" in choice and isinstance(choice["delta"], dict):
choice["delta"]["content"] = redacted_str
if "reasoning_content" in choice["delta"]:
@ -181,6 +221,7 @@ def _redact_model_response_dict_choices(choices, redacted_str: str):
choice["delta"]["thinking_blocks"] = None
if "audio" in choice["delta"]:
choice["delta"]["audio"] = None
_redact_tool_calls_dict(choice["delta"], redacted_str)
else:
_redact_choice_content(choice)

170
tests/test_litellm/litellm_core_utils/test_redact_messages.py
View file

@ -320,6 +320,176 @@ class TestPerformRedaction:
assert choice.message.content == "redacted-by-litellm"
assert choice.message.reasoning_content == "redacted-by-litellm"
def test_redacts_tool_call_arguments_in_model_response_dict(self):
"""Assistant tool call arguments must not leak when redaction is on."""
result = {
"choices": [
{
"message": {
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
},
}
],
"function_call": {
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
},
}
}
]
}
redacted = perform_redaction({}, result)
message = redacted["choices"][0]["message"]
assert message["content"] == "redacted-by-litellm"
tool_call = message["tool_calls"][0]
assert tool_call["function"]["arguments"] == "redacted-by-litellm"
assert tool_call["function"]["name"] == "get_weather"
assert message["function_call"]["arguments"] == "redacted-by-litellm"
def test_redacts_tool_call_arguments_in_streaming_delta_dict(self):
result = {
"choices": [
{
"delta": {
"content": None,
"tool_calls": [
{
"index": 0,
"function": {
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
},
}
],
}
}
]
}
redacted = perform_redaction({}, result)
delta = redacted["choices"][0]["delta"]
assert delta["tool_calls"][0]["function"]["arguments"] == "redacted-by-litellm"
def test_redacts_tool_call_arguments_on_model_response_object(self):
result = litellm.ModelResponse(
id="resp-1",
choices=[
litellm.Choices(
message=litellm.Message(
content=None,
role="assistant",
tool_calls=[
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
},
}
],
)
)
],
model="gpt-4o",
)
redacted = perform_redaction({}, result)
tool_call = redacted.choices[0].message.tool_calls[0]
assert tool_call.function.arguments == "redacted-by-litellm"
assert tool_call.function.name == "get_weather"
assert result.choices[0].message.tool_calls[0].function.arguments == (
'{"city": "sensitive-city"}'
)
def test_redacts_tool_call_arguments_on_streaming_response_object(self):
"""Reproduces the Stream=True path where tool calls arrive as deltas."""
streaming_choice = litellm.utils.StreamingChoices(
delta=litellm.utils.Delta(
content=None,
role="assistant",
tool_calls=[
{
"index": 0,
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
},
}
],
)
)
streaming_response = SimpleNamespace(choices=[streaming_choice])
details = {
"stream": True,
"complete_streaming_response": streaming_response,
}
perform_redaction(details, None)
tool_call = streaming_response.choices[0].delta.tool_calls[0]
assert tool_call.function.arguments == "redacted-by-litellm"
def test_redacts_tool_call_arguments_in_standard_logging_object(self):
details = {
"standard_logging_object": {
"response": {
"choices": [
{
"message": {
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
},
}
],
}
}
]
}
}
}
perform_redaction(details, None)
message = details["standard_logging_object"]["response"]["choices"][0]["message"]
assert message["tool_calls"][0]["function"]["arguments"] == "redacted-by-litellm"
def test_redacts_responses_api_function_call_arguments_dict(self):
result = {
"output": [
{
"type": "function_call",
"name": "get_weather",
"arguments": '{"city": "sensitive-city"}',
"call_id": "call_1",
}
]
}
redacted = perform_redaction({}, result)
assert redacted["output"][0]["arguments"] == "redacted-by-litellm"
assert redacted["output"][0]["name"] == "get_weather"
def test_redacts_response_output_objects_with_top_level_text(self):
output_items = [
SimpleNamespace(text="top-level output"),