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

1"""Conduct Guard as a LiteLLM guardrail, backed by the ``conduct-litellm-guard`` PyPI package. 

2 

3Install: ``pip install "conduct-litellm-guard>=0.2.5"`` 

4Source: https://github.com/sseshachala/conductai/tree/main/packages/conduct-litellm-guard 

5""" 

6 

7from __future__ import annotations 

8 

9import inspect 

10from collections.abc import Awaitable, Callable, Mapping 

11from functools import partial 

12from types import MappingProxyType 

13from typing import TYPE_CHECKING, Final, Literal, Protocol 

14 

15from pydantic import BaseModel, ConfigDict 

16 

17from litellm.integrations.custom_guardrail import CustomGuardrail, log_guardrail_information 

18from litellm.types.llms.openai import ChatCompletionUserMessage 

19from litellm.types.proxy.guardrails.guardrail_hooks.conduct import ConductGuardrailConfigModel 

20 

21if TYPE_CHECKING: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj 

23 from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailStatus 

24 

25MISSING_PACKAGE_MESSAGE: Final = ( 

26 "conduct-litellm-guard>=0.2.5 is required for the Conduct guardrail. " 

27 'Install it with: pip install "conduct-litellm-guard>=0.2.5"' 

28) 

29 

30BLOCKING_VERDICTS: Final = frozenset({"block", "approval"}) 

31FLAGGED_VERDICTS: Final = frozenset({"warning", "advisory"}) 

32 

33 

34class ConductDecision(Protocol): 

35 @property 

36 def verdict(self) -> str: ... 36 ↛ exitline 36 didn't return from function 'verdict' because

37 

38 @property 

39 def rule_id(self) -> str | None: ... 39 ↛ exitline 39 didn't return from function 'rule_id' because

40 

41 

42class ConductCheck(Protocol): 

43 def __call__(self, *, data: Mapping[str, object], call_type: str) -> Awaitable[ConductDecision]: ... 43 ↛ exitline 43 didn't return from function '__call__' because

44 

45 

46def request_payload( 

47 inputs: GenericGuardrailAPIInputs, 

48 request_data: Mapping[str, object], 

49 input_type: Literal["request", "response"], 

50) -> Mapping[str, object] | None: 

51 if input_type != "request": 

52 return None 

53 messages: Final = inputs.get("structured_messages") or tuple( 

54 ChatCompletionUserMessage(role="user", content=text) for text in inputs.get("texts") or () 

55 ) 

56 return MappingProxyType({**request_data, "prompt": None, "messages": messages}) 

57 

58 

59def decision_status(decision: ConductDecision) -> GuardrailStatus: 

60 return "guardrail_flagged" if decision.verdict in FLAGGED_VERDICTS else "success" 

61 

62 

63class ConductVerdict(BaseModel): 

64 model_config = ConfigDict(frozen=True) 

65 

66 verdict: str 

67 rule_id: str | None = None 

68 

69 

70def record_decision( 

71 guardrail: CustomGuardrail, 

72 request_data: dict[str, object], # mutable-ok: the logging helper writes metadata into it 

73 decision: ConductDecision, 

74) -> None: 

75 guardrail.add_standard_logging_guardrail_information_to_request_data( 

76 guardrail_json_response=ConductVerdict(verdict=decision.verdict, rule_id=decision.rule_id).model_dump(), 

77 request_data=request_data, 

78 guardrail_status=decision_status(decision), 

79 ) 

80 

81 

82async def apply_conduct_guardrail( 

83 inputs: GenericGuardrailAPIInputs, 

84 request_data: Mapping[str, object], 

85 input_type: Literal["request", "response"], 

86 check: ConductCheck, 

87 blocked: Callable[[ConductDecision], Exception], 

88 record: Callable[[ConductDecision], None], 

89) -> GenericGuardrailAPIInputs: 

90 payload: Final = request_payload(inputs, request_data, input_type) 

91 if payload is None: 

92 return inputs 

93 decision: Final = await check(data=payload, call_type=input_type) 

94 if decision.verdict in BLOCKING_VERDICTS: 

95 raise blocked(decision) 

96 record(decision) 

97 return inputs 

98 

99 

100def binds_unreachable_fallback(guardrail_cls: type[object]) -> bool: 

101 return "unreachable_fallback" in inspect.signature(guardrail_cls.__init__).parameters 

102 

103 

104try: 

105 from conduct_litellm_guard.guardrail import ConductGuard, ConductGuardBlocked 

106 

107 if not binds_unreachable_fallback(ConductGuard): 

108 raise ImportError(MISSING_PACKAGE_MESSAGE) 

109except ImportError as import_error: 

110 _import_error: Final = import_error 

111 

112 class ConductGuardrail(CustomGuardrail): 

113 def __init__(self, **kwargs: object) -> None: # kwargs-ok: mirrors the plugin constructor, only raises 

114 raise ImportError(MISSING_PACKAGE_MESSAGE) from _import_error 

115 

116 @staticmethod 

117 def get_config_model() -> type[ConductGuardrailConfigModel]: 

118 return ConductGuardrailConfigModel 

119 

120else: 

121 

122 class ConductGuardrail(ConductGuard): # pyright: ignore[reportUntypedBaseClass] # optional dep, absent at type-check 

123 @staticmethod 

124 def get_config_model() -> type[ConductGuardrailConfigModel]: 

125 return ConductGuardrailConfigModel 

126 

127 @log_guardrail_information 

128 async def apply_guardrail( 

129 self, 

130 inputs: GenericGuardrailAPIInputs, 

131 request_data: dict[str, object], # mutable-ok: CustomGuardrail.apply_guardrail contract 

132 input_type: Literal["request", "response"], 

133 logging_obj: LiteLLMLoggingObj | None = None, 

134 ) -> GenericGuardrailAPIInputs: 

135 return await apply_conduct_guardrail( 

136 inputs, 

137 request_data, 

138 input_type, 

139 self.check, 

140 ConductGuardBlocked, 

141 partial(record_decision, self, request_data), 

142 ) 

143 

144 

145__all__ = ( 

146 "BLOCKING_VERDICTS", 

147 "FLAGGED_VERDICTS", 

148 "MISSING_PACKAGE_MESSAGE", 

149 "ConductCheck", 

150 "ConductDecision", 

151 "ConductGuardrail", 

152 "ConductVerdict", 

153 "apply_conduct_guardrail", 

154 "binds_unreachable_fallback", 

155 "decision_status", 

156 "record_decision", 

157 "request_payload", 

158)