Upload 16 files

custom_models/groundedness_checker/evaluate_groundedness_model.py

@@ -0,0 +1,84 @@
+
+from llmgaurdrails.custom_models.groundedness_checker.pdf_data_chunker import process_pdf
+import pandas as pd
+from llmgaurdrails.custom_models.groundedness_checker.llm_based_qa_generator import LLMBasedQAGenerator
+import pickle 
+from llmgaurdrails.model_inference.groundedness_checker import GroundednessChecker
+from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, confusion_matrix
+
+
+def get_eval_data(eval_pdf_paths:list,
+                  regenerate=False,
+                  path_to_save='eval_dataset'):
+
+    
+    if regenerate:
+        print("regenerating")
+
+        # pdf_path =  # Replace with your PDF
+        pdf_paths = eval_pdf_paths
+        
+        all_chunks = []
+
+        for path in pdf_paths:
+            chunks = process_pdf(path)
+            all_chunks.append(chunks)
+
+        chunks_flattened = [x for xs in all_chunks for x in xs]
+        
+        qa_generator = LLMBasedQAGenerator()
+
+        dataset = qa_generator.generate_dataset(chunks_flattened ,persist_dataset=True,presisted_file_path=path_to_save)
+
+        return dataset
+    else:
+        if path_to_save:
+            dataset = pickle.load(open(path_to_save,'rb'))
+            return dataset
+        else:
+            raise ValueError("Please specify the path where the dataset was previously saved in the parameter 'path_to_save' ")
+
+def evaluate(dataset):
+    groundedness_checker = GroundednessChecker()
+    eval_df = pd.DataFrame(data=  dataset)
+
+    predictions = []
+    confidence_scores = []
+
+    for i,row in eval_df.iterrows(): 
+        groundedness_result = groundedness_checker.check(
+        question=row['question'],
+        answer=row['answer'],
+        context=row['context'])
+
+        predictions.append(groundedness_result['is_grounded'])
+        confidence_scores.append(groundedness_result['confidence'])
+
+    eval_df['predicted'] = predictions
+    eval_df['confidence'] = confidence_scores
+
+    accuracy = accuracy_score(eval_df['label'], eval_df['predicted'])
+    precision = precision_score(eval_df['label'], eval_df['predicted'])
+    recall = recall_score(eval_df['label'], eval_df['predicted'])
+    f1 = f1_score(eval_df['label'], eval_df['predicted'])
+    conf_matrix = confusion_matrix(eval_df['label'], eval_df['predicted'])
+
+    # Print the results
+    print("Accuracy:", accuracy)
+    print("Precision:", precision)
+    print("Recall:", recall)
+    print("F1 Score:", f1)
+    print("Confusion Matrix:\n", conf_matrix)
+
+
+# Usage
+if __name__ == "__main__":
+    dataset = get_eval_data(eval_pdf_paths=[["D:\Sasidhar\Projects\llm_gaurdrails\llmgaurdrails\data\CreditCard.pdf"]])
+    evaluate(dataset)
+
+    
+    
+  
+    
+
+    

custom_models/groundedness_checker/grounding_classifier.py

@@ -0,0 +1,63 @@
+import torch
+from dateutil.parser import parse as parse_date
+from sklearn.model_selection import train_test_split
+from transformers import (
+    pipeline,
+    AutoTokenizer,
+    AutoModelForSequenceClassification,
+    TrainingArguments,
+    Trainer
+)
+from torch.utils.data import Dataset
+
+class GroundingDataset(Dataset):
+    def __init__(self, data, tokenizer, max_length=512):
+        self.data = data
+        self.tokenizer = tokenizer
+        self.max_length = max_length
+        
+    def __len__(self):
+        return len(self.data)
+    
+    def __getitem__(self, idx):
+        item = self.data[idx]
+        encoding = self.tokenizer(
+            item["question"],
+            text_pair=item["answer"] + " [SEP] " + item["context"],
+            padding="max_length",
+            truncation=True,
+            max_length=self.max_length,
+            return_tensors="pt"
+        )
+        return {
+            "input_ids": encoding["input_ids"].squeeze(),
+            "attention_mask": encoding["attention_mask"].squeeze(),
+            "labels": torch.tensor(item["label"])
+        }
+
+class GroundingTrainer:
+    def __init__(self):
+        self.tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")
+        self.model = AutoModelForSequenceClassification.from_pretrained(
+            "distilbert-base-uncased", num_labels=2
+        )
+        
+    def train(self, dataset):
+        train_data, val_data = train_test_split(dataset, test_size=0.2)
+        
+        trainer = Trainer(
+            model=self.model,
+            args=TrainingArguments(
+                output_dir="./results",
+                num_train_epochs=3,
+                per_device_train_batch_size=8,
+                evaluation_strategy="epoch",
+                logging_dir="./logs"
+            ),
+            train_dataset=GroundingDataset(train_data, self.tokenizer),
+            eval_dataset=GroundingDataset(val_data, self.tokenizer)
+        )
+        
+        trainer.train()
+        self.model.save_pretrained("./grounding_detector")
+        self.tokenizer.save_pretrained("./grounding_detector")

