Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, HTTPException | |
from pydantic import BaseModel | |
import re | |
app = FastAPI() | |
# Configuration for enabled guardrail checks as part of the request payload. | |
class GuardrailsConfig(BaseModel): | |
factual_consistency: bool = True | |
toxicity: bool = True | |
# Extend with more flags for additional guardrails | |
# Request model now includes both the response and the configuration. | |
class CheckRequest(BaseModel): | |
response: str | |
config: GuardrailsConfig = GuardrailsConfig() # Default config if not provided | |
class CheckResponse(BaseModel): | |
grounded: bool | |
details: dict | |
# A simple result class to hold individual check outcomes. | |
class Result: | |
def __init__(self): | |
self.details = {} | |
def add(self, rule_name: str, passed: bool): | |
self.details[rule_name] = passed | |
def grounded(self) -> bool: | |
# The response is considered "grounded" if all enabled rules pass. | |
return all(self.details.values()) | |
# Define guardrail rule classes. | |
class FactualConsistencyRule: | |
name = "FactualConsistency" | |
def check(self, response_text: str) -> bool: | |
# For demonstration: pass if the response contains the word "fact". | |
return "fact" in response_text.lower() | |
class ToxicityRule: | |
name = "Toxicity" | |
def check(self, response_text: str) -> bool: | |
# For demonstration: fail if negative words like "hate" or "kill" are found. | |
return not re.search(r"(hate|kill)", response_text, re.IGNORECASE) | |
# Manager class to load and execute the enabled guardrail rules. | |
class GuardrailsManager: | |
def __init__(self, config: GuardrailsConfig): | |
self.config = config | |
self.rules = self.load_rules() | |
def load_rules(self): | |
rules = [] | |
if self.config.factual_consistency: | |
rules.append(FactualConsistencyRule()) | |
if self.config.toxicity: | |
rules.append(ToxicityRule()) | |
# Add additional rules based on configuration here. | |
return rules | |
def check(self, response_text: str) -> Result: | |
result = Result() | |
for rule in self.rules: | |
rule_result = rule.check(response_text) | |
result.add(rule.name, rule_result) | |
return result | |
# Define the POST endpoint for guardrail checking. | |
async def check_guardrails(request: CheckRequest): | |
manager = GuardrailsManager(request.config) | |
result = manager.check(request.response) | |
return CheckResponse(grounded=result.grounded(), details=result.details) | |
def home(): | |
return {"Msg": "This is a LLM Gaurdrails!"} | |
def test(): | |
return {"Msg": "This is a Test!"} | |