🚧 first draft of wrapping three guardrails-ai inside the detector logic

Author: m-misiura

detectors/guardrails_ai_wrapper/__init__.py

@@ -0,0 +1,55 @@
+import os
+import sys
+
+sys.path.insert(0, os.path.abspath(".."))
+from common.app import logger
+from scheme import (
+    ContentAnalysisHttpRequest,
+    ContentAnalysisResponse,
+    ContentsAnalysisResponse,
+)
+
+from guardrails import Guard, OnFailAction
+from guardrails.hub import RegexMatch, CompetitorCheck, ToxicLanguage
+
+
+class Detector:
+    def __init__(self):
+        self.guard = Guard().use_many(
+            RegexMatch(regex=r"^(?!.*\bpotatoe\b).*$", on_fail=OnFailAction.EXCEPTION),
+            CompetitorCheck(
+                ["Apple", "Microsoft", "Google"], on_fail=OnFailAction.EXCEPTION
+            ),
+            ToxicLanguage(
+                threshold=0.5,
+                validation_method="sentence",
+                on_fail=OnFailAction.EXCEPTION,
+            ),
+        )
+        logger.info("Guardrails AI Wrapper initialized")
+
+    def run(self, input: ContentAnalysisHttpRequest) -> ContentsAnalysisResponse:
+        contents_analyses = []
+        for text in input.contents:
+            content_analyses = []
+            try:
+                logger.info(f"Validating text: {text}")
+                validation_result = self.guard.validate(text)
+                logger.info(f"Validation successful for text: {text}")
+            except Exception as e:
+                logger.error(f"Validation failed for text: {text} with error: {e}")
+                validation_result = e
+
+            content_analyses.append(
+                ContentAnalysisResponse(
+                    start=0,
+                    end=len(text),
+                    detection_type="guardrails_ai",
+                    text=text,
+                    validation_result=validation_result,
+                    evidences=[],
+                )
+            )
+            contents_analyses.append(content_analyses)
+        logger.info("Content analysis completed")
+        return contents_analyses

detectors/guardrails_ai_wrapper/app.py

@@ -0,0 +1,63 @@
+from enum import Enum
+from typing import List, Optional, Union
+from pydantic import BaseModel, Field, RootModel, ConfigDict
+from guardrails.classes.validation_outcome import ValidationOutcome
+
+
+class Evidence(BaseModel):
+    source: str = Field(
+        title="Source",
+        example="https://en.wikipedia.org/wiki/IBM",
+        description="Source of the evidence, it can be url of the evidence etc",
+    )
+
+
+class EvidenceType(str, Enum):
+    url = "url"
+    title = "title"
+
+
+class EvidenceObj(BaseModel):
+    type: EvidenceType = Field(
+        title="EvidenceType",
+        example="url",
+        description="Type field signifying the type of evidence provided. Example url, title etc",
+    )
+    evidence: Evidence = Field(
+        description="Evidence object, currently only containing source, but in future can contain other optional arguments like id, etc",
+    )
+
+
+class ContentAnalysisHttpRequest(BaseModel):
+    contents: List[str] = Field(
+        min_length=1,
+        title="Contents",
+        description="Field allowing users to provide list of texts for analysis. Note, results of this endpoint will contain analysis / detection of each of the provided text in the order they are present in the contents object.",
+        example=["My email address is [email protected] and [email protected]"],
+    )
+
+
+class ContentAnalysisResponse(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    start: int = Field(example=14)
+    end: int = Field(example=26)
+    detection_type: str = Field(example="detection_type")
+    text: str = Field(
+        example="My email address is [email protected] and [email protected]"
+    )
+    validation_result: Union[Exception, ValidationOutcome]
+    evidences: Optional[List[EvidenceObj]] = Field(
+        description="Optional field providing evidences for the provided detection",
+        default=[],
+    )
+
+
+class ContentsAnalysisResponse(RootModel):
+    root: List[List[ContentAnalysisResponse]] = Field(
+        title="Response Text Content Analysis Unary Handler Api V1 Text Content Post"
+    )
+
+
+class Error(BaseModel):
+    code: int
+    message: str