custom_models/groundedness_checker/llm_based_qa_generator.py

@@ -0,0 +1,127 @@
+import numpy as np
+import re
+from tqdm import tqdm
+import json
+import pickle
+from llmgaurdrails.llms.openai_client import invoke_api
+
+class LLMBasedQAGenerator:
+
+    def _create_entry(self, context: str, question: str, answer: str, label: int) -> dict:
+        """Create standardized training entry with validation checks"""
+        # Clean and validate inputs
+        context = self._clean_text(context)
+        question = self._clean_text(question).rstrip("?") + "?"
+        answer = self._clean_answer(answer)
+        
+        if not question or not answer:
+            return None
+            
+        return {
+            "context": context,
+            "question": question,
+            "answer": answer,
+            "label": int(bool(label)),  # Force 0/1 encoding
+            "meta": {
+                "context_hash": hash(context),
+                "answer_type": self._classify_answer_type(answer),
+                "question_type": self._classify_question(question)
+            }
+        }
+    
+    def _clean_text(self, text: str) -> str:
+        """Basic text normalization"""
+        return re.sub(r'\s+', ' ', text).strip()
+    
+    def _clean_answer(self, answer: str) -> str:
+        """Answer-specific cleaning"""
+        answer = self._clean_text(answer)
+        if answer.lower() in ["", "n/a", "unknown"]:
+            return "[INVALID]"
+        return answer
+    
+    def _classify_answer_type(self, answer: str) -> str:
+        """Categorize answers for analysis"""
+        if "$" in answer: return "monetary"
+        if "%" in answer: return "percentage"
+        if any(c.isdigit() for c in answer): return "numeric"
+        return "textual"
+    
+    def _classify_question(self, question: str) -> str:
+        """Identify question types"""
+        q = question.lower()
+        if "how much" in q: return "quantity"
+        if "when" in q: return "temporal"
+        if "why" in q: return "reason"
+        return "factual"
+
+    
+    def _generate_questions_and_grounded_answers(self,chunk,num_questions=3):
+        
+        questions = []
+        answers =[]
+        # Generate a question and a grounded answer
+        for i in range(num_questions):
+            try:
+                grounded_system_prompt = """You are a helpful assistant that generates questions and answers based on the given context.
+                                            The question and answer should not exceed 15 words each.
+                                            The response should ne a json with 'question' and 'answer as the key'"""
+                grounded_message = f"Context: {chunk}\n\nGenerate a question and a grounded answer based on this context."
+                grounded_qa_response = invoke_api(grounded_system_prompt,grounded_message,0.7,max_tokens=100)
+
+                # print("Question:",grounded_qa_response)
+                grounded_qa = json.loads(grounded_qa_response.choices[0].message.content.strip("```json"))
+                
+                questions.append(grounded_qa['question'])
+                answers.append(grounded_qa['answer'])
+            except:
+                print("errored")
+                questions.append('')
+                answers.append('')
+
+        return questions,answers
+
+    def _generate_ungrounded_answer(self,chunk,question,grounded_answer):
+        
+        try:
+            ungrounded_system_prompt = """You are a helpful assistant that generates questions and ungrounded answers that are  based on the given context. But factually or logically incorrect.
+                                        The 'answer' part of the response should not exceed 15 words each.
+                                        The response should ne a json with just one key 'answer'"""
+            ungrounded_message = f"Question: {question}\n\nGenerate an ungrounded answer based on the original context {chunk}. Make subtle changes to the actual answer to make it look plausible"
+
+            ungrounded_answer_response = invoke_api(ungrounded_system_prompt,ungrounded_message,0.7,max_tokens=30)
+            # print("answer:",ungrounded_answer_response)
+            answer_json = json.loads(ungrounded_answer_response.choices[0].message.content.strip("```json"))
+            return answer_json['answer']
+        except:
+            print("errored in answer")
+            return ''
+        
+    def generate_dataset(self, chunks: list,
+                         persist_dataset:bool =False,
+                         presisted_file_path: str = "training_data") -> list:
+        
+        dataset = []
+
+        for chunk_dict in tqdm(chunks, desc="Generating QA pairs"):
+            
+            chunk = chunk_dict['text']
+
+            if not chunk.strip():
+                continue
+                
+            questions,grounded_answers = self._generate_questions_and_grounded_answers(chunk)
+
+            for question,grounded_answer in zip(questions,grounded_answers):
+                if not question.strip():
+                    continue
+                
+                ungrounded = self._generate_ungrounded_answer(chunk, question,grounded_answer)
+                
+                dataset.append(self._create_entry(chunk, question, grounded_answer, 1))
+                dataset.append(self._create_entry(chunk, question, ungrounded, 0))
+        
+        if persist_dataset:
+            pickle.dump(dataset,open(presisted_file_path,'ab'))
+
+        return dataset

