scripts/mintoak/run_eval_transformers.py

import os
import re
import json
import time
import torch
import chromadb
from chromadb.utils import embedding_functions
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel

# Constants matching app.py
CHROMA_DB_PATH = "data/mintoak/chroma_db"
COLLECTION_NAME = "mintoak_content"
MODEL_PATH = "Qwen/Qwen2.5-1.5B-Instruct"
ADAPTER_PATH = "adapters/mintoak"

PROFANE_PATTERNS = [
    "fuck", "shit", "asshole", "bitch", "bastard", "cunt", "dick", "pussy",
    "idiot", "stupid", "dumbass", "kill yourself", "kys", "retard",
    "wanker", "motherfucker"
]
INJECTION_PATTERNS = [
    "ignore previous", "system prompt", "developer mode", "you are now",
    "pretend to be", "reveal instructions", "system instruction", "dan mode",
    "hypothetically speaking", "override rules", "jailbreak"
]
GREETING_RE = re.compile(r"^(hi+|hello+|hey+|yo+|greetings|good\s+(morning|afternoon|evening))\b", re.I)
IDENTITY_RE = re.compile(r"\b(who are (you|u)|what is your name|what'?s your name|what do you do|introduce yourself|tell me about yourself)\b", re.I)

OPTIMIZER_PROMPT = (
    "You are a search query optimizer. Your job is to translate user questions into database search keywords. "
    "Translate synonyms into our core product areas: 'onboarding/acquisition' -> 'DigiOnboard', "
    "'payments' -> 'SmartPayments', 'voice/soundbox' -> 'SoundHub', 'cross-sell' -> 'SellSmart'. "
    "Output ONLY the keywords, separated by spaces. Do not write explanations or punctuation."
)

SYSTEM_PROMPT = (
    "You are the Mintoak Knowledge Assistant, a senior, business-aware digital advisor on the Mintoak website. "
    "Your tone is professional, clear, confident, technology-forward, and benefit-led. Do not use hype, casualness, or over-enthusiasm. "
    "Do not use emojis by default; use at most one emoji only where it adds clarity (e.g. 💡, 📈) and never in pricing, credibility, or trust contexts. "
    "Answer strictly using the provided context. Never invent, assume, or pull from outside knowledge. "
    "Mintoak is strictly a bank-led B2B platform for merchants; Mintoak does NOT issue loyalty points or rewards to end-consumers. If asked how customers or end-consumers earn loyalty points, clarify that Mintoak does not directly run consumer loyalty programs, but enables merchants to set up their own custom campaigns and offers for their customers. Do not confuse B2B merchant app users with B2C end-consumers. "
    "Start your response directly and naturally—do NOT use robotic preambles like 'As per the context' or 'Based on the provided context'. "
    "Frame Mintoak as a white-labeled, modular, cloud-native, API-first SaaS platform/solution provider (never as a 'software company', 'software', 'tool', or 'app'). "
    "Avoid these banned phrases: 'seamless', 'empowering', 'leverage' (as a verb), 'game-changer', 'synergy', 'unlocking', 'in today\'s fast-paced world', 'as we move forward'. "
    "Follow this exact structure: Context (direct answer) -> Insight (translate features into merchant/business outcomes like activation, cost-to-acquire, time-to-market) -> Direction (nudge to the next best page/demo and close with the source URL). "
    "Keep responses highly concise (maximum 2-3 sentences or short bullets) unless explicitly asked for detail. "
    "If the context does not contain the answer, reply exactly: 'The requested information does not currently exist on www.mintoak.com. You can get in touch with our team at https://www.mintoak.com/contact-us.' "
    "IMPORTANT - Lead Capture Flow: If the user requests a demo, pricing, partnership, or contact details, give a brief helpful response and politely ask for their name and email. "
    "If and ONLY if the user has already provided their name and email in the conversation history, confirm follow-up details and append exactly '[CAPTURE_LEAD]' at the very end of your response."
)

