Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/content_text.py: 10%

1"""Shared content-part helpers for compression guardrails (headroom, compresr). 

2 

3Compression services only transform plain-string message content: every 

4transform in the service pipeline gates on ``isinstance(content, str)`` and 

5silently skips the OpenAI list-of-parts shape. Guardrails that send messages 

6to such a service collapse text-bearing part lists to strings here, and write 

7the rewritten text back through ``merge_rewritten_text_parts``. 

8 

9Anthropic ``cache_control`` breakpoints are positional: each one caches the 

10prefix ending at the part that carries it. A single compressed string can 

11therefore only be written back over a run of text parts, never across a 

12non-text part, which is what ``is_all_text_parts`` gates. 

13""" 

14 

15from collections.abc import Sequence 

16from typing import Final 

17 

18from litellm.litellm_core_utils.prompt_templates.factory import get_attribute_or_key 

19 

20 

21def content_to_text(content: object) -> str: 

22 """Collapse a message ``content`` (str or list-of-parts) to plain text. 

23 

24 For the multimodal list shape, joins ``{type: "text", text: ...}`` parts 

25 with blank-line separators; non-text parts are ignored. 

26 """ 

27 if isinstance(content, str): 

28 return content 

29 if isinstance(content, list): 

30 parts: Final[list[str]] = [] 

31 for part in content: 

32 if isinstance(part, dict) and part.get("type") == "text": 

33 text = part.get("text") 

34 if isinstance(text, str): 

35 parts.append(text) 

36 return "\n\n".join(parts) 

37 return "" 

38 

39 

40def is_all_text_parts(content: object) -> bool: 

41 """True when ``content`` is a non-empty part list holding only text parts.""" 

42 if not isinstance(content, list) or not content: 

43 return False 

44 return all(isinstance(part, dict) and part.get("type") == "text" for part in content) 

45 

46 

47def merge_rewritten_text_parts(parts: Sequence[object], new_text: str) -> list[object]: 

48 """Collapse a rewritten all-text part list into one part carrying ``new_text``. 

49 

50 Only all-text rows are ever flattened, so the merged part IS the whole row: 

51 it keeps the first part's fields and the LAST declared cache_control 

52 breakpoint. A breakpoint caches the prefix ending at its part, so after the 

53 merge the last one (and its TTL) is the one that still describes the row. 

54 """ 

55 dict_parts: Final = tuple(part for part in parts if isinstance(part, dict)) 

56 breakpoints: Final = tuple(part["cache_control"] for part in dict_parts if part.get("cache_control") is not None) 

57 base: Final = {**dict_parts[0], "text": new_text} if dict_parts else {"type": "text", "text": new_text} 

58 return [{**base, "cache_control": breakpoints[-1]} if breakpoints else base] 

59 

60 

61def assistant_text_from_response(response: object) -> str | None: 

62 """The assistant's natural-language text from a model response, across chat, 

63 Anthropic, and Responses shapes. Preserved when the turn is rebuilt for the 

64 retrieval follow-up so the model's reasoning is not lost.""" 

65 choices: Final = get_attribute_or_key(response, "choices", None) 

66 if isinstance(choices, list) and choices: 

67 message: Final = get_attribute_or_key(choices[0], "message", None) 

68 if message is not None: 

69 text: Final = content_to_text(get_attribute_or_key(message, "content", None)) 

70 if text: 

71 return text 

72 content: Final = get_attribute_or_key(response, "content", None) 

73 if isinstance(content, list): 

74 parts: Final = [ 

75 text 

76 for block in content 

77 if get_attribute_or_key(block, "type", None) == "text" 

78 for text in (get_attribute_or_key(block, "text", None),) 

79 if isinstance(text, str) and text 

80 ] 

81 if parts: 

82 return "".join(parts) 

83 output: Final = get_attribute_or_key(response, "output", None) 

84 if isinstance(output, list): 

85 output_parts: Final = [ 

86 text 

87 for item in output 

88 if get_attribute_or_key(item, "type", None) == "message" 

89 for chunk in (get_attribute_or_key(item, "content", None) or ()) 

90 if get_attribute_or_key(chunk, "type", None) == "output_text" 

91 for text in (get_attribute_or_key(chunk, "text", None),) 

92 if isinstance(text, str) and text 

93 ] 

94 if output_parts: 

95 return "".join(output_parts) 

96 return None