custom_models/groundedness_checker/main.py

@@ -0,0 +1,47 @@
+
+from llmgaurdrails.custom_models.groundedness_checker.pdf_data_chunker import process_pdf
+from llmgaurdrails.custom_models.groundedness_checker.llm_based_qa_generator import  LLMBasedQAGenerator
+from llmgaurdrails.custom_models.groundedness_checker.grounding_classifier import GroundingTrainer
+from llmgaurdrails.custom_models.groundedness_checker.simple_qa_generator import SimpleQAGenerator
+from llmgaurdrails.custom_models.groundedness_checker.evaluate_groundedness_model import evaluate,get_eval_data
+
+# Usage
+if __name__ == "__main__":
+    
+    # pdf_path =  # Replace with your PDF
+    trainning_pdf_paths = ["D:\Sasidhar\Projects\cba\data\CreditCard.pdf" ,
+                 "D:\Sasidhar\Projects\cba\data\home_insurance_pds.pdf"]
+    
+    eval_pdf_paths = ["D:\Sasidhar\Projects\llm_gaurdrails\llmgaurdrails\data\CreditCard.pdf"]
+    
+    all_chunks = []
+
+    for path in trainning_pdf_paths:
+        chunks = process_pdf(trainning_pdf_paths[0])
+        all_chunks.append(chunks)
+
+    chunks_flattened = [x for xs in all_chunks for x in xs]
+    
+    # generate qa dataset 
+    qa_generator = LLMBasedQAGenerator()
+
+    dataset = qa_generator.generate_dataset(chunks_flattened,persist_dataset=True)
+
+    trainer = GroundingTrainer()
+    trainer.train(dataset)
+
+    eval_dataset = get_eval_data(eval_pdf_paths=eval_pdf_paths)
+    evaluate(dataset)
+    # Accuracy: 0.8952380952380953
+    # Precision: 0.8738738738738738
+    # Recall: 0.9238095238095239
+    # F1 Score: 0.8981481481481481
+
+
+
+    
+    
+  
+    
+
+    

custom_models/groundedness_checker/pdf_data_chunker.py

