scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(guardrails): walk reasoning summary_text items in _content_utils

Browse source
This commit is contained in:
Yucheng Zhu 2026-07-11 17:40:14 -07:00
parent e6d916b82e
commit 88064d4463
2 changed files with 162 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
litellm/proxy/guardrails/_content_utils.py
View file

@ -32,7 +32,25 @@ def is_text_content_call_type(call_type: str) -> bool:
return call_type in TEXT_CONTENT_CALL_TYPES
TEXT_PART_TYPES: FrozenSet[str] = frozenset({"text", "input_text", "output_text"})
TEXT_PART_TYPES: frozenset[str] = frozenset({"text", "input_text", "output_text", "summary_text"})
def _iter_summary_texts(summary: Any) -> Iterator[str]:
"""Yield text fragments from a ``reasoning`` item's ``summary`` list —
``summary_text`` parts and bare strings; anything else is skipped."""
if not isinstance(summary, list):
return
for summary_part in summary:
if isinstance(summary_part, str):
if summary_part:
yield summary_part
continue
if not isinstance(summary_part, dict):
continue
if summary_part.get("type") in TEXT_PART_TYPES:
summary_text = summary_part.get("text")
if isinstance(summary_text, str) and summary_text:
yield summary_text
# Responses-API item types whose ``output`` field carries user/tool text
# that guardrails should inspect. ``function_call_output`` is the
@ -43,7 +61,12 @@ _OUTPUT_ITEM_TYPES: frozenset[str] = frozenset({"function_call_output", "custom_
def _iter_text_parts_in_content(content: Any) -> Iterator[str]:
"""Yield text fragments from a ``message.content`` value (string or
multimodal list). Non-text parts (images, audio, …) are skipped."""
multimodal list). Non-text parts (images, audio, …) are skipped.
Also descends into ``reasoning`` items whose ``summary`` list may
contain ``summary_text`` parts carrying chain-of-thought text that
guardrails need to inspect/redact.
"""
if isinstance(content, str):
if content:
yield content
@ -61,6 +84,10 @@ def _iter_text_parts_in_content(content: Any) -> Iterator[str]:
text = part.get("text")
if isinstance(text, str) and text:
yield text
elif part.get("type") == "reasoning":
# Reasoning items carry a ``summary`` list of
# ``{"type": "summary_text", "text": "..."}`` parts.
yield from _iter_summary_texts(part.get("summary"))
def _coerce_input_to_messages(input_value: Any) -> List[Dict[str, Any]]:
@ -80,6 +107,13 @@ def _coerce_input_to_messages(input_value: Any) -> List[Dict[str, Any]]:
messages.append({"role": item.get("role") or "user", "content": item["content"]})
elif item.get("type") in _OUTPUT_ITEM_TYPES and "output" in item:
messages.append({"role": item.get("role") or "tool", "content": item["output"]})
elif item.get("type") == "reasoning":
# Reasoning items carry chain-of-thought text in their
# ``summary`` list. Synthesise a message so guardrails
# can inspect/redact that text.
summary = item.get("summary")
if isinstance(summary, list):
messages.append({"role": "assistant", "content": summary})
return messages
@ -112,6 +146,25 @@ def walk_user_text(data: Dict[str, Any], visit: Callable[[str], str]) -> int:
"""
visited = 0
def _rewrite_summary(summary: list[Any]) -> list[Any]:
nonlocal visited
new_parts: List[Any] = []
for summary_part in summary:
if isinstance(summary_part, str) and summary_part:
visited += 1
new_parts.append(visit(summary_part))
elif (
isinstance(summary_part, dict)
and summary_part.get("type") in TEXT_PART_TYPES
and isinstance(summary_part.get("text"), str)
and summary_part["text"]
):
visited += 1
new_parts.append({**summary_part, "text": visit(summary_part["text"])})
else:
new_parts.append(summary_part)
return new_parts
def _rewrite_content(content: Any) -> Any:
nonlocal visited
if isinstance(content, str):
@ -133,6 +186,10 @@ def walk_user_text(data: Dict[str, Any], visit: Callable[[str], str]) -> int:
):
visited += 1
new_parts.append({**part, "text": visit(part["text"])})
elif (
isinstance(part, dict) and part.get("type") == "reasoning" and isinstance(part.get("summary"), list)
):
new_parts.append({**part, "summary": _rewrite_summary(part["summary"])})
else:
new_parts.append(part)
return new_parts
@ -165,6 +222,10 @@ def walk_user_text(data: Dict[str, Any], visit: Callable[[str], str]) -> int:
item["content"] = _rewrite_content(item["content"])
elif item.get("type") in _OUTPUT_ITEM_TYPES and "output" in item:
item["output"] = _rewrite_content(item["output"])
elif item.get("type") == "reasoning":
summary = item.get("summary")
if isinstance(summary, list):
item["summary"] = _rewrite_content(summary)
return visited
return visited