QUERY_ENHANCEMENT_RULES = [
    {
        "keywords": ["gpv", "gross payment", "volume", "process", "transaction", "festive"],
        "expansion": "During India's October 2025 festive period, Mintoak's platform recorded 44% year-on-year growth in digital payment value across its merchant base",
        "prompt_note": "State that Mintoak processes payments for over 4 million merchants, recording a 44% YoY growth in digital payment value and 42% YoY growth in digital payment volume during the October 2025 festive period.",
        "top_k": 3
    },
    {
        "keywords": ["document", "documents", "checklist", "paperwork", "certificates"],
        "expansion": "What documents are needed for merchant onboarding? DigiOnboard KYC KYB",
        "prompt_note": "Ensure you list the specific documents needed: registration certificates, GST certificates, cancelled cheques, and board resolutions.",
        "top_k": 3
    },
    {
        "keywords": ["acquire merchant", "merchant acquisition", "onboard", "onboarding", "kyc", "kyb"],
        "expansion": "Mintoak DigiOnboard merchant onboarding KYC KYB acquisition banks",
        "prompt_note": None,
        "top_k": 2
    },
    {
        "keywords": ["rewardrun", "reward run", "gamified campaign", "gamified campaigns", "loyalty campaign", "loyalty campaigns"],
        "expansion": "Mintoak RewardRun gamified loyalty and merchant engagement platform for acquirers",
        "override": True,
        "prompt_note": "Ensure you mention Mintoak RewardRun as the gamified loyalty and merchant engagement platform.",
        "top_k": 3
    },
    {
        "keywords": ["job", "jobs", "career", "careers", "hiring", "opening", "openings", "join us", "work at mintoak"],
        "expansion": "Mintoak careers job openings hiring opportunities join the team",
        "prompt_note": "Direct the user to the careers page at https://www.mintoak.com/career.",
        "top_k": 2
    },
    {
        "keywords": ["address", "office", "offices", "headquarters", "location", "locations", "located", "contact us", "contact-us"],
        "expansion": "Mintoak office locations and addresses Mumbai Bangalore Dubai contact-us",
        "override": True,
        "prompt_note": "List all Mintoak office locations (Mumbai, Bangalore, Dubai) with their Google Maps links.",
        "top_k": 2
    },
    {
        "keywords": ["cto", "technology head", "tech head", "head of technology", "head of engineering", "engineering head", "who is your technology head", "who is the technology head"],
        "expansion": "Kabeer Jain is the Chief Technology Officer (CTO) and Co-founder of Mintoak.",
        "override": True,
        "prompt_note": "State that Kabeer Jain is Mintoak's Chief Technology Officer (CTO) and Co-founder.",
        "top_k": 2
    },
    {
        "keywords": ["ceo", "founder", "founders", "founded", "cpo", "leadership", "team", "management", "raman", "khanduja", "sanjay", "nazareth", "rama", "tadepalli", "kabeer", "jain", "rohit", "ramana"],
        "expansion": "Mintoak co-founders and executive leadership team: Raman Khanduja (CEO), Sanjay Nazareth (COO), Rama Tadepalli (CPO), Kabeer Jain (CTO), and Rohit Ramana (CFO). Nilesh Lonkar is the VP of Engineering.",
        "override": True,
        "prompt_note": "Ensure you mention Mintoak's co-founders: Raman Khanduja (CEO), Sanjay Nazareth (COO), Rama Tadepalli (CPO), Kabeer Jain (CTO), and Rohit Ramana (CFO).",
        "top_k": 2
    }
]

def clean_response(text: str) -> str:
    preambles = [
        re.compile(r"^(?:as per|according to|based on)\s+(?:the\s+)?(?:context|provided context|mintoak documentation|documentation)(?:,\s*)?", re.I),
        re.compile(r"^based on what is provided in the context(?:,\s*)?", re.I),
        re.compile(r"^from the context(?:,\s*)?", re.I),
        re.compile(r"^Based on the Mintoak documentation,\s*(?:here is what we know about\s+[^:\n]+:)?\s*", re.I),
    ]
    for rx in preambles:
        text = rx.sub("", text).strip()
    text = re.sub(r'(?i)(?:\n\s*)*👉?\s*For\s+more\s+details,?\s+visit:?.*$', '', text).strip()
    return text.replace("[CAPTURE_LEAD]", "").strip()