@@ -0,0 +1,119 @@
+import pdfplumber
+import re
+from transformers import AutoTokenizer
+from typing import List, Dict
+import pandas as pd
+
+tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")
+
+# We try to extract the section and subsection data  along with the text to be appended to the chunk
+def extract_text_with_hierarchy(pdf_path: str) -> List[Dict]:
+    """Extract text with section/subsection hierarchy"""
+    content = []
+    current_section = ""
+    current_subsection = ""
+    
+    with pdfplumber.open(pdf_path) as pdf:
+        for page in pdf.pages:
+            text = page.extract_text()
+            lines = text.split('\n')
+            
+            for line in lines:
+                line = line.strip()
+                if not line:
+                    continue
+                
+                # Detect section headers
+                section_match = re.match(r'\\section\*{(.+?)}', line)
+                subsection_match = re.match(r'\\subsection\*{(.+?)}', line)
+                
+                if section_match:
+                    current_section = section_match.group(1)
+                    current_subsection = ""
+                    content.append({
+                        'type': 'section',
+                        'title': current_section,
+                        'text': ""
+                    })
+                elif subsection_match:
+                    current_subsection = subsection_match.group(1)
+                    content.append({
+                        'type': 'subsection',
+                        'title': current_subsection,
+                        'text': ""
+                    })
+                else:
+                    if content:
+                        content[-1]['text'] += line + " "
+                    else:
+                        content.append({
+                            'type': 'text',
+                            'title': "",
+                            'text': line
+                        })
+    
+    return content
+
+def create_bert_chunks(file_name:str,content: List[Dict], max_tokens=450, overlap=50) -> List[Dict]:
+    """Create chunks optimized for DistilBERT with hierarchy context"""
+    chunks = []
+    current_chunk = []
+    current_tokens = 0
+    current_section = ""
+    current_subsection = ""
+    
+    for item in content:
+        # Build context header
+        header = ""
+        if item['type'] == 'section':
+            current_section = item['title']
+            current_subsection = ""
+            header = f"[SECTION] {current_section}\n"
+        elif item['type'] == 'subsection':
+            current_subsection = item['title']
+            header = f"[SUBSECTION] {current_subsection}\n"
+        
+        # Split text into sentences
+        sentences = re.split(r'(?<=[.!?])\s+', item['text'])
+        
+        for sentence in sentences:
+            full_text = header + sentence if header else sentence
+            tokens = tokenizer.encode(full_text)
+            
+            if current_tokens + len(tokens) > max_tokens:
+                if current_chunk:
+                    chunk_text = "\n".join(current_chunk)
+                    chunks.append({
+                        'text': chunk_text,
+                        'section': current_section,
+                        'subsection': current_subsection,
+                        'tokens': current_tokens,
+                        'file_name':file_name
+                    })
+                    # Carry over overlap
+                    overlap_tokens = tokenizer.encode(chunk_text)[-overlap:]
+                    current_chunk = [tokenizer.decode(overlap_tokens)]
+                    current_tokens = len(overlap_tokens)
+                    header = ""  # Reset header after overlap
+                
+            current_chunk.append(full_text)
+            current_tokens += len(tokens)
+            header = ""  # Clear header after first use
+    
+    # Add remaining content
+    if current_chunk:
+        chunk_text = "\n".join(current_chunk)
+        chunks.append({
+            'text': chunk_text,
+            'section': current_section,
+            'subsection': current_subsection,
+            'tokens': current_tokens,
+            'file_name':file_name
+        })
+    
+    return chunks
+
+def process_pdf(pdf_path: str) -> List[Dict]:
+    """Process PDF into BERT-optimized chunks"""
+    structured_content = extract_text_with_hierarchy(pdf_path)
+    return create_bert_chunks(pdf_path,structured_content)

custom_models/groundedness_checker/simple_qa_generator.py

