Validator Hub Telemetry

guardrails/cli_dir/hub/credentials.py

@@ -0,0 +1,38 @@
+import os
+from dataclasses import dataclass
+from os.path import expanduser
+from typing import Optional
+
+from guardrails.cli_dir.logger import logger
+from guardrails.cli_dir.server.serializeable import Serializeable
+
+
+@dataclass
+class Credentials(Serializeable):
+    id: Optional[str] = None
+    client_id: Optional[str] = None
+    client_secret: Optional[str] = None
+    no_metrics: Optional[bool] = False
+
+    @staticmethod
+    def from_rc_file() -> "Credentials":
+        try:
+            home = expanduser("~")
+            guardrails_rc = os.path.join(home, ".guardrailsrc")
+            with open(guardrails_rc) as rc_file:
+                lines = rc_file.readlines()
+                creds = {}
+                for line in lines:
+                    key, value = line.split("=", 1)
+                    creds[key.strip()] = value.strip()
+                rc_file.close()
+                return Credentials.from_dict(creds)
+
+        except FileNotFoundError as e:
+            logger.error(e)
+            logger.error(
+                "Guardrails Hub credentials not found!"
+                "You will need to sign up to use any authenticated Validators here:"
+                "{insert url}"
+            )
+            return Credentials()

guardrails/cli_dir/logger.py

@@ -0,0 +1,20 @@
+import logging
+import os
+
+import coloredlogs
+
+os.environ[
+    "COLOREDLOGS_LEVEL_STYLES"
+] = "spam=white,faint;success=green,bold;debug=magenta;verbose=blue;notice=cyan,bold;warning=yellow;error=red;critical=background=red"  # noqa
+LEVELS = {
+    "SPAM": 5,
+    "VERBOSE": 15,
+    "NOTICE": 25,
+    "SUCCESS": 35,
+}
+for key in LEVELS:
+    logging.addLevelName(LEVELS.get(key), key)  # type: ignore
+
+
+logger = logging.getLogger("guardrails-cli")
+coloredlogs.install(level="DEBUG", logger=logger)

guardrails/cli_dir/server/serializeable.py

@@ -0,0 +1,33 @@
+import inspect
+import json
+from dataclasses import InitVar, asdict, dataclass, field, is_dataclass
+from json import JSONEncoder
+from typing import Any, Dict
+
+
+class SerializeableJSONEncoder(JSONEncoder):
+    def default(self, o):
+        if is_dataclass(o):
+            return asdict(o)
+        return super().default(o)
+
+
+@dataclass
+class Serializeable:
+    encoder: InitVar[JSONEncoder] = field(
+        kw_only=True, default=SerializeableJSONEncoder  # type: ignore
+    )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]):
+        annotations = inspect.get_annotations(cls)
+        attributes = dict.keys(annotations)
+        kwargs = {k: data.get(k) for k in data if k in attributes}
+        return cls(**kwargs)  # type: ignore
+
+    @property
+    def __dict__(self) -> Dict[str, Any]:
+        return asdict(self)
+
+    def to_json(self):
+        return json.dumps(self, cls=self.encoder)  # type: ignore

guardrails/guard.py

@@ -24,6 +24,7 @@
 from guardrails.classes.generic import Stack
 from guardrails.classes.history import Call
 from guardrails.classes.history.call_inputs import CallInputs
+from guardrails.cli_dir.hub.credentials import Credentials
 from guardrails.llm_providers import get_async_llm_ask, get_llm_ask
 from guardrails.logger import logger, set_scope
 from guardrails.prompt import Instructions, Prompt
@@ -37,6 +38,7 @@
     set_tracer,
     set_tracer_context,
 )
+from guardrails.utils.hub_telemetry_utils import HubTelemetry
 from guardrails.validators import Validator
 
 add_destinations(logger.debug)
@@ -61,6 +63,9 @@ class Guard(Generic[OT]):
 
     _tracer = None
     _tracer_context = None