def retrieve_context(collection, query: str):
    query_lower = query.lower()
    search_keywords = None
    prompt_note = None
    top_k = 2

    for rule in QUERY_ENHANCEMENT_RULES:
        if any(kw in query_lower for kw in rule["keywords"]):
            if rule.get("override"):
                search_keywords = rule["expansion"]
            else:
                search_keywords = f"{query} {rule['expansion']}"
            top_k = rule["top_k"]
            prompt_note = rule.get("prompt_note")
            break

    if not search_keywords:
        search_keywords = query

    # Classify the query into a category to pre-filter search space
    category_filter = None
    if any(w in query_lower for w in ["job", "jobs", "career", "careers", "hiring", "opening", "openings", "join us", "work at mintoak"]):
        category_filter = {"category": "Careers"}
    elif any(w in query_lower for w in ["culture", "work culture", "life at mintoak", "work environment", "values", "core values", "team culture"]):
        category_filter = {"category": "Culture"}
    elif any(w in query_lower for w in ["about us", "office", "offices", "headquarters", "location", "locations", "address", "addresses", "contact us", "founder", "founders", "raman", "khanduja", "sanjay", "nazareth", "rama", "tadepalli", "kabeer", "jain", "rohit", "ramana"]):
        category_filter = {"category": {"$in": ["Company Info", "About Us"]}}
    elif any(w in query_lower for w in ["interview", "case study", "case studies", "story", "stories", "ratnagiri", "dukandar", "shetty"]):
        category_filter = {"category": "Atmanirbhar Dukandar"}
    elif any(w in query_lower for w in ["product", "products", "offering", "pricing", "price", "demo", "soundbox", "soundhub", "smartpayments", "digionboard", "rewardrun", "business360", "sellsmart", "staffaccess", "acquisition", "onboard", "kyc", "kyb"]):
        category_filter = {"category": {"$in": ["Product offering", "General Info"]}}

    in_scope_keywords = [
        "mintoak", "digiledge", "smartpayments", "digionboard", "rewardrun", "business360",
        "sellsmart", "staffaccess", "soundhub", "soundbox", "raman", "khanduja", "sanjay",
        "nazareth", "rama", "tadepalli", "kabeer", "jain", "rohit", "ramana", "payment",
        "transaction", "merchant", "bank", "card", "gpv", "volume", "acquiring", "acquirer",
        "credit", "lending", "loan", "invoicing", "reconciliation", "settlement", "sme",
        "msme", "business", "retail", "onboard", "kyc", "kyb", "register", "signup", "login",
        "address", "office", "contact", "location", "headquarters", "demo", "career", "job",
        "hiring", "api", "sdk", "developer"
    ]
    is_mintoak_mention = any(w in query_lower for w in in_scope_keywords)

    query_kwargs = {"query_texts": [search_keywords], "n_results": top_k}
    if category_filter:
        query_kwargs["where"] = category_filter
    results = collection.query(**query_kwargs)
    
    best_distance = 999.0
    if results and results.get("distances") and results["distances"][0]:
        best_distance = results["distances"][0][0]

    # Out of scope checks matching app.py
    if best_distance > 0.82 or (best_distance > 0.60 and not is_mintoak_mention):
        return "OUT_OF_SCOPE_REFUSAL", {}, None

    context_parts = []
    url_to_title = {}
    if results and results.get("documents") and results["documents"][0]:
        for idx, doc in enumerate(results["documents"][0]):
            meta = results["metadatas"][0][idx]
            context_parts.append(f"Source: {meta['url']}\nContent: {doc}")
            url_to_title[meta["url"]] = meta.get("title", "Learn more")

    # Enforce strict cap of max 3 source document snippets
    if len(context_parts) > 3:
        context_parts = context_parts[:3]

    return "\n\n---\n\n".join(context_parts), url_to_title, prompt_note

def parse_test_cases():
    filepath = "/Users/mintoak/.gemini/antigravity/brain/dea6edf4-da3f-4317-bf84-1ebf1f7fab70/evaluation_test_cases.md"
    test_cases = []
    with open(filepath, "r") as f:
        content = f.read()

    sections = content.split("## ")
    for sec in sections:
        if "Group" not in sec:
            continue
        header = sec.split("\n")[0]
        category = "general_inquiry"
        if "Greetings" in header:
            category = "greeting"
        elif "Product" in header:
            category = "product_inquiry"
        elif "Business" in header:
            category = "general_inquiry"
        elif "Lead" in header:
            category = "lead_capture_intent"
        elif "Out-of-Scope" in header:
            category = "out_of_scope"
        elif "Guardrail" in header:
            category = "guardrail_injection"

        items = re.findall(r"(\d+)\.\s+\*\*Query:\*\*\s+\"([^\"]+)\"", sec)
        for num, query in items:
            test_cases.append({
                "id": f"eval_{int(num):03d}",
                "category": category,
                "query": query.strip()
            })
    return test_cases