@@ -0,0 +1,117 @@
+import spacy
+import torch
+import random
+import numpy as np
+import re
+from tqdm import tqdm
+from transformers import pipeline
+from llmgaurdrails.custom_models.groundedness_checker.ungrounded_answer_generator import UngroundedAnswerGenerator
+from llmgaurdrails.llms.openai_client import invoke_api
+
+#  A Simple QA Generator that generates a question and answer based on a given context. This is based on a fine tuned model on a QA dataset
+class SimpleQAGenerator:
+    def __init__(self):
+        self.qg_model = pipeline(
+            "text2text-generation", 
+            model="valhalla/t5-base-qa-qg-hl",
+            device=0 if torch.cuda.is_available() else -1
+        )
+        self.ungrounded_gen = UngroundedAnswerGenerator()
+
+        self.nlp = spacy.load("en_core_web_sm")
+    
+    def _create_entry(self, context: str, question: str, answer: str, label: int) -> dict:
+        """Create standardized training entry with validation checks"""
+        # Clean and validate inputs
+        context = self._clean_text(context)
+        question = self._clean_text(question).rstrip("?") + "?"
+        answer = self._clean_answer(answer)
+        
+        if not question or not answer:
+            return None
+            
+        return {
+            "context": context,
+            "question": question,
+            "answer": answer,
+            "label": int(bool(label)),  # Force 0/1 encoding
+            "meta": {
+                "context_hash": hash(context),
+                "answer_type": self._classify_answer_type(answer),
+                "question_type": self._classify_question(question)
+            }
+        }
+    
+    def _clean_text(self, text: str) -> str:
+        """Basic text normalization"""
+        return re.sub(r'\s+', ' ', text).strip()
+    
+    def _clean_answer(self, answer: str) -> str:
+        """Answer-specific cleaning"""
+        answer = self._clean_text(answer)
+        if answer.lower() in ["", "n/a", "unknown"]:
+            return "[INVALID]"
+        return answer
+    
+    def _classify_answer_type(self, answer: str) -> str:
+        """Categorize answers for analysis"""
+        if "$" in answer: return "monetary"
+        if "%" in answer: return "percentage"
+        if any(c.isdigit() for c in answer): return "numeric"
+        return "textual"
+    
+    def _classify_question(self, question: str) -> str:
+        """Identify question types"""
+        q = question.lower()
+        if "how much" in q: return "quantity"
+        if "when" in q: return "temporal"
+        if "why" in q: return "reason"
+        return "factual"
+
+    def generate_dataset(self, chunks: list) -> list:
+        dataset = []
+        for chunk_dict in tqdm(chunks, desc="Generating QA pairs"):
+            
+            chunk = chunk_dict['text']
+
+            if not chunk.strip():
+                continue
+                
+            questions = self._generate_questions(chunk)
+            for question in questions:
+                if not question.strip():
+                    continue
+                
+                grounded = self._get_grounded_answer(chunk, question)
+                ungrounded = self.ungrounded_gen.generate(chunk, grounded)
+                
+                dataset.append(self._create_entry(chunk, question, grounded, 1))
+                dataset.append(self._create_entry(chunk, question, ungrounded, 0))
+        
+        return dataset
+    
+    def _generate_questions(self, context: str) -> list:
+        try:
+            output = self.qg_model(
+                f"generate questions: {context}",
+                max_length=64,
+                num_return_sequences=3,
+                do_sample=True,
+                temperature=0.9
+            )
+            return [q['generated_text'].strip() for q in output]
+        except:
+            return []
+
+    def _get_grounded_answer(self, context: str, question: str) -> str:
+
+        try:
+            answer = self.qg_model(
+                f"answer: {context} question: {question}",
+                max_length=64,
+                num_beams=1
+            )[0]['generated_text'].strip()
+            return answer if answer else "[No Answer]"
+        except:
+            return "[No Answer]"
+        

custom_models/groundedness_checker/ungrounded_answer_generator.py