+    _hub_telemetry = None
+    _guard_id = None
+    _user_id = None
 
     def __init__(
         self,
@@ -77,6 +82,24 @@ def __init__(
         self.base_model = base_model
         self._set_tracer(tracer)
 
+        # Get unique id of user from credentials
+        self._user_id = Credentials.from_rc_file().id
+
+        # Get metrics opt-out from credentials
+        self._disable_tracer = Credentials.from_rc_file().no_metrics
+        if self._disable_tracer.strip().lower() == "true":
+            self._disable_tracer = True
+        elif self._disable_tracer.strip().lower() == "false":
+            self._disable_tracer = False
+
+        # Get id of guard object (that is unique)
+        self._guard_id = id(self)  # id of guard object; not the class
+
+        # Initialize Hub Telemetry singleton and get the tracer
+        #  if it is not disabled
+        if not self._disable_tracer:
+            self._hub_telemetry = HubTelemetry()
+
     @property
     def prompt_schema(self) -> Optional[StringSchema]:
         """Return the input schema."""
@@ -359,6 +382,24 @@ def __call(
             if prompt_params is None:
                 prompt_params = {}
 
+            if not self._disable_tracer:
+                # Create a new span for this guard call
+                self._hub_telemetry.create_new_span(
+                    span_name="/guard_call",
+                    attributes=[
+                        ("guard_id", self._guard_id),
+                        ("user_id", self._user_id),
+                        ("llm_api", llm_api.__name__ if llm_api else "None"),
+                        ("custom_reask_prompt", self.reask_prompt is not None),
+                        (
+                            "custom_reask_instructions",
+                            self.reask_instructions is not None,
+                        ),
+                    ],
+                    is_parent=True,  # It will have children
+                    has_parent=False,  # Has no parents
+                )
+
             set_call_kwargs(kwargs)
             set_tracer(self._tracer)
             set_tracer_context(self._tracer_context)
@@ -650,6 +691,24 @@ def __parse(
             final_num_reasks = (
                 num_reasks if num_reasks is not None else 0 if llm_api is None else None
             )
+
+            if not self._disable_tracer:
+                self._hub_telemetry.create_new_span(
+                    span_name="/guard_parse",
+                    attributes=[
+                        ("guard_id", self._guard_id),
+                        ("user_id", self._user_id),
+                        ("llm_api", llm_api.__name__ if llm_api else "None"),
+                        ("custom_reask_prompt", self.reask_prompt is not None),
+                        (
+                            "custom_reask_instructions",
+                            self.reask_instructions is not None,
+                        ),
+                    ],
+                    is_parent=True,  # It will have children
+                    has_parent=False,  # Has no parents
+                )
+
             self.configure(final_num_reasks)
             if self.num_reasks is None:
                 raise RuntimeError(

guardrails/run.py

@@ -6,6 +6,7 @@
 from pydantic import BaseModel
 
 from guardrails.classes.history import Call, Inputs, Iteration, Outputs
+from guardrails.cli_dir.hub.credentials import Credentials
 from guardrails.datatypes import verify_metadata_requirements
 from guardrails.llm_providers import (
     AsyncPromptCallableBase,
@@ -17,6 +18,7 @@
 from guardrails.prompt import Instructions, Prompt
 from guardrails.schema import Schema, StringSchema
 from guardrails.utils.exception_utils import UserFacingException
+from guardrails.utils.hub_telemetry_utils import HubTelemetry
 from guardrails.utils.llm_response import LLMResponse
 from guardrails.utils.openai_utils import OPENAI_VERSION
 from guardrails.utils.reask_utils import (
@@ -103,6 +105,17 @@ def __init__(
         self.base_model = base_model
         self.full_schema_reask = full_schema_reask
 
+        # Get metrics opt-out from credentials
+        self._disable_tracer = Credentials.from_rc_file().no_metrics
+        if self._disable_tracer.strip().lower() == "true":
+            self._disable_tracer = True
+        elif self._disable_tracer.strip().lower() == "false":
+            self._disable_tracer = False
+
+        if not self._disable_tracer:
+            # Get the HubTelemetry singleton
+            self._hub_telemetry = HubTelemetry()
+
     def __call__(self, call_log: Call, prompt_params: Optional[Dict] = None) -> Call:
         """Execute the runner by repeatedly calling step until the reask budget
         is exhausted.
@@ -195,6 +208,17 @@ def __call__(self, call_log: Call, prompt_params: Optional[Dict] = None) -> Call
                         prompt_params=prompt_params,
                         include_instructions=include_instructions,
                     )
+
+                # Log how many times we reasked
+                # Use the HubTelemetry singleton
+                if not self._disable_tracer:
+                    self._hub_telemetry.create_new_span(
+                        span_name="/reasks",
+                        attributes=[("reask_count", index)],
+                        is_parent=False,  # This span has no children
+                        has_parent=True,  # This span has a parent
+                    )
+
         except UserFacingException as e:
             # Because Pydantic v1 doesn't respect property setters
             call_log._exception = e.original_exception

guardrails/utils/hub_telemetry_utils.py

@@ -0,0 +1,122 @@
+# Imports
+import logging
+
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import (  # HTTP Exporter
+    OTLPSpanExporter,
+)
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+
+
+class HubTelemetry:
+    """Singleton class for initializing a tracer for Guardrails Hub."""
+
+    _instance = None
+    _service_name = None
+    _endpoint = None
+    _tracer_name = None
+    _resource = None
+    _tracer_provider = None
+    _processor = None
+    _tracer = None
+    _prop = None
+    _carrier = {}
+
+    def __new__(
+        cls,
+        service_name: str = "guardrails-hub",
+        tracer_name: str = "gr_hub",
+        export_locally: bool = False,
+    ):
+        if cls._instance is None:
+            logging.debug("Creating HubTelemetry instance...")
+            cls._instance = super(HubTelemetry, cls).__new__(cls)
+            logging.debug("Initializing HubTelemetry instance...")
+            cls._instance.initialize_tracer(service_name, tracer_name, export_locally)
+        else:
+            logging.debug("Returning existing HubTelemetry instance...")
+        return cls._instance
+
+    def initialize_tracer(
+        self,
+        service_name: str,
+        tracer_name: str,
+        export_locally: bool,
+    ):
+        """Initializes a tracer for Guardrails Hub."""
+
+        self._service_name = service_name
+        # self._endpoint = "http://localhost:4318/v1/traces"
+        self._endpoint = (
+            "https://hty0gc1ok3.execute-api.us-east-1.amazonaws.com/v1/traces"
+        )
+        self._tracer_name = tracer_name
+
+        # Create a resource
+        # Service name is required for most backends
+        self._resource = Resource(attributes={SERVICE_NAME: self._service_name})
+
+        # Create a tracer provider and a processor
+        self._tracer_provider = TracerProvider(resource=self._resource)
+
+        if export_locally:
+            self._processor = SimpleSpanProcessor(ConsoleSpanExporter())
+        else:
+            self._processor = SimpleSpanProcessor(
+                OTLPSpanExporter(endpoint=self._endpoint)
+            )
+
+        # Add the processor to the provider
+        self._tracer_provider.add_span_processor(self._processor)
+
+        # Set the tracer provider and return a tracer
+        trace.set_tracer_provider(self._tracer_provider)
+        self._tracer = trace.get_tracer(self._tracer_name)
+
+        self._prop = TraceContextTextMapPropagator()
+
+    def inject_current_context(self) -> None:
+        """Injects the current context into the carrier."""
+        self._prop.inject(carrier=self._carrier)
+
+    def extract_current_context(self):
+        """Extracts the current context from the carrier."""
+
+        context = self._prop.extract(carrier=self._carrier)
+        return context
+
+    def create_new_span(
+        self,
+        span_name: str,
+        attributes: list,
+        is_parent: bool,  # Inject current context if IS a parent span
+        has_parent: bool,  # Extract current context if HAS a parent span
+    ):
+        """Creates a new span within the tracer with the given name and
+        attributes.
+
+        If it's a parent span, the current context is injected into the carrier.
+        If it has a parent span, the current context is extracted from the carrier.
+        Both the conditions can co-exist e.g. a span can be a parent span which
+        also has a parent span.
+
+        Args:
+            span_name (str): The name of the span.
+            attributes (list): A list of attributes to set on the span.
+            is_parent (bool): True if the span is a parent span.
+            has_parent (bool): True if the span has a parent span.
+        """
+
+        with self._tracer.start_as_current_span(
+            span_name,
+            context=self.extract_current_context() if has_parent else None,
+        ) as span:
+            if is_parent:
+                # Inject the current context
+                self.inject_current_context()
+
+            for attribute in attributes:
+                span.set_attribute(attribute[0], attribute[1])