99
tests/test_litellm/proxy/guardrails/test_content_utils.py
View file

@ -563,3 +563,102 @@ def test_build_inspection_messages_custom_tool_call_output():
}
msgs = build_inspection_messages(data)
assert any("custom-tool-leak" in m["content"] for m in msgs)
# -------------------------------------------------------------------
# LIT-4303: reasoning.summary_text walking
# -------------------------------------------------------------------
def test_iter_message_text_walks_reasoning_summary_text():
"""Reasoning items with summary_text parts should yield their text."""
data = {
"input": [
{"type": "reasoning", "summary": [
{"type": "summary_text", "text": "secret-in-cot"},
]},
]
}
from litellm.proxy.guardrails._content_utils import iter_message_text
texts = list(iter_message_text(data))
assert "secret-in-cot" in texts
def test_iter_message_text_reasoning_part_inside_message_content():
"""A reasoning item nested inside a message's content list is walked
directly by ``_iter_text_parts_in_content``: summary_text parts and bare
strings in the summary yield their text, everything else is skipped."""
data = {
"messages": [
{
"role": "assistant",
"content": [
{
"type": "reasoning",
"summary": [
{"type": "summary_text", "text": "cot-secret"},
"bare-summary-string",
{"type": "image_url", "image_url": {"url": "..."}},
42,
],
},
],
},
]
}
assert list(iter_message_text(data)) == ["cot-secret", "bare-summary-string"]
def test_walk_user_text_redacts_reasoning_summary_text():
"""walk_user_text should rewrite text inside reasoning.summary."""
data = {
"input": [
{"type": "reasoning", "summary": [
{"type": "summary_text", "text": "SSN-123"},
]},
]
}
count = walk_user_text(data, lambda t: t.replace("SSN-123", "[REDACTED]"))
assert count >= 1
assert data["input"][0]["summary"][0]["text"] == "[REDACTED]"
def test_walk_user_text_redacts_reasoning_part_inside_message_content():
"""Greptile P1: a reasoning item nested inside ``messages[i]["content"]``
was detected by ``iter_message_text`` but copied through unchanged by
``walk_user_text``; read and write coverage must agree."""
data = {
"messages": [
{
"role": "assistant",
"content": [
{
"type": "reasoning",
"summary": [
{"type": "summary_text", "text": "SSN-123 in cot"},
"bare SSN-123",
{"type": "image_url", "image_url": {"url": "..."}},
],
},
],
},
]
}
count = walk_user_text(data, lambda t: t.replace("SSN-123", "[REDACTED]"))
assert count == 2
summary = data["messages"][0]["content"][0]["summary"]
assert summary[0] == {"type": "summary_text", "text": "[REDACTED] in cot"}
assert summary[1] == "bare [REDACTED]"
assert summary[2] == {"type": "image_url", "image_url": {"url": "..."}}
def test_build_inspection_messages_reasoning_summary():
"""build_inspection_messages should include reasoning summary text."""
data = {
"input": [
{"type": "reasoning", "summary": [
{"type": "summary_text", "text": "chain-of-thought leak"},
]},
]
}
msgs = build_inspection_messages(data)
assert any("chain-of-thought leak" in m["content"] for m in msgs)