Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-guard/src/lexigram/ai/guard/input/topic.py: 44%

1"""Topic restrictor input guard. 

2 

3Blocks or warns on inputs that touch topics declared off-limits for 

4a deployment. Matching uses exact keyword/phrase lookup and optional 

5word-boundary enforcement — no external API calls. 

6 

7For more accurate topic classification use an LLM-based classifier as 

8a second guard in the pipeline after this fast pre-filter. 

9""" 

10 

11from __future__ import annotations 

12 

13import re 

14from typing import TYPE_CHECKING, Any 

15 

16from lexigram.ai.guard.input.base import AbstractInputGuard 

17from lexigram.ai.guard.pipeline.result import GuardCheckResult 

18from lexigram.contracts.ai.guards import GuardResultProtocol 

19from lexigram.result import Ok, Result 

20 

21if TYPE_CHECKING: 

22 from lexigram.contracts.ai.exceptions import GuardError 

23 

24 

25def _build_pattern(topic: str, whole_word: bool) -> re.Pattern[str]: 

26 """Compile a regex pattern for a topic phrase. 

27 

28 Args: 

29 topic: The topic phrase to match. 

30 whole_word: Whether to require word boundaries around the match. 

31 

32 Returns: 

33 Compiled regex pattern. 

34 """ 

35 escaped = re.escape(topic) 

36 if whole_word: 

37 return re.compile(r"\b" + escaped + r"\b", re.I) 

38 return re.compile(escaped, re.I) 

39 

40 

41class TopicRestrictor(AbstractInputGuard): 

42 """Input guard that blocks prohibited topics. 

43 

44 Args: 

45 restricted_topics: List of topic keywords/phrases to prohibit. 

46 action: Action when a restricted topic is found — 

47 ``"block"`` (default) or ``"warn"``. 

48 whole_word: If ``True`` (default), only match whole words / 

49 phrase boundaries to reduce false positives. 

50 

51 Example:: 

52 

53 guard = TopicRestrictor( 

54 restricted_topics=["weapons", "explosives", "self-harm"], 

55 action="block", 

56 ) 

57 result = await guard.check(user_message) 

58 """ 

59 

60 def __init__( 

61 self, 

62 restricted_topics: list[str], 

63 action: str = "block", 

64 whole_word: bool = True, 

65 ) -> None: 

66 """Initialise the topic restrictor. 

67 

68 Args: 

69 restricted_topics: List of prohibited topic keywords/phrases. 

70 action: ``"block"`` or ``"warn"``. 

71 whole_word: Enforce word boundaries on matches. 

72 """ 

73 super().__init__(action=action) 

74 self._patterns: list[tuple[str, re.Pattern[str]]] = [ 

75 (topic, _build_pattern(topic, whole_word)) for topic in restricted_topics 

76 ] 

77 

78 async def check( 

79 self, 

80 content: str, 

81 *, 

82 messages: list[Any] | None = None, 

83 metadata: dict[str, Any] | None = None, 

84 ) -> Result[GuardResultProtocol, GuardError]: 

85 """Check content against restricted topics. 

86 

87 Args: 

88 content: Input text to evaluate. 

89 messages: Unused — present for protocol compatibility. 

90 metadata: Optional metadata. 

91 

92 Returns: 

93 PASS if no restricted topics found; BLOCK or WARN otherwise. 

94 """ 

95 matched_topics: list[str] = [] 

96 for topic, pattern in self._patterns: 

97 if pattern.search(content): 

98 matched_topics.append(topic) 

99 

100 if not matched_topics: 

101 return Ok(GuardCheckResult.allow(self.name)) 

102 

103 if self._action == "warn": 

104 return Ok( 

105 GuardCheckResult.warn( 

106 self.name, 

107 reason=f"Restricted topics mentioned: {', '.join(matched_topics)}", 

108 matched_topics=matched_topics, 

109 ) 

110 ) 

111 

112 return Ok( 

113 GuardCheckResult.block( 

114 self.name, 

115 reason=f"Input mentions restricted topics: {', '.join(matched_topics)}", 

116 matched_topics=matched_topics, 

117 ) 

118 ) 

119 

120 

121__all__ = ["TopicRestrictor"]