1  import warnings
 2  from typing import Optional
 3  
 4  from evidently.guardrails import GuardException
 5  from evidently.guardrails.core import GuardsException
 6  
 7  try:
 8      import tracely
 9      from tracely import SpanObject
10      from tracely.interceptors import InterceptorContext
11  except ImportError:
12      warnings.warn("tracely not installed, but required to use traces in guardrails")
13      raise
14  
15  
16  class GuardrailsInterceptor(tracely.Interceptor):
17      def before_call(self, span: SpanObject, context: InterceptorContext, *args, **kwargs):
18          pass
19  
20      def after_call(self, span: SpanObject, context: InterceptorContext, return_value):
21          guards = span.get_context_value("evidently.guardrails")
22          if guards:
23              for idx, guard in enumerate(guards):
24                  self._set_span_for_guard(span, f"g_{idx}", guard.name(), "passed", None)
25  
26      def on_exception(self, span: SpanObject, context: InterceptorContext, ex: Exception) -> bool:
27          if not isinstance(ex, GuardException):
28              return False
29          guards = span.get_context_value("evidently.guardrails")
30          if not guards:
31              return False
32  
33          if isinstance(ex, GuardsException):
34              for idx, guard in enumerate(guards):
35                  if guard in ex.failed_guards:
36                      self._set_span_for_guard(span, f"g_{idx}", guard.name(), "failed", str(ex.failed_guards.get(guard)))
37                  else:
38                      self._set_span_for_guard(span, f"g_{idx}", guard.name(), "passed", None)
39          else:
40              self._set_span_for_guard(span, "guard", guards[0].name(), "failed", str(ex))
41          return True
42  
43      def _set_span_for_guard(self, span: SpanObject, guard_id: str, guard_name: str, status: str, error: Optional[str]):
44          span.set_attribute(f"evidently.guardrail.{guard_id}.name", guard_name)
45          span.set_attribute(f"evidently.guardrail.{guard_id}.status", status)
46          if error:
47              span.set_attribute(f"evidently.guardrail.{guard_id}.error", error)