Spaces:
Running
Running
File size: 1,692 Bytes
cebfa40 eb31fdf cebfa40 0fdec96 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | import re
import logging
from typing import List
logger = logging.getLogger("axiom.query")
class QueryProcessor:
"""
Cleans, validates, and expands the user query
before it hits the retrieval pipeline.
"""
BANNED_PATTERNS = [
r"ignore previous instructions",
r"you are now",
r"forget everything",
r"act as",
]
def __init__(self):
logger.info("Ready.")
def clean(self, query: str) -> str:
"""Basic cleaning — strip, normalize whitespace."""
query = query.strip()
query = re.sub(r"\s+", " ", query)
return query
def is_safe(self, query: str) -> bool:
"""Guardrail — reject prompt injection attempts."""
lowered = query.lower()
for pattern in self.BANNED_PATTERNS:
if re.search(pattern, lowered):
return False
return True
def expand(self, query: str) -> str:
"""
Simple query expansion:
appends 'explain', 'definition', 'overview'
to help retrieval find introductory chunks.
"""
expansion_suffix = "definition overview explanation"
return f"{query} {expansion_suffix}"
def process(self, raw_query: str) -> dict:
"""
Full query processing pipeline.
Returns dict with cleaned query, expanded query, and safety flag.
"""
cleaned = self.clean(raw_query)
safe = self.is_safe(cleaned)
expanded = self.expand(cleaned) if safe else ""
return {
"original": raw_query,
"cleaned": cleaned,
"expanded": expanded,
"is_safe": safe
} |