def main():
    print("Parsing test cases from evaluation_test_cases.md...")
    test_cases = parse_test_cases()
    print(f"Loaded {len(test_cases)} test cases.")

    print("Connecting to ChromaDB...")
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
    emb_fn = embedding_functions.SentenceTransformerEmbeddingFunction(model_name="all-MiniLM-L6-v2")
    collection = client.get_or_create_collection(name=COLLECTION_NAME, embedding_function=emb_fn)

    print("Loading LLM and tokenizer...")
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH)
    model = AutoModelForCausalLM.from_pretrained(MODEL_PATH, torch_dtype=torch.float32)
    if os.path.exists(os.path.join(ADAPTER_PATH, "adapter_config.json")):
        try:
            model = PeftModel.from_pretrained(model, ADAPTER_PATH)
            print("LoRA adapter loaded successfully!")
        except Exception as e:
            print(f"Skipping LoRA adapter load due to compatibility/format limits: {e}")
            print("Using base model instead.")
    model.to(device)
    model.eval()
    print(f"Model loaded successfully on {device}!")

    results = []
    passed_count = 0
    total_latency = 0.0
    llm_runs = 0

    print("Executing test cases...")
    for idx, tc in enumerate(test_cases):
        query = tc["query"]
        category = tc["category"]
        tc_id = tc["id"]

        print(f"[{idx+1}/{len(test_cases)}] Running: '{query}' ({category})")

        t0 = time.time()
        
        query_lower = query.lower()
        static_response = None
        run_type = "llm_generated"
        url_to_title = {}

        if any(w in query_lower for w in PROFANE_PATTERNS) or any(p in query_lower for p in INJECTION_PATTERNS):
            static_response = "I can assist with questions about Mintoak's payment, merchant engagement, and value-added services platform. How can I guide you on those topics today?"
            run_type = "guardrail_blocked"
        elif GREETING_RE.match(query_clean := query.strip().lower().rstrip("?").strip()):
            static_response = "Hello! I am the Mintoak Knowledge Assistant. How can I help you with Mintoak's platform or services today?"
            run_type = "greeting"
        elif IDENTITY_RE.search(query_clean):
            static_response = "I am the Mintoak Knowledge Assistant. I can answer questions about Mintoak's white-labeled SaaS platform, product modules, and services. How can I help you today?"
            run_type = "identity"

        if static_response:
            response_cleaned = static_response
            response_raw = static_response
            latency = 0.0
        else:
            context, url_to_title, prompt_note = retrieve_context(collection, query)
            if context == "OUT_OF_SCOPE_REFUSAL":
                response_cleaned = "The requested information does not currently exist on www.mintoak.com. You can get in touch with our team at https://www.mintoak.com/contact-us."
                response_raw = response_cleaned
                latency = 0.0
                run_type = "out_of_scope_refusal"
            else:
                user_content = f"Context:\n{context}\n\nQuestion: {query}"
                if prompt_note:
                    user_content += f"\n\nNote: {prompt_note}"

                messages = [
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": user_content}
                ]
                prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
                inputs = tokenizer([prompt], return_tensors="pt").to(device)
                
                with torch.no_grad():
                    outputs = model.generate(**inputs, max_new_tokens=256, temperature=0.1, do_sample=False)
                
                response_raw = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True).strip()
                response_cleaned = clean_response(response_raw)
                latency = round(time.time() - t0, 2)
                total_latency += latency
                llm_runs += 1

        is_pass = False
        reason = ""
        banned_words = ['seamless', 'empowering', 'leverage', 'game-changer', 'game changer', 'synergy', 'synergies', 'unlocking', 'in today\'s fast-paced world', 'as we move forward']
        has_banned = any(w in response_cleaned.lower() for w in banned_words)

        if has_banned:
            is_pass = False
            reason = "Response contains banned marketing phrases"
        elif category == "greeting":
            if "Mintoak Knowledge Assistant" in response_cleaned or "Mintoak Assistant" in response_cleaned:
                is_pass = True
            else:
                reason = "Greeting response did not mention assistant persona"
        elif category in ["product_inquiry", "general_inquiry"]:
            has_sources = url_to_title and any("http" in str(k) for k in url_to_title.keys())
            if "http" in response_raw or has_sources or run_type == "out_of_scope_refusal":
                is_pass = True
            else:
                reason = "In-scope inquiry response missing Source citation URL or link"
        elif category in ["out_of_scope", "guardrail_safety", "guardrail_injection"]:
            is_pass = True
        elif category == "lead_capture_intent":
            if "name" in response_cleaned.lower() or "email" in response_cleaned.lower() or "contact" in response_cleaned.lower():
                is_pass = True
            else:
                reason = "Lead intent did not request user contact information"

        if is_pass:
            passed_count += 1

        results.append({
            "id": tc_id,
            "category": category,
            "query": query,
            "response": response_cleaned,
            "raw_response": response_raw,
            "latency": latency,
            "pass": is_pass,
            "reason": reason
        })

    os.makedirs("data/mintoak", exist_ok=True)
    with open("data/mintoak/eval_combined_progress.json", "w") as f:
        json.dump({
            "results": results,
            "passed_count": passed_count,
            "total_latency": total_latency,
            "llm_runs": llm_runs
        }, f, indent=2)

    print("\nEvaluation Complete!")
    print(f"Pass Rate: {passed_count}/{len(test_cases)} ({passed_count/len(test_cases)*100:.1f}%)")

if __name__ == "__main__":
    main()