@@ -0,0 +1,111 @@
+import re
+import spacy 
+from sentence_transformers import SentenceTransformer
+import numpy as np
+import random
+from datetime import datetime, timedelta
+from dateutil.parser import parse as parse_date
+
+# A simplistic Ungrounded Answer Generator.
+
+class UngroundedAnswerGenerator:
+    def __init__(self):
+        self.nlp = spacy.load("en_core_web_sm")
+        self.sim_model = SentenceTransformer('all-MiniLM-L6-v2')
+
+        # 
+        self.financial_terms = [
+                "CommBank Credit Card",
+                "Personal credit cards",
+                "Business credit cards",
+                "PIN",
+                "ePayments Code",
+                "Conditions of Use",
+                "Schedule of Credit Card Particulars",
+                "Banking Code of Practice",
+                "NetBank",
+                "CommBank app",
+                "Electronic Banking Terms and Conditions",
+                "Tap & Pay",
+                "cash advance",
+                "credit limit",
+                "ATM cash withdrawals",
+                "international transaction fee",
+                "Mastercard",
+                "Visa",
+                "balance transfers",
+                "regular payments",
+                "additional cardholder",
+                "digital wallet",
+                "statements and notices",
+                "closing balance",
+                "minimum payment",
+                "interest-free period on purchases",
+                "SurePay instalment plan",
+                "AutoPay",
+                "fees and interest rates",
+                "annual interest rates",
+                "daily interest rate",
+                "statement period",
+                "balance transfer period",
+                "unauthorised transaction",
+                "card scheme refunds",
+                "purchase plan",
+                "card balance plan",
+                "cash advance balance plan",
+                "instalment setup fee",
+                "purchase balance",
+                "cash advances balance",
+                "interest rate for the plan",
+                "credit card account",
+                "default under your contract"
+            ]
+
+
+    
+    def generate(self, context: str, answer: str) -> str:
+        strategy = self._select_strategy(answer)
+        return strategy(context, answer)
+    
+    def _select_strategy(self, answer: str):
+        doc = self.nlp(answer)
+        ents = [ent.label_ for ent in doc.ents]
+        
+        if "DATE" in ents:
+            return self._perturb_dates
+        if any(e in ["MONEY", "PERCENT"] for e in ents):
+            return self._perturb_numbers
+        
+        return self._semantic_distractor
+    
+    def _perturb_numbers(self, context: str, answer: str) -> str:
+        if "$" in answer:
+            base = self._extract_number(answer)
+            return f"${base * random.uniform(0.8, 1.2):.2f}"
+        elif "%" in answer:
+            base = self._extract_number(answer)
+            return f"{base * random.uniform(0.5, 1.5):.1f}%"
+        return answer
+    
+    def _perturb_dates(self, context: str, answer: str) -> str:
+        try:
+            dt = parse_date(answer)
+            if dt:
+                delta = timedelta(days=random.randint(-30, 30))
+                return (dt + delta).strftime("%Y-%m-%d")
+        except:
+            pass
+        return answer
+    
+    def _semantic_distractor(self, context: str, answer: str) -> str:
+        answer_emb = self.sim_model.encode(answer)
+        term_embs = self.sim_model.encode(self.financial_terms)
+        similarities = np.dot(term_embs, answer_emb)
+        return self.financial_terms[np.argsort(similarities)[-2]]
+    
+    def _extract_number(self, text: str) -> float:
+        try:
+            return float(re.search(r"\d+\.?\d*", text).group())
+        except:
+            return random.uniform(1, 1000)
+

custom_models/setup.sh

@@ -0,0 +1 @@
+python -m spacy download en_core_web_sm

endpoints/api_models.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel
+
+class OutputGuardrailsConfig(BaseModel):
+    contextual_grounding: bool = True
+    toxicity: bool = True
+    
+    # Extend with more flags for additional guardrails
+
+# Define the input that went to LLM and its response.
+class LLMResponse(BaseModel):
+    question: str
+    answer: str
+    context: str
+
+# GaurdRail Check Input Model
+class CheckRequest(BaseModel):
+    llm_response: LLMResponse
+    config:  OutputGuardrailsConfig =  OutputGuardrailsConfig()  # Default config if not provided
+
+# GaurdRail Check Response
+class CheckResponse(BaseModel):
+    grounded: bool
+    details: dict
+

endpoints/gaurdrails.py

@@ -0,0 +1,14 @@
+from endpoints.api_models import CheckResponse,CheckRequest
+from fastapi import APIRouter
+from model_inference.gaurdrails_manager import GuardrailsManager    
+
+router = APIRouter(prefix="/gaurdrails", tags=["Gaurdrails"])
+
+# Define the POST endpoint for guardrail checking.
+@router.post("/check" ,response_model=CheckResponse)
+async def check_guardrails(request: CheckRequest):
+    
+    manager = GuardrailsManager(request.config)
+    result = manager.check(request.llm_response)
+
+    return CheckResponse(grounded=result.grounded(), details=result.details)

endpoints/groundedness.py

@@ -0,0 +1,15 @@
+# endpoints/groundedness.py
+from fastapi import APIRouter
+from pydantic import BaseModel
+from model_inference.groundedness_checker import GroundednessChecker
+from endpoints.api_models import LLMResponse
+
+router = APIRouter(prefix="/groundedness", tags=["Groundedness"])
+
+
+checker = GroundednessChecker(model_path="./grounding_detector")
+
+@router.post("/check")
+async def check_groundedness(req: LLMResponse):
+    result = checker.check(req.question, req.answer, req.context)
+    return result