Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/javelin/javelin.py: 17%
114 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1from datetime import datetime
2from typing import TYPE_CHECKING, Final
4from fastapi import HTTPException
6import litellm
7from litellm._logging import verbose_proxy_logger
8from litellm.integrations.custom_guardrail import CustomGuardrail
9from litellm.llms.custom_httpx.http_handler import (
10 get_async_httpx_client,
11 httpxSpecialProvider,
12)
13from litellm.proxy._types import UserAPIKeyAuth
14from litellm.secret_managers.main import get_secret_str
15from litellm.types.guardrails import GuardrailEventHooks
16from litellm.types.proxy.guardrails.guardrail_hooks.javelin import (
17 JavelinGuardInput,
18 JavelinGuardRequest,
19 JavelinGuardResponse,
20)
21from litellm.types.utils import CallTypesLiteral, GuardrailStatus
23if TYPE_CHECKING: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true
24 from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
27class JavelinGuardrail(CustomGuardrail):
28 @classmethod
29 def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
30 return [
31 GuardrailEventHooks.pre_call,
32 ]
34 def __init__(
35 self,
36 api_key: str | None = None,
37 api_base: str | None = None,
38 default_on: bool = True,
39 guardrail_name: str = "trustsafety",
40 javelin_guard_name: str | None = None,
41 api_version: str = "v1",
42 metadata: dict | None = None,
43 config: dict | None = None,
44 application: str | None = None,
45 **kwargs,
46 ):
47 """
48 Initialize the JavelinGuardrail class.
50 This calls: {api_base}/{api_version}/guardrail/{guardrail_name}/apply
52 Args:
53 api_key: str = None,
54 api_base: str = None,
55 default_on: bool = True,
56 api_version: str = "v1",
57 guardrail_name: str = "trustsafety",
58 metadata: Optional[Dict] = None,
59 config: Optional[Dict] = None,
60 application: Optional[str] = None,
61 """
63 self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
64 self.javelin_api_key = api_key or get_secret_str("JAVELIN_API_KEY")
65 self.api_base = api_base or get_secret_str("JAVELIN_API_BASE") or "https://api-dev.javelin.live"
66 self.api_version = api_version
67 self.guardrail_name = guardrail_name
68 self.javelin_guard_name = javelin_guard_name or guardrail_name
69 self.default_on = default_on
70 self.metadata = metadata
71 self.config = config
72 self.application = application
73 verbose_proxy_logger.debug(
74 "Javelin Guardrail: Initialized with guardrail_name=%s, javelin_guard_name=%s, api_base=%s, api_version=%s",
75 self.guardrail_name,
76 self.javelin_guard_name,
77 self.api_base,
78 self.api_version,
79 )
81 kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
82 super().__init__(guardrail_name=guardrail_name, default_on=default_on, **kwargs)
84 async def call_javelin_guard(
85 self,
86 request: JavelinGuardRequest,
87 event_type: GuardrailEventHooks,
88 ) -> JavelinGuardResponse:
89 """
90 Call the Javelin guard API.
91 """
92 start_time: Final = datetime.now()
93 # Create a new request with metadata if it's not already set
94 if request.get("metadata") is None and self.metadata is not None:
95 request = {**request, "metadata": self.metadata}
96 headers: Final = {
97 "x-javelin-apikey": self.javelin_api_key,
98 }
99 if self.application:
100 headers["x-javelin-application"] = self.application
102 status: GuardrailStatus = "guardrail_failed_to_respond"
103 javelin_response: JavelinGuardResponse | None = None
104 exception_str = ""
106 try:
107 verbose_proxy_logger.debug("Javelin Guardrail: Calling Javelin guard API with request: %s", request)
108 url: Final = f"{self.api_base}/{self.api_version}/guardrail/{self.javelin_guard_name}/apply"
109 verbose_proxy_logger.debug("Javelin Guardrail: Calling URL: %s", url)
110 response: Final = await self.async_handler.post(
111 url=url,
112 headers=headers,
113 json=dict(request),
114 )
115 verbose_proxy_logger.debug("Javelin Guardrail: Javelin guard API response: %s", response.json())
116 response_data: Final = response.json()
117 # Ensure the response has the required assessments field
118 if "assessments" not in response_data:
119 response_data["assessments"] = []
121 javelin_response = {"assessments": response_data.get("assessments", [])}
122 status = "success"
123 return javelin_response
124 except Exception as e:
125 status = "guardrail_failed_to_respond"
126 exception_str = str(e)
127 return {"assessments": []}
128 finally:
129 ####################################################
130 # Create Guardrail Trace for logging on Langfuse, Datadog, etc.
131 ####################################################
132 guardrail_json_response: Exception | str | dict | list[dict] = {}
133 if status == "success" and javelin_response is not None:
134 guardrail_json_response = dict(javelin_response)
135 else:
136 guardrail_json_response = exception_str
138 # Create a clean request data copy for logging (without guardrail responses)
139 clean_request_data: Final = {
140 "input": request.get("input", {}),
141 "metadata": request.get("metadata", {}),
142 "config": request.get("config", {}),
143 }
144 # Remove any existing guardrail logging information to prevent recursion
145 if "metadata" in clean_request_data and clean_request_data["metadata"]:
146 clean_request_data["metadata"] = {
147 k: v
148 for k, v in clean_request_data["metadata"].items()
149 if k != "standard_logging_guardrail_information"
150 }
152 self.add_standard_logging_guardrail_information_to_request_data(
153 guardrail_json_response=guardrail_json_response,
154 request_data=clean_request_data,
155 guardrail_status=status,
156 start_time=start_time.timestamp(),
157 end_time=datetime.now().timestamp(),
158 duration=(datetime.now() - start_time).total_seconds(),
159 event_type=event_type,
160 )
162 async def async_pre_call_hook(
163 self,
164 user_api_key_dict: UserAPIKeyAuth,
165 cache: litellm.DualCache,
166 data: dict,
167 call_type: CallTypesLiteral,
168 ) -> Exception | str | dict | None:
169 """
170 Pre-call hook for the Javelin guardrail.
171 """
172 from litellm.litellm_core_utils.prompt_templates.common_utils import (
173 get_last_user_message,
174 )
175 from litellm.proxy.common_utils.callback_utils import (
176 add_guardrail_to_applied_guardrails_header,
177 )
179 verbose_proxy_logger.debug("Javelin Guardrail: pre_call_hook")
180 verbose_proxy_logger.debug("Javelin Guardrail: Request data: %s", data)
182 event_type: Final[GuardrailEventHooks] = GuardrailEventHooks.pre_call
183 if self.should_run_guardrail(data=data, event_type=event_type) is not True:
184 verbose_proxy_logger.debug("Javelin Guardrail: not running guardrail. Guardrail is disabled.")
185 return data
187 if "messages" not in data:
188 return data
190 text: Final = get_last_user_message(data["messages"])
191 if text is None:
192 return data
194 clean_metadata = {}
195 if self.metadata:
196 clean_metadata = {k: v for k, v in self.metadata.items() if k != "standard_logging_guardrail_information"}
198 javelin_guard_request: Final = JavelinGuardRequest(
199 input=JavelinGuardInput(text=text),
200 metadata=clean_metadata,
201 config=self.config if self.config else {},
202 )
204 javelin_response: Final = await self.call_javelin_guard(
205 request=javelin_guard_request, event_type=GuardrailEventHooks.pre_call
206 )
208 assessments: Final = javelin_response.get("assessments", [])
209 reject_prompt = ""
210 should_reject = False
212 # Debug: Log the full Javelin response
213 verbose_proxy_logger.debug("Javelin Guardrail: Full Javelin response: %s", javelin_response)
215 for assessment in assessments:
216 verbose_proxy_logger.debug("Javelin Guardrail: Processing assessment: %s", assessment)
217 for assessment_type, assessment_data in assessment.items():
218 verbose_proxy_logger.debug(
219 "Javelin Guardrail: Processing assessment_type: %s, data: %s",
220 assessment_type,
221 assessment_data,
222 )
223 # Check if this assessment indicates rejection
224 if assessment_data.get("request_reject") is True:
225 should_reject = True
226 verbose_proxy_logger.debug(
227 "Javelin Guardrail: Request rejected by Javelin guardrail: %s (assessment_type: %s)",
228 self.guardrail_name,
229 assessment_type,
230 )
232 results = assessment_data.get("results", {})
233 reject_prompt = str(results.get("reject_prompt", ""))
235 verbose_proxy_logger.debug(
236 "Javelin Guardrail: Extracted reject_prompt: '%s'",
237 reject_prompt,
238 )
239 break
240 if should_reject:
241 break
243 verbose_proxy_logger.debug(
244 "Javelin Guardrail: should_reject=%s, reject_prompt='%s'",
245 should_reject,
246 reject_prompt,
247 )
249 if should_reject:
250 if not reject_prompt:
251 reject_prompt = f"Request blocked by Javelin guardrails due to {self.guardrail_name} violation."
253 verbose_proxy_logger.debug(
254 "Javelin Guardrail: Blocking request with reject_prompt: '%s'",
255 reject_prompt,
256 )
258 # Raise HTTPException to prevent the request from going to the LLM
259 raise HTTPException(
260 status_code=500,
261 detail={
262 "error": "Violated guardrail policy",
263 "javelin_guardrail_response": javelin_response,
264 "reject_prompt": reject_prompt,
265 },
266 )
268 add_guardrail_to_applied_guardrails_header(request_data=data, guardrail_name=self.guardrail_name)
270 return data
272 @staticmethod
273 def get_config_model() -> type["GuardrailConfigModel"] | None:
274 """
275 Get the config model for the Javelin guardrail.
276 """
277 from litellm.types.proxy.guardrails.guardrail_hooks.javelin import (
278 JavelinGuardrailConfigModel,
279 )
281 return JavelinGuardrailConfigModel