BobAi
a96ddba3c9
Complete rebrand to eliminate any Claude/Anthropic references: Directory Changes: - claude_rag/ → mini_rag/ (preserving git history) Content Changes: - Replaced 930+ Claude references across 40+ files - Updated all imports: from claude_rag → from mini_rag - Updated all file paths: .claude-rag → .mini-rag - Updated documentation and comments - Updated configuration files and examples Testing Changes: - All tests updated to use mini_rag imports - Integration tests verify new module structure This ensures complete independence from Claude/Anthropic branding while maintaining all functionality and git history.
320 lines
12 KiB
Python
320 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
LLM Safeguards for Small Model Management
|
|
|
|
Provides runaway prevention, context management, and intelligent detection
|
|
of problematic model behaviors to ensure reliable user experience.
|
|
"""
|
|
|
|
import re
|
|
import time
|
|
import logging
|
|
from typing import Optional, Dict, List, Tuple
|
|
from dataclasses import dataclass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@dataclass
|
|
class SafeguardConfig:
|
|
"""Configuration for LLM safeguards."""
|
|
max_output_tokens: int = 2000 # Prevent excessive generation
|
|
max_repetition_ratio: float = 0.3 # Max ratio of repeated content
|
|
max_response_time: int = 60 # Max seconds for response
|
|
min_useful_length: int = 20 # Minimum useful response length
|
|
context_window: int = 32768 # Ollama context window
|
|
enable_thinking_detection: bool = True # Detect thinking patterns
|
|
|
|
class ModelRunawayDetector:
|
|
"""Detects and prevents model runaway behaviors."""
|
|
|
|
def __init__(self, config: SafeguardConfig = None):
|
|
self.config = config or SafeguardConfig()
|
|
self.response_patterns = self._compile_patterns()
|
|
|
|
def _compile_patterns(self) -> Dict[str, re.Pattern]:
|
|
"""Compile regex patterns for runaway detection."""
|
|
return {
|
|
# Excessive repetition patterns
|
|
'word_repetition': re.compile(r'\b(\w+)\b(?:\s+\1\b){3,}', re.IGNORECASE),
|
|
'phrase_repetition': re.compile(r'(.{10,50}?)\1{2,}', re.DOTALL),
|
|
|
|
# Thinking loop patterns (small models get stuck)
|
|
'thinking_loop': re.compile(r'(let me think|i think|thinking|consider|actually|wait|hmm|well)\s*[.,:]*\s*\1', re.IGNORECASE),
|
|
|
|
# Rambling patterns
|
|
'excessive_filler': re.compile(r'\b(um|uh|well|you know|like|basically|actually|so|then|and|but|however)\b(?:\s+[^.!?]*){5,}', re.IGNORECASE),
|
|
|
|
# JSON corruption patterns
|
|
'broken_json': re.compile(r'\{[^}]*\{[^}]*\{'), # Nested broken JSON
|
|
'json_repetition': re.compile(r'("[\w_]+"\s*:\s*"[^"]*",?\s*){4,}'), # Repeated JSON fields
|
|
}
|
|
|
|
def check_response_quality(self, response: str, query: str, start_time: float) -> Tuple[bool, Optional[str], Optional[str]]:
|
|
"""
|
|
Check response quality and detect runaway behaviors.
|
|
|
|
Returns:
|
|
(is_valid, issue_type, user_explanation)
|
|
"""
|
|
if not response or len(response.strip()) < self.config.min_useful_length:
|
|
return False, "too_short", self._explain_too_short()
|
|
|
|
# Check response time
|
|
elapsed = time.time() - start_time
|
|
if elapsed > self.config.max_response_time:
|
|
return False, "timeout", self._explain_timeout()
|
|
|
|
# Check for repetition issues
|
|
repetition_issue = self._check_repetition(response)
|
|
if repetition_issue:
|
|
return False, repetition_issue, self._explain_repetition(repetition_issue)
|
|
|
|
# Check for thinking loops
|
|
if self.config.enable_thinking_detection:
|
|
thinking_issue = self._check_thinking_loops(response)
|
|
if thinking_issue:
|
|
return False, thinking_issue, self._explain_thinking_loop()
|
|
|
|
# Check for rambling
|
|
rambling_issue = self._check_rambling(response)
|
|
if rambling_issue:
|
|
return False, rambling_issue, self._explain_rambling()
|
|
|
|
# Check JSON corruption (for structured responses)
|
|
if '{' in response and '}' in response:
|
|
json_issue = self._check_json_corruption(response)
|
|
if json_issue:
|
|
return False, json_issue, self._explain_json_corruption()
|
|
|
|
return True, None, None
|
|
|
|
def _check_repetition(self, response: str) -> Optional[str]:
|
|
"""Check for excessive repetition."""
|
|
# Word repetition
|
|
if self.response_patterns['word_repetition'].search(response):
|
|
return "word_repetition"
|
|
|
|
# Phrase repetition
|
|
if self.response_patterns['phrase_repetition'].search(response):
|
|
return "phrase_repetition"
|
|
|
|
# Calculate repetition ratio
|
|
words = response.split()
|
|
if len(words) > 10:
|
|
unique_words = set(words)
|
|
repetition_ratio = 1 - (len(unique_words) / len(words))
|
|
if repetition_ratio > self.config.max_repetition_ratio:
|
|
return "high_repetition_ratio"
|
|
|
|
return None
|
|
|
|
def _check_thinking_loops(self, response: str) -> Optional[str]:
|
|
"""Check for thinking loops (common in small models)."""
|
|
if self.response_patterns['thinking_loop'].search(response):
|
|
return "thinking_loop"
|
|
|
|
# Check for excessive meta-commentary
|
|
thinking_words = ['think', 'considering', 'actually', 'wait', 'hmm', 'let me']
|
|
thinking_count = sum(response.lower().count(word) for word in thinking_words)
|
|
|
|
if thinking_count > 5 and len(response.split()) < 200:
|
|
return "excessive_thinking"
|
|
|
|
return None
|
|
|
|
def _check_rambling(self, response: str) -> Optional[str]:
|
|
"""Check for rambling or excessive filler."""
|
|
if self.response_patterns['excessive_filler'].search(response):
|
|
return "excessive_filler"
|
|
|
|
# Check for extremely long sentences (sign of rambling)
|
|
sentences = re.split(r'[.!?]+', response)
|
|
long_sentences = [s for s in sentences if len(s.split()) > 50]
|
|
|
|
if len(long_sentences) > 2:
|
|
return "excessive_rambling"
|
|
|
|
return None
|
|
|
|
def _check_json_corruption(self, response: str) -> Optional[str]:
|
|
"""Check for JSON corruption in structured responses."""
|
|
if self.response_patterns['broken_json'].search(response):
|
|
return "broken_json"
|
|
|
|
if self.response_patterns['json_repetition'].search(response):
|
|
return "json_repetition"
|
|
|
|
return None
|
|
|
|
def _explain_too_short(self) -> str:
|
|
return """🤔 The AI response was too short to be helpful.
|
|
|
|
**Why this happens:**
|
|
• The model might be confused by the query
|
|
• Context might be insufficient
|
|
• Model might be overloaded
|
|
|
|
**What to try:**
|
|
• Rephrase your question more specifically
|
|
• Try a broader search term first
|
|
• Use exploration mode for complex questions: `rag-mini explore`"""
|
|
|
|
def _explain_timeout(self) -> str:
|
|
return """⏱️ The AI took too long to respond (over 60 seconds).
|
|
|
|
**Why this happens:**
|
|
• Small models sometimes get "stuck" thinking
|
|
• Complex queries can overwhelm smaller models
|
|
• System might be under load
|
|
|
|
**What to try:**
|
|
• Try a simpler, more direct question
|
|
• Use synthesis mode for faster responses: `--synthesize`
|
|
• Consider using a larger model if available"""
|
|
|
|
def _explain_repetition(self, issue_type: str) -> str:
|
|
return f"""🔄 The AI got stuck in repetition loops ({issue_type}).
|
|
|
|
**Why this happens:**
|
|
• Small models sometimes repeat when uncertain
|
|
• Query might be too complex for the model size
|
|
• Context window might be exceeded
|
|
|
|
**What to try:**
|
|
• Try a more specific question
|
|
• Break complex questions into smaller parts
|
|
• Use exploration mode which handles context better: `rag-mini explore`
|
|
• Consider: A larger model (qwen3:1.7b or qwen3:3b) would help"""
|
|
|
|
def _explain_thinking_loop(self) -> str:
|
|
return """🧠 The AI got caught in a "thinking loop" - overthinking the response.
|
|
|
|
**Why this happens:**
|
|
• Small models sometimes over-analyze simple questions
|
|
• Thinking mode can cause loops in smaller models
|
|
• Query complexity exceeds model capabilities
|
|
|
|
**What to try:**
|
|
• Ask more direct, specific questions
|
|
• Use synthesis mode (no thinking) for faster results
|
|
• Try: "What does this code do?" instead of "Explain how this works"
|
|
• Larger models (qwen3:1.7b+) handle thinking better"""
|
|
|
|
def _explain_rambling(self) -> str:
|
|
return """💭 The AI started rambling instead of giving focused answers.
|
|
|
|
**Why this happens:**
|
|
• Small models sometimes lose focus on complex topics
|
|
• Query might be too broad or vague
|
|
• Model trying to cover too much at once
|
|
|
|
**What to try:**
|
|
• Ask more specific questions
|
|
• Break broad questions into focused parts
|
|
• Example: "How is data validated?" instead of "Explain the whole system"
|
|
• Exploration mode helps maintain focus across questions"""
|
|
|
|
def _explain_json_corruption(self) -> str:
|
|
return """🔧 The AI response format got corrupted.
|
|
|
|
**Why this happens:**
|
|
• Small models sometimes struggle with structured output
|
|
• Context limits can cause format errors
|
|
• Complex analysis might overwhelm formatting
|
|
|
|
**What to try:**
|
|
• Try the question again (often resolves itself)
|
|
• Use simpler questions for better formatting
|
|
• Synthesis mode sometimes gives cleaner output
|
|
• This is less common with larger models"""
|
|
|
|
def get_recovery_suggestions(self, issue_type: str, query: str) -> List[str]:
|
|
"""Get specific recovery suggestions based on the issue."""
|
|
suggestions = []
|
|
|
|
if issue_type in ['thinking_loop', 'excessive_thinking']:
|
|
suggestions.extend([
|
|
f"Try synthesis mode: `rag-mini search . \"{query}\" --synthesize`",
|
|
"Ask more direct questions without 'why' or 'how'",
|
|
"Break complex questions into smaller parts"
|
|
])
|
|
|
|
elif issue_type in ['word_repetition', 'phrase_repetition', 'high_repetition_ratio']:
|
|
suggestions.extend([
|
|
"Try rephrasing your question completely",
|
|
"Use more specific technical terms",
|
|
f"Try exploration mode: `rag-mini explore .`"
|
|
])
|
|
|
|
elif issue_type == 'timeout':
|
|
suggestions.extend([
|
|
"Try a simpler version of your question",
|
|
"Use synthesis mode for faster responses",
|
|
"Check if Ollama is under heavy load"
|
|
])
|
|
|
|
# Universal suggestions
|
|
suggestions.extend([
|
|
"Consider using a larger model if available (qwen3:1.7b or qwen3:3b)",
|
|
"Check model status: `ollama list`"
|
|
])
|
|
|
|
return suggestions
|
|
|
|
def get_optimal_ollama_parameters(model_name: str) -> Dict[str, any]:
|
|
"""Get optimal parameters for different Ollama models."""
|
|
|
|
base_params = {
|
|
"num_ctx": 32768, # Good context window for most uses
|
|
"num_predict": 2000, # Reasonable response length
|
|
"temperature": 0.3, # Balanced creativity/consistency
|
|
}
|
|
|
|
# Model-specific optimizations
|
|
if "qwen3:0.6b" in model_name.lower():
|
|
return {
|
|
**base_params,
|
|
"repeat_penalty": 1.15, # Prevent repetition in small model
|
|
"presence_penalty": 1.5, # Suppress repetitive outputs
|
|
"top_p": 0.8, # Focused sampling
|
|
"top_k": 20, # Limit choices
|
|
"num_predict": 1500, # Shorter responses for reliability
|
|
}
|
|
|
|
elif "qwen3:1.7b" in model_name.lower():
|
|
return {
|
|
**base_params,
|
|
"repeat_penalty": 1.1, # Less aggressive for larger model
|
|
"presence_penalty": 1.0, # Balanced
|
|
"top_p": 0.9, # More creative
|
|
"top_k": 40, # More choices
|
|
}
|
|
|
|
elif any(size in model_name.lower() for size in ["3b", "7b", "8b"]):
|
|
return {
|
|
**base_params,
|
|
"repeat_penalty": 1.05, # Minimal for larger models
|
|
"presence_penalty": 0.5, # Light touch
|
|
"top_p": 0.95, # High creativity
|
|
"top_k": 50, # Many choices
|
|
"num_predict": 3000, # Longer responses OK
|
|
}
|
|
|
|
return base_params
|
|
|
|
# Quick test
|
|
def test_safeguards():
|
|
"""Test the safeguard system."""
|
|
detector = ModelRunawayDetector()
|
|
|
|
# Test repetition detection
|
|
bad_response = "The user authentication system works by checking user credentials. The user authentication system works by checking user credentials. The user authentication system works by checking user credentials."
|
|
|
|
is_valid, issue, explanation = detector.check_response_quality(bad_response, "auth", time.time())
|
|
|
|
print(f"Repetition test: Valid={is_valid}, Issue={issue}")
|
|
if explanation:
|
|
print(explanation)
|
|
|
|
if __name__ == "__main__":
|
|
test_safeguards() |