INTRODUCTION TO PROMPT OPTIMIZATION
The quality of responses from large language models depends critically on the quality of the prompts they receive. A well-crafted prompt can mean the difference between a vague, unhelpful response and a precise, actionable answer. However, most users struggle to formulate effective prompts. They often omit crucial context, use ambiguous language, or fail to specify the desired output format. This tutorial presents a comprehensive system called Critique that addresses these challenges by analyzing user prompts, identifying weaknesses, gathering missing information, and reconstructing optimized prompts that follow established best practices.
The Critique system operates as an intelligent intermediary between users and language models. When a user submits a prompt, Critique examines it across multiple dimensions including clarity, completeness, specificity, and potential for bias or hallucination. It engages in a dialogue with the user to clarify ambiguities and gather missing context. Finally, it synthesizes an improved prompt or a sequence of prompts that maximize the likelihood of obtaining high-quality responses.
This system supports both local and remote language models, accommodating diverse hardware configurations including Intel GPUs, AMD GPUs with ROCm, Apple Silicon with Metal Performance Shaders, and Nvidia GPUs with CUDA. This flexibility ensures that users can leverage whatever computational resources they have available.
ARCHITECTURAL OVERVIEW
The Critique system comprises several interconnected components that work together to analyze and optimize prompts. At the highest level, the architecture consists of a prompt analyzer, a dialogue manager, a prompt reconstructor, and an LLM interface layer that abstracts away the differences between various model backends.
The prompt analyzer examines incoming prompts using a combination of rule-based heuristics and LLM-powered semantic analysis. It identifies issues such as vague terminology, missing context, ambiguous instructions, and potential sources of bias. The analyzer produces a structured assessment that categorizes problems by severity and type.
The dialogue manager orchestrates the conversation with the user to gather missing information. It generates targeted questions based on the analyzer's findings and maintains conversation state to ensure coherent multi-turn interactions. The dialogue manager knows when to ask for clarification versus when to make reasonable assumptions.
The prompt reconstructor takes the original prompt along with all gathered information and synthesizes an optimized version. It applies best practices such as providing clear role definitions, specifying output formats, including relevant examples, and breaking complex requests into manageable sub-tasks. When appropriate, it splits a single complex prompt into a sequence of simpler prompts that build upon each other.
The LLM interface layer provides a unified API for interacting with different language model backends. It handles model loading, inference, and resource management across various hardware platforms. This abstraction allows the rest of the system to remain agnostic to the underlying model implementation.
HARDWARE ACCELERATION SUPPORT
Supporting multiple GPU architectures requires careful abstraction of the acceleration layer. Different vendors provide different APIs and runtime environments. Nvidia uses CUDA, AMD uses ROCm which exposes a CUDA-compatible API, Intel uses oneAPI with XPU device support, and Apple uses Metal Performance Shaders accessible through the MPS backend. The system must detect available hardware and select the appropriate backend.
The hardware detection module queries the system for available accelerators and their capabilities. It checks for CUDA-capable devices, ROCm installations, Intel GPU drivers, and Apple Silicon. Based on what it finds, it configures the model loading parameters appropriately. The detection process is robust and handles cases where drivers are installed but not properly configured.
Here is a code snippet showing the hardware detection logic:
import torch
import platform
import sys
class HardwareDetector:
def __init__(self):
self.available_backends = []
self.preferred_backend = None
self.device_info = {}
self._detect_hardware()
def _detect_hardware(self):
# Check for CUDA (Nvidia) and ROCm (AMD)
if torch.cuda.is_available():
# Could be CUDA or ROCm since ROCm uses CUDA API
try:
device_count = torch.cuda.device_count()
device_name = torch.cuda.get_device_name(0)
# Check if this is ROCm
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
self.available_backends.append('rocm')
self.device_info['rocm'] = {
'count': device_count,
'name': device_name,
'version': torch.version.hip
}
print(f"Detected {device_count} ROCm device(s): {device_name}")
print(f" ROCm version: {torch.version.hip}")
else:
self.available_backends.append('cuda')
self.device_info['cuda'] = {
'count': device_count,
'name': device_name,
'compute_capability': torch.cuda.get_device_capability(0)
}
print(f"Detected {device_count} CUDA device(s): {device_name}")
print(f" Compute capability: {torch.cuda.get_device_capability(0)}")
except Exception as e:
print(f"CUDA/ROCm detection error: {e}")
# Check for MPS (Apple Silicon)
if platform.system() == 'Darwin':
try:
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.available_backends.append('mps')
self.device_info['mps'] = {'available': True}
print("Detected Apple MPS (Metal Performance Shaders)")
except Exception as e:
print(f"MPS detection error: {e}")
# Check for Intel GPU support
try:
import intel_extension_for_pytorch as ipex
# Verify XPU is actually available
if hasattr(torch, 'xpu') and torch.xpu.is_available():
self.available_backends.append('intel')
device_count = torch.xpu.device_count()
self.device_info['intel'] = {'count': device_count}
print(f"Detected {device_count} Intel XPU device(s)")
except ImportError:
pass
except Exception as e:
print(f"Intel GPU detection error: {e}")
# Fallback to CPU
if not self.available_backends:
self.available_backends.append('cpu')
self.device_info['cpu'] = {'cores': 'available'}
print("No GPU acceleration detected, using CPU")
self.preferred_backend = self.available_backends[0]
print(f"Selected backend: {self.preferred_backend.upper()}\n")
def get_device(self):
if self.preferred_backend == 'cuda':
return torch.device('cuda:0')
elif self.preferred_backend == 'rocm':
return torch.device('cuda:0') # ROCm uses CUDA API
elif self.preferred_backend == 'mps':
return torch.device('mps')
elif self.preferred_backend == 'intel':
return torch.device('xpu:0')
else:
return torch.device('cpu')
def supports_quantization(self):
# Quantization support varies by backend
return self.preferred_backend in ['cuda', 'rocm']
def get_backend_name(self):
return self.preferred_backend
This hardware detector examines the system environment and determines which acceleration backend to use. It prioritizes GPU acceleration when available but gracefully falls back to CPU execution. The detector returns a PyTorch device object that the model loading code uses to place tensors on the appropriate hardware. The supports_quantization method indicates whether the current backend can use quantization libraries like bitsandbytes, which currently only support CUDA and ROCm.
LLM INTERFACE ABSTRACTION
The LLM interface layer provides a consistent API regardless of whether the model runs locally or remotely. For local models, it uses the transformers library from Hugging Face, which supports a wide variety of open-source models. For remote models, it provides connectors for popular API services like OpenAI and Anthropic.
The interface defines a common set of operations including model initialization, text generation, and resource cleanup. Each backend implements these operations according to its specific requirements. This design pattern allows the higher-level components to work with any model without modification.
Here is the base interface definition:
from abc import ABC, abstractmethod
from typing import List, Dict, Optional
class LLMInterface(ABC):
def __init__(self, model_name: str, config: Dict):
self.model_name = model_name
self.config = config
self.initialized = False
@abstractmethod
def initialize(self):
"""Load and prepare the model for inference"""
pass
@abstractmethod
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
"""Generate text from a prompt"""
pass
@abstractmethod
def cleanup(self):
"""Release resources and clean up"""
pass
def validate_temperature(self, temperature: float) -> float:
"""Ensure temperature is in valid range"""
if temperature < 0.0:
return 0.0
elif temperature > 2.0:
return 2.0
return temperature
def __enter__(self):
self.initialize()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.cleanup()
This abstract base class defines the contract that all LLM implementations must fulfill. The context manager protocol ensures proper resource management even when exceptions occur. The validate_temperature method ensures that temperature values stay within acceptable bounds. Concrete implementations override the abstract methods to provide backend-specific functionality.
LOCAL MODEL IMPLEMENTATION
The local model implementation uses the transformers library to load and run models directly on the user's hardware. It handles model quantization for memory efficiency when supported, configures the appropriate device placement based on hardware detection, and manages the generation parameters. The implementation includes robust error handling and validation to ensure reliable operation across different hardware configurations.
class LocalLLM(LLMInterface):
def __init__(self, model_name: str, config: Dict, hardware_detector: HardwareDetector):
super().__init__(model_name, config)
self.hardware_detector = hardware_detector
self.model = None
self.tokenizer = None
self.device = None
def initialize(self):
if self.initialized:
return
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
except ImportError:
raise RuntimeError("transformers library not installed. Install with: pip install transformers")
self.device = self.hardware_detector.get_device()
print(f"Loading model {self.model_name} on {self.device}")
# Configure quantization for memory efficiency if supported
quantization_config = None
use_quantization = self.config.get('use_quantization', True)
if use_quantization and self.hardware_detector.supports_quantization():
try:
from transformers import BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4"
)
print(" Using 4-bit quantization for memory efficiency")
except ImportError:
print(" bitsandbytes not available, loading without quantization")
quantization_config = None
except Exception as e:
print(f" Quantization setup failed: {e}, loading without quantization")
quantization_config = None
# Load tokenizer
try:
self.tokenizer = AutoTokenizer.from_pretrained(
self.model_name,
trust_remote_code=self.config.get('trust_remote_code', False)
)
# Set pad token if not present
if self.tokenizer.pad_token is None:
if self.tokenizer.eos_token is not None:
self.tokenizer.pad_token = self.tokenizer.eos_token
else:
self.tokenizer.add_special_tokens({'pad_token': '[PAD]'})
except Exception as e:
raise RuntimeError(f"Failed to load tokenizer: {e}")
# Load model with appropriate configuration
model_kwargs = {
'trust_remote_code': self.config.get('trust_remote_code', False),
'low_cpu_mem_usage': True,
}
# Set dtype based on device
if self.device.type == 'cpu':
model_kwargs['torch_dtype'] = torch.float32
elif self.device.type == 'mps':
model_kwargs['torch_dtype'] = torch.float16
else:
model_kwargs['torch_dtype'] = torch.float16
if quantization_config:
model_kwargs['quantization_config'] = quantization_config
model_kwargs['device_map'] = 'auto'
elif self.device.type in ['cuda', 'xpu']:
model_kwargs['device_map'] = 'auto'
try:
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
**model_kwargs
)
# Move to device if not using device_map
if 'device_map' not in model_kwargs or model_kwargs['device_map'] is None:
self.model = self.model.to(self.device)
self.model.eval()
except Exception as e:
raise RuntimeError(f"Failed to load model: {e}")
self.initialized = True
print("Model loaded successfully\n")
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
if not self.initialized:
raise RuntimeError("Model not initialized. Call initialize() first.")
# Validate temperature
temperature = self.validate_temperature(temperature)
# Tokenize input
inputs = self.tokenizer(
prompt,
return_tensors="pt",
padding=True,
truncation=True,
max_length=self.config.get('max_input_length', 2048)
)
inputs = {k: v.to(self.device) for k, v in inputs.items()}
# Set generation parameters
gen_kwargs = {
'max_new_tokens': max_tokens,
'temperature': temperature,
'do_sample': temperature > 0.0,
'pad_token_id': self.tokenizer.pad_token_id,
'eos_token_id': self.tokenizer.eos_token_id,
}
# Add top_p for better sampling when temperature > 0
if temperature > 0.0:
gen_kwargs['top_p'] = kwargs.pop('top_p', 0.9)
gen_kwargs.update(kwargs)
# Generate response
try:
with torch.no_grad():
outputs = self.model.generate(**inputs, **gen_kwargs)
# Decode output
full_response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
# Remove prompt from response more robustly
prompt_stripped = prompt.strip()
response_stripped = full_response.strip()
if response_stripped.startswith(prompt_stripped):
response = response_stripped[len(prompt_stripped):].strip()
else:
# Try to find where the actual response starts
response = full_response
return response
except Exception as e:
raise RuntimeError(f"Generation failed: {e}")
def cleanup(self):
if self.model is not None:
del self.model
self.model = None
if self.tokenizer is not None:
del self.tokenizer
self.tokenizer = None
# Clear GPU cache based on backend
if self.device.type == 'cuda':
torch.cuda.empty_cache()
elif self.device.type == 'xpu':
if hasattr(torch.xpu, 'empty_cache'):
torch.xpu.empty_cache()
elif self.device.type == 'mps':
if hasattr(torch.mps, 'empty_cache'):
torch.mps.empty_cache()
self.initialized = False
print("Model resources cleaned up")
This local model implementation handles the complexities of loading large language models efficiently. It uses 4-bit quantization when running on supported devices to reduce memory consumption. The generate method tokenizes the input, runs inference, and decodes the output while handling device placement transparently. The cleanup method properly releases resources and clears caches specific to each backend.
REMOTE MODEL IMPLEMENTATION
The remote model implementation provides connectivity to cloud-based language model APIs. It handles authentication, request formatting, rate limiting, and error recovery. Different API providers have different interfaces, so the implementation includes adapters for each supported service. The implementation uses exponential backoff for retries to handle transient network errors gracefully.
import requests
import time
from typing import Optional
class RemoteLLM(LLMInterface):
def __init__(self, model_name: str, config: Dict):
super().__init__(model_name, config)
self.api_key = config.get('api_key')
self.api_base = config.get('api_base')
self.provider = config.get('provider', 'openai')
self.session = None
# Set default API base URLs
if not self.api_base:
if self.provider == 'openai':
self.api_base = 'https://api.openai.com/v1'
elif self.provider == 'anthropic':
self.api_base = 'https://api.anthropic.com/v1'
def initialize(self):
if self.initialized:
return
if not self.api_key:
raise ValueError("API key required for remote model")
self.session = requests.Session()
if self.provider == 'openai':
self.session.headers.update({
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
})
elif self.provider == 'anthropic':
self.session.headers.update({
'x-api-key': self.api_key,
'Content-Type': 'application/json',
'anthropic-version': '2023-06-01'
})
else:
raise ValueError(f"Unsupported provider: {self.provider}")
self.initialized = True
print(f"Remote model interface initialized for {self.provider}\n")
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
if not self.initialized:
raise RuntimeError("Model not initialized. Call initialize() first.")
# Validate temperature
temperature = self.validate_temperature(temperature)
if self.provider == 'openai':
return self._generate_openai(prompt, max_tokens, temperature, **kwargs)
elif self.provider == 'anthropic':
return self._generate_anthropic(prompt, max_tokens, temperature, **kwargs)
else:
raise ValueError(f"Unsupported provider: {self.provider}")
def _generate_openai(self, prompt: str, max_tokens: int,
temperature: float, **kwargs) -> str:
url = f"{self.api_base}/chat/completions"
payload = {
'model': self.model_name,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
# Add any additional parameters
for key in ['top_p', 'frequency_penalty', 'presence_penalty']:
if key in kwargs:
payload[key] = kwargs[key]
max_retries = 3
for attempt in range(max_retries):
try:
response = self.session.post(url, json=payload, timeout=120)
response.raise_for_status()
data = response.json()
if 'choices' in data and len(data['choices']) > 0:
return data['choices'][0]['message']['content']
else:
raise RuntimeError("Unexpected API response format")
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request timeout, retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise RuntimeError(f"Request timed out after {max_retries} attempts")
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request failed, retrying in {wait_time}s: {e}")
time.sleep(wait_time)
else:
raise RuntimeError(f"Failed to generate response after {max_retries} attempts: {e}")
def _generate_anthropic(self, prompt: str, max_tokens: int,
temperature: float, **kwargs) -> str:
url = f"{self.api_base}/messages"
payload = {
'model': self.model_name,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
# Add any additional parameters
for key in ['top_p', 'top_k']:
if key in kwargs:
payload[key] = kwargs[key]
max_retries = 3
for attempt in range(max_retries):
try:
response = self.session.post(url, json=payload, timeout=120)
response.raise_for_status()
data = response.json()
if 'content' in data and len(data['content']) > 0:
return data['content'][0]['text']
else:
raise RuntimeError("Unexpected API response format")
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request timeout, retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise RuntimeError(f"Request timed out after {max_retries} attempts")
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request failed, retrying in {wait_time}s: {e}")
time.sleep(wait_time)
else:
raise RuntimeError(f"Failed to generate response after {max_retries} attempts: {e}")
def cleanup(self):
if self.session:
self.session.close()
self.session = None
self.initialized = False
The remote model implementation abstracts the differences between API providers. It includes retry logic with exponential backoff to handle transient network errors gracefully. The provider-specific methods format requests according to each API's requirements and validate the response structure before extracting the generated text.
PROMPT ANALYSIS FRAMEWORK
The prompt analyzer examines user prompts across multiple dimensions to identify potential issues. It checks for clarity, completeness, specificity, context, output format specification, and potential sources of bias or hallucination. The analyzer combines rule-based heuristics with LLM-powered semantic analysis to produce a comprehensive assessment.
The analysis process begins with structural checks that examine the prompt's length, sentence structure, and presence of key elements. It looks for question marks, imperative verbs, and explicit instructions. It identifies vague terms like "good," "better," or "some" that lack precise meaning.
The semantic analysis uses the language model itself to understand the prompt's intent and identify missing context. It generates questions that would help clarify the user's goals and constraints. This meta-analysis approach leverages the model's language understanding capabilities to go beyond simple pattern matching.
import re
from typing import List, Dict, Tuple
from dataclasses import dataclass
import json
@dataclass
class AnalysisIssue:
category: str
severity: str # 'high', 'medium', 'low'
description: str
suggestion: str
location: Optional[str] = None
class PromptAnalyzer:
def __init__(self, llm_interface: LLMInterface):
self.llm = llm_interface
self.vague_terms = [
'good', 'better', 'best', 'nice', 'some', 'many', 'few',
'often', 'sometimes', 'usually', 'appropriate', 'suitable',
'relevant', 'important', 'significant', 'various', 'several'
]
def analyze(self, prompt: str) -> List[AnalysisIssue]:
issues = []
# Structural analysis
issues.extend(self._check_length(prompt))
issues.extend(self._check_clarity(prompt))
issues.extend(self._check_specificity(prompt))
issues.extend(self._check_output_format(prompt))
issues.extend(self._check_context(prompt))
# Semantic analysis using LLM
semantic_issues = self._semantic_analysis(prompt)
if semantic_issues:
issues.extend(semantic_issues)
return issues
def _check_length(self, prompt: str) -> List[AnalysisIssue]:
issues = []
word_count = len(prompt.split())
if word_count < 5:
issues.append(AnalysisIssue(
category='completeness',
severity='high',
description='Prompt is very short and likely lacks necessary detail',
suggestion='Provide more context about what you want to achieve'
))
elif word_count > 500:
issues.append(AnalysisIssue(
category='clarity',
severity='medium',
description='Prompt is very long and may contain unnecessary information',
suggestion='Consider breaking this into multiple focused prompts'
))
return issues
def _check_clarity(self, prompt: str) -> List[AnalysisIssue]:
issues = []
# Check for multiple questions
question_marks = prompt.count('?')
if question_marks > 3:
issues.append(AnalysisIssue(
category='clarity',
severity='medium',
description='Prompt contains multiple questions',
suggestion='Focus on one main question or clearly separate distinct requests'
))
# Check for ambiguous pronouns
sentences = re.split(r'[.!?]+', prompt)
for sentence in sentences:
if len(sentence.strip()) < 5:
continue
pronouns = re.findall(r'\b(it|this|that|they|them)\b', sentence.lower())
if len(pronouns) > 2:
issues.append(AnalysisIssue(
category='clarity',
severity='low',
description='Sentence contains ambiguous pronouns',
suggestion='Replace pronouns with specific nouns for clarity',
location=sentence.strip()[:100]
))
return issues
def _check_specificity(self, prompt: str) -> List[AnalysisIssue]:
issues = []
# Check for vague terms
prompt_lower = prompt.lower()
found_vague_terms = []
for term in self.vague_terms:
# Use word boundaries to avoid false matches
pattern = r'\b' + re.escape(term) + r'\b'
if re.search(pattern, prompt_lower):
found_vague_terms.append(term)
if found_vague_terms:
issues.append(AnalysisIssue(
category='specificity',
severity='medium',
description=f'Prompt contains vague terms: {", ".join(found_vague_terms[:5])}',
suggestion='Replace vague terms with specific quantities, criteria, or examples'
))
# Check for missing constraints
has_constraints = any(word in prompt_lower for word in
['must', 'should', 'require', 'limit', 'maximum', 'minimum', 'exactly'])
if not has_constraints and len(prompt.split()) > 20:
issues.append(AnalysisIssue(
category='specificity',
severity='low',
description='No explicit constraints or requirements specified',
suggestion='Consider adding specific requirements or constraints'
))
return issues
def _check_output_format(self, prompt: str) -> List[AnalysisIssue]:
issues = []
format_keywords = ['format', 'structure', 'json', 'list', 'table', 'bullet', 'numbered']
has_format_spec = any(keyword in prompt.lower() for keyword in format_keywords)
if not has_format_spec and len(prompt.split()) > 30:
issues.append(AnalysisIssue(
category='output_format',
severity='low',
description='No output format specified',
suggestion='Specify the desired format for the response'
))
return issues
def _check_context(self, prompt: str) -> List[AnalysisIssue]:
issues = []
# Check for context indicators
context_indicators = ['because', 'since', 'for', 'background', 'context', 'purpose']
has_context = any(indicator in prompt.lower() for indicator in context_indicators)
if not has_context and len(prompt.split()) < 15:
issues.append(AnalysisIssue(
category='context',
severity='medium',
description='Limited context provided',
suggestion='Add background information about why you need this and how you will use it'
))
return issues
def _semantic_analysis(self, prompt: str) -> List[AnalysisIssue]:
issues = []
analysis_prompt = f"""Analyze the following user prompt for potential issues:
Prompt: "{prompt}"
Identify any of the following problems:
- Ambiguous instructions that could be interpreted multiple ways
- Missing critical information needed to provide a complete answer
- Potential for biased or unfair responses
- Risk of hallucination due to requesting information that may not exist
- Conflicting requirements or contradictions
For each issue found, provide a JSON object with these exact keys:
- category: one of (ambiguity, missing_info, bias_risk, hallucination_risk, contradiction)
- severity: one of (high, medium, low)
- description: brief explanation
- suggestion: specific recommendation
Return ONLY a JSON array of issue objects. If no issues found, return []. Example: [{{"category": "ambiguity", "severity": "medium", "description": "...", "suggestion": "..."}}] """
try:
response = self.llm.generate(analysis_prompt, temperature=0.3, max_tokens=800)
# Try to parse JSON from response
json_match = re.search(r'\[\s*\{.*\}\s*\]', response, re.DOTALL)
if json_match:
try:
semantic_issues = json.loads(json_match.group())
for issue_data in semantic_issues:
# Validate required fields
if all(key in issue_data for key in ['category', 'severity', 'description', 'suggestion']):
issues.append(AnalysisIssue(
category=issue_data['category'],
severity=issue_data['severity'],
description=issue_data['description'],
suggestion=issue_data['suggestion']
))
except json.JSONDecodeError as e:
print(f" Warning: Failed to parse semantic analysis JSON: {e}")
except Exception as e:
print(f" Warning: Semantic analysis failed: {e}")
# Continue with rule-based analysis only
return issues
The prompt analyzer combines multiple analysis strategies to provide comprehensive feedback. The structural checks use regular expressions and simple heuristics to identify common problems. The semantic analysis leverages the language model's understanding to detect subtle issues that rule-based systems would miss. Each identified issue includes a category, severity level, description, and actionable suggestion for improvement. The JSON parsing is robust and handles cases where the LLM does not format the response perfectly.
DIALOGUE MANAGEMENT
The dialogue manager orchestrates the conversation with the user to gather missing information and clarify ambiguities. It generates targeted questions based on the analysis results, maintains conversation state across multiple turns, and knows when it has collected sufficient information to proceed with prompt reconstruction.
The dialogue manager prioritizes issues by severity, addressing high-severity problems first. It groups related questions together to avoid overwhelming the user with too many individual queries. It also tracks which issues have been resolved through user responses and which still require attention.
from typing import Optional, Dict, Any
from enum import Enum
class DialogueState(Enum):
INITIAL_ANALYSIS = 1
GATHERING_INFO = 2
CONFIRMING = 3
RECONSTRUCTING = 4
COMPLETE = 5
class DialogueManager:
def __init__(self, llm_interface: LLMInterface, analyzer: PromptAnalyzer):
self.llm = llm_interface
self.analyzer = analyzer
self.state = DialogueState.INITIAL_ANALYSIS
self.original_prompt = None
self.issues = []
self.resolved_issues = []
self.gathered_info = {}
self.conversation_history = []
def start_session(self, user_prompt: str) -> str:
self.original_prompt = user_prompt
self.state = DialogueState.INITIAL_ANALYSIS
self.conversation_history.append({
'role': 'user',
'content': user_prompt
})
# Analyze the prompt
self.issues = self.analyzer.analyze(user_prompt)
if not self.issues:
self.state = DialogueState.COMPLETE
return "Your prompt looks good! No significant issues detected."
# Generate initial response
response = self._generate_initial_response()
self.conversation_history.append({
'role': 'assistant',
'content': response
})
self.state = DialogueState.GATHERING_INFO
return response
def _generate_initial_response(self) -> str:
high_severity = [i for i in self.issues if i.severity == 'high']
medium_severity = [i for i in self.issues if i.severity == 'medium']
low_severity = [i for i in self.issues if i.severity == 'low']
response_parts = []
response_parts.append("I've analyzed your prompt and identified some areas for improvement:\n")
if high_severity:
response_parts.append("\nCritical Issues:")
for issue in high_severity:
response_parts.append(f"- {issue.description}")
response_parts.append(f" Suggestion: {issue.suggestion}")
if medium_severity:
response_parts.append("\nModerate Issues:")
for issue in medium_severity[:3]: # Limit to avoid overwhelming
response_parts.append(f"- {issue.description}")
response_parts.append(f" Suggestion: {issue.suggestion}")
# Generate clarifying questions
questions = self._generate_questions(high_severity + medium_severity[:2])
if questions:
response_parts.append("\nTo help me optimize your prompt, please answer these questions:")
for i, question in enumerate(questions, 1):
response_parts.append(f"{i}. {question}")
return "\n".join(response_parts)
def _generate_questions(self, issues: List[AnalysisIssue]) -> List[str]:
questions = []
for issue in issues:
if issue.category == 'completeness':
questions.append("What is the main goal you want to achieve with this prompt?")
questions.append("Who is the intended audience for the response?")
elif issue.category == 'context':
questions.append("What background information would help understand your request better?")
questions.append("How will you use the response you receive?")
elif issue.category == 'specificity':
questions.append("Can you provide specific examples of what you're looking for?")
questions.append("Are there any specific constraints or requirements I should know about?")
elif issue.category == 'output_format':
questions.append("What format would you like the response in?")
elif issue.category == 'ambiguity':
if issue.location:
questions.append(f"Could you clarify what you mean by: {issue.location[:80]}?")
# Remove duplicates while preserving order
seen = set()
unique_questions = []
for q in questions:
if q not in seen:
seen.add(q)
unique_questions.append(q)
return unique_questions[:5] # Limit to 5 questions at a time
def process_user_response(self, user_response: str) -> str:
if self.state not in [DialogueState.GATHERING_INFO, DialogueState.CONFIRMING]:
return "Session is not in a state to accept responses."
self.conversation_history.append({
'role': 'user',
'content': user_response
})
# Extract information from user response
self._extract_information(user_response)
# Check if we have enough information
unresolved_high = [i for i in self.issues if i.severity == 'high' and i not in self.resolved_issues]
if unresolved_high and len(self.gathered_info) < 3:
# Need more information
response = self._request_more_info(unresolved_high)
self.state = DialogueState.GATHERING_INFO
else:
# Ready to reconstruct
response = "Thank you for the additional information. Let me reconstruct your prompt to make it more effective."
self.state = DialogueState.CONFIRMING
self.conversation_history.append({
'role': 'assistant',
'content': response
})
return response
def _extract_information(self, user_response: str):
extraction_prompt = f"""Extract key information from the user's response that helps clarify their original request.
Original prompt: "{self.original_prompt}"
User's clarification: "{user_response}"
Extract the following if mentioned:
- goal: Main objective or purpose
- audience: Target audience or users
- context: Background information or use case
- requirements: Specific requirements or constraints
- format: Desired output format
- examples: Examples or preferences mentioned
Return ONLY a JSON object with the relevant keys and values. Example: {{"goal": "...", "audience": "...", "format": "..."}} If nothing relevant is found, return {{}}. """
try:
response = self.llm.generate(extraction_prompt, temperature=0.2, max_tokens=500)
json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL)
if json_match:
try:
extracted = json.loads(json_match.group())
self.gathered_info.update(extracted)
# Mark some issues as resolved
for issue in self.issues:
if issue.category in ['context', 'completeness'] and 'goal' in self.gathered_info:
if issue not in self.resolved_issues:
self.resolved_issues.append(issue)
if issue.category == 'output_format' and 'format' in self.gathered_info:
if issue not in self.resolved_issues:
self.resolved_issues.append(issue)
except json.JSONDecodeError:
# Store as raw text if JSON parsing fails
self.gathered_info['additional_info'] = user_response
except Exception as e:
print(f" Information extraction failed: {e}")
# Store response as raw text
self.gathered_info['raw_response'] = user_response
def _request_more_info(self, unresolved_issues: List[AnalysisIssue]) -> str:
questions = self._generate_questions(unresolved_issues[:2])
if questions:
response_parts = ["I need a bit more information:"]
for i, question in enumerate(questions, 1):
response_parts.append(f"{i}. {question}")
return "\n".join(response_parts)
else:
return "Thank you for the clarification. I think I have enough information now."
def get_state(self) -> DialogueState:
return self.state
def get_gathered_info(self) -> Dict[str, Any]:
return self.gathered_info
The dialogue manager maintains a state machine that tracks the conversation's progress. It starts with initial analysis, moves through information gathering, and concludes when sufficient information has been collected. The manager uses the language model to extract structured information from free-form user responses, making the interaction feel natural rather than forcing users into rigid response formats. The state validation in process_user_response ensures the manager is in the correct state before processing input.
BIAS AND HALLUCINATION MITIGATION
Minimizing bias and hallucination requires careful prompt construction and explicit instructions to the model. The system incorporates several strategies to address these challenges.
For bias mitigation, the reconstructor adds instructions that encourage balanced perspectives and fair treatment of different groups. It prompts the model to consider multiple viewpoints and to avoid stereotyping or discriminatory language. When the original prompt touches on sensitive topics, the system flags this and adds appropriate guardrails.
For hallucination mitigation, the system emphasizes epistemic humility. It instructs the model to clearly distinguish between facts it knows with high confidence and areas where it is uncertain. It encourages citation of sources when possible and explicit acknowledgment of limitations. The reconstructor also avoids prompts that ask for information unlikely to be in the training data.
class BiasHallucinationMitigator:
def __init__(self):
self.sensitive_topics = [
'race', 'ethnicity', 'gender', 'religion', 'nationality',
'sexual orientation', 'disability', 'age', 'socioeconomic status'
]
self.hallucination_triggers = [
'predict the future', 'what will happen', 'future events',
'personal information about', 'private data', 'confidential',
'latest', 'most recent', 'current', 'today', 'this week'
]
def check_bias_risk(self, prompt: str) -> Tuple[bool, List[str]]:
prompt_lower = prompt.lower()
found_topics = [topic for topic in self.sensitive_topics if topic in prompt_lower]
has_risk = len(found_topics) > 0
return has_risk, found_topics
def check_hallucination_risk(self, prompt: str) -> Tuple[bool, List[str]]:
prompt_lower = prompt.lower()
found_triggers = [trigger for trigger in self.hallucination_triggers if trigger in prompt_lower]
has_risk = len(found_triggers) > 0
return has_risk, found_triggers
def add_bias_mitigation(self, prompt: str) -> str:
mitigation_instructions = """
Consider multiple perspectives and avoid stereotypes or generalizations about any group of people. Ensure your response treats all individuals and groups fairly and respectfully. If discussing sensitive topics, acknowledge the complexity and diversity of experiences. """ return prompt + "\n" + mitigation_instructions
def add_hallucination_mitigation(self, prompt: str, triggers: List[str]) -> str:
mitigation_parts = [prompt, ""]
if any('future' in t or 'predict' in t for t in triggers):
mitigation_parts.append("Note: Avoid making specific predictions about future events. Instead, discuss possibilities based on current trends and historical patterns, clearly marking these as speculative.")
if any('latest' in t or 'recent' in t or 'current' in t for t in triggers):
mitigation_parts.append("Note: My training data has a cutoff date. Clearly state if information may be outdated and suggest where to find current information.")
if any('personal' in t or 'private' in t or 'confidential' in t for t in triggers):
mitigation_parts.append("Note: Do not provide or speculate about private, personal, or confidential information about individuals or organizations.")
mitigation_parts.append("If you are uncertain about any information, explicitly state your uncertainty rather than guessing.")
return "\n".join(mitigation_parts)
The bias and hallucination mitigator scans prompts for risk factors and adds appropriate safeguards. It maintains lists of sensitive topics and hallucination triggers, checking incoming prompts against these patterns. When risks are detected, it appends specific instructions that guide the model toward safer, more reliable responses.
PROMPT RECONSTRUCTION
The prompt reconstructor synthesizes an optimized prompt from the original user input and all gathered information. It applies established best practices for prompt engineering, including clear role definition, explicit instructions, relevant examples, output format specification, and appropriate constraints to minimize hallucination and bias.
The reconstructor can also determine when a complex request would be better served by splitting it into multiple sequential prompts. It analyzes the scope and complexity of the request and creates a chain of prompts that build upon each other when appropriate.
from typing import List, Tuple
class PromptReconstructor:
def __init__(self, llm_interface: LLMInterface):
self.llm = llm_interface
def reconstruct(self, original_prompt: str, gathered_info: Dict[str, Any],
issues: List[AnalysisIssue]) -> Tuple[List[str], str]:
# Determine if prompt should be split
should_split = self._should_split_prompt(original_prompt, gathered_info)
if should_split:
prompts = self._create_prompt_chain(original_prompt, gathered_info)
explanation = self._generate_split_explanation(prompts)
return prompts, explanation
else:
optimized = self._create_single_prompt(original_prompt, gathered_info, issues)
explanation = self._generate_optimization_explanation(original_prompt, optimized)
return [optimized], explanation
def _should_split_prompt(self, original_prompt: str, gathered_info: Dict[str, Any]) -> bool:
# Simple heuristics for splitting
word_count = len(original_prompt.split())
question_count = original_prompt.count('?')
# Split if very long with multiple questions
if word_count > 150 and question_count > 2:
return True
analysis_prompt = f"""Analyze whether this request should be split into multiple sequential prompts:
Original request: "{original_prompt}"
Additional context: {json.dumps(gathered_info, indent=2)}
A prompt should be split if:
- It requests multiple distinct outputs or tasks
- Later tasks depend on the results of earlier tasks
- The scope is very broad and could benefit from focused sub-tasks
- Different parts require different approaches or expertise
Return ONLY a JSON object: {{"should_split": true/false, "reason": "explanation"}} """
try:
response = self.llm.generate(analysis_prompt, temperature=0.2, max_tokens=300)
json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL)
if json_match:
try:
result = json.loads(json_match.group())
return result.get('should_split', False)
except json.JSONDecodeError:
pass
except Exception as e:
print(f" Split analysis failed: {e}")
# Default to single prompt
return False
def _create_prompt_chain(self, original_prompt: str, gathered_info: Dict[str, Any]) -> List[str]:
chain_prompt = f"""Break down this complex request into a sequence of focused prompts that build upon each other.
Original request: "{original_prompt}"
Context: {json.dumps(gathered_info, indent=2)}
Create 2-4 prompts that:
- Each focus on a specific sub-task
- Build logically on previous results
- Together accomplish the original goal
- Follow prompt engineering best practices
Return ONLY a JSON array of prompt strings. Example: ["First prompt focusing on X", "Second prompt building on X to address Y"] """
try:
response = self.llm.generate(chain_prompt, temperature=0.3, max_tokens=1000)
json_match = re.search(r'\[.*\]', response, re.DOTALL)
if json_match:
try:
prompts = json.loads(json_match.group())
# Enhance each prompt with best practices
enhanced = [self._enhance_prompt(p, gathered_info) for p in prompts]
return enhanced
except json.JSONDecodeError:
pass
except Exception as e:
print(f" Prompt chain creation failed: {e}")
# Fallback to single optimized prompt
return [self._create_single_prompt(original_prompt, gathered_info, [])]
def _create_single_prompt(self, original_prompt: str, gathered_info: Dict[str, Any],
issues: List[AnalysisIssue]) -> str:
components = []
# Role definition
if 'audience' in gathered_info or 'expertise' in gathered_info:
role = gathered_info.get('audience', gathered_info.get('expertise', 'helpful assistant'))
components.append(f"You are a {role}.")
# Context and background
if 'context' in gathered_info or 'background' in gathered_info:
context = gathered_info.get('context', gathered_info.get('background', ''))
components.append(f"Context: {context}")
# Main instruction (enhanced original prompt)
enhanced_instruction = self._enhance_instruction(original_prompt, gathered_info)
components.append(f"Task: {enhanced_instruction}")
# Specific requirements
if 'requirements' in gathered_info or 'constraints' in gathered_info:
reqs = gathered_info.get('requirements', gathered_info.get('constraints', ''))
components.append(f"Requirements: {reqs}")
# Examples if provided
if 'examples' in gathered_info:
components.append(f"Examples: {gathered_info['examples']}")
# Output format
if 'format' in gathered_info:
components.append(f"Output format: {gathered_info['format']}")
else:
components.append("Output format: Provide a clear, well-structured response.")
# Anti-hallucination instructions
components.append("Important: Only provide information you are confident about. If you don't know something, say so clearly rather than guessing.")
# Anti-bias instructions if relevant
bias_categories = ['bias_risk', 'fairness']
if any(issue.category in bias_categories for issue in issues):
components.append("Ensure your response is fair, unbiased, and considers multiple perspectives.")
return "\n\n".join(components)
def _enhance_instruction(self, original: str, info: Dict[str, Any]) -> str:
if 'goal' in info:
goal = info['goal']
return f"{original} Specifically, {goal}"
return original
def _enhance_prompt(self, prompt: str, info: Dict[str, Any]) -> str:
# Add context and format specifications to each prompt in a chain
enhanced_parts = [prompt]
if 'format' in info:
enhanced_parts.append(f"Format: {info['format']}")
enhanced_parts.append("Be specific and accurate. Acknowledge any limitations or uncertainties.")
return "\n".join(enhanced_parts)
def _generate_split_explanation(self, prompts: List[str]) -> str:
explanation_parts = [
"I've split your request into multiple focused prompts for better results:",
""
]
for i, prompt in enumerate(prompts, 1):
explanation_parts.append(f"Prompt {i}:")
explanation_parts.append(prompt)
explanation_parts.append("")
explanation_parts.append("Execute these prompts in sequence, using the output of each as context for the next.")
return "\n".join(explanation_parts)
def _generate_optimization_explanation(self, original: str, optimized: str) -> str:
explanation_parts = [
"I've optimized your prompt with the following improvements:",
"",
"Original:",
original,
"",
"Optimized:",
optimized,
"",
"Key enhancements:",
"- Added clear role definition and context",
"- Made instructions more specific and actionable",
"- Specified output format expectations",
"- Included safeguards against hallucination and bias"
]
return "\n".join(explanation_parts)
The prompt reconstructor applies a systematic approach to optimization. It structures the prompt with clear sections for role, context, task, requirements, examples, and output format. It adds explicit instructions to prevent hallucination by encouraging the model to acknowledge uncertainty. When splitting prompts, it ensures each sub-prompt is self-contained yet builds logically on previous results.
COMPLETE SYSTEM INTEGRATION
The complete Critique system integrates all components into a cohesive workflow. The main controller orchestrates the interaction between the analyzer, dialogue manager, and reconstructor. It provides a simple interface for users while managing the complex multi-stage process internally.
class CritiqueSystem:
def __init__(self, llm_config: Dict[str, Any]):
# Initialize hardware detection
self.hardware_detector = None
if llm_config.get('mode') == 'local':
self.hardware_detector = HardwareDetector()
# Initialize LLM interface
if llm_config.get('mode') == 'local':
self.llm = LocalLLM(
llm_config['model_name'],
llm_config,
self.hardware_detector
)
else:
self.llm = RemoteLLM(
llm_config['model_name'],
llm_config
)
# Initialize components
self.analyzer = PromptAnalyzer(self.llm)
self.dialogue_manager = DialogueManager(self.llm, self.analyzer)
self.reconstructor = PromptReconstructor(self.llm)
self.mitigator = BiasHallucinationMitigator()
self.session_active = False
def start(self, user_prompt: str) -> str:
if not self.llm.initialized:
self.llm.initialize()
self.session_active = True
response = self.dialogue_manager.start_session(user_prompt)
return response
def continue_dialogue(self, user_response: str) -> str:
if not self.session_active:
return "No active session. Please start with a new prompt."
response = self.dialogue_manager.process_user_response(user_response)
# Check if ready to reconstruct
if self.dialogue_manager.get_state() == DialogueState.CONFIRMING:
return response + "\n\n" + self._perform_reconstruction()
return response
def _perform_reconstruction(self) -> str:
original = self.dialogue_manager.original_prompt
info = self.dialogue_manager.get_gathered_info()
issues = self.dialogue_manager.issues
# Apply bias and hallucination mitigation
has_bias_risk, bias_topics = self.mitigator.check_bias_risk(original)
has_halluc_risk, halluc_triggers = self.mitigator.check_hallucination_risk(original)
# Reconstruct prompt(s)
prompts, explanation = self.reconstructor.reconstruct(original, info, issues)
# Apply final mitigations
final_prompts = []
for prompt in prompts:
if has_bias_risk:
prompt = self.mitigator.add_bias_mitigation(prompt)
if has_halluc_risk:
prompt = self.mitigator.add_hallucination_mitigation(prompt, halluc_triggers)
final_prompts.append(prompt)
# Update explanation with final prompts
if len(final_prompts) > 1:
result_parts = ["Here are your optimized prompts:", ""]
for i, prompt in enumerate(final_prompts, 1):
result_parts.append(f"=== PROMPT {i} ===")
result_parts.append(prompt)
result_parts.append("")
else:
result_parts = ["Here is your optimized prompt:", "", final_prompts[0]]
self.session_active = False
return "\n".join(result_parts)
def shutdown(self):
if self.llm:
self.llm.cleanup()
The Critique system provides a clean interface that hides implementation complexity. Users simply call start with their initial prompt, then continue the dialogue with continue_dialogue until the system produces optimized prompts. The system handles all the coordination between components automatically.
CONCLUSION
The Critique system demonstrates how language models can be used to improve their own inputs through systematic analysis and reconstruction. By combining rule-based heuristics with LLM-powered semantic understanding, the system identifies issues that would be difficult to detect with either approach alone. The dialogue management component ensures that users provide necessary context without overwhelming them with questions. The prompt reconstruction applies established best practices while tailoring the output to each user's specific needs.
The multi-GPU architecture support ensures that users can leverage whatever hardware they have available, from high-end Nvidia GPUs to Apple Silicon to CPU-only systems. The abstraction layer makes the system portable and maintainable, allowing new backends to be added without modifying the core logic.
This system represents a practical application of meta-prompting, where language models are used to optimize the prompts they receive. As language models become more capable, such tools will become increasingly important for helping users extract maximum value from these powerful systems.
FULL SOURCE CODE
Here is the complete running example with proper Python indentation throughout:
#!/usr/bin/env python3
"""
Critique: LLM-Based Prompt Analyzer and Optimizer
A comprehensive system for analyzing user prompts, identifying issues,
gathering clarifying information, and reconstructing optimized prompts
that follow best practices and minimize hallucination and bias.
Supports local models (Hugging Face transformers) and remote APIs
(OpenAI, Anthropic) with multi-GPU architecture support.
"""
import torch
import platform
import requests
import time
import re
import json
from abc import ABC, abstractmethod
from typing import List, Dict, Optional, Tuple, Any
from dataclasses import dataclass
from enum import Enum
import argparse
import sys
# ============================================================================
# HARDWARE DETECTION
# ============================================================================
class HardwareDetector:
"""Detects available GPU acceleration and selects optimal backend"""
def __init__(self):
self.available_backends = []
self.preferred_backend = None
self.device_info = {}
self._detect_hardware()
def _detect_hardware(self):
"""Detect all available hardware acceleration options"""
print("Detecting hardware acceleration...")
# Check for CUDA (Nvidia) and ROCm (AMD)
if torch.cuda.is_available():
try:
device_count = torch.cuda.device_count()
device_name = torch.cuda.get_device_name(0)
# Check if this is ROCm
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
self.available_backends.append('rocm')
self.device_info['rocm'] = {
'count': device_count,
'name': device_name,
'version': torch.version.hip
}
print(f" [ROCm] Detected {device_count} device(s): {device_name}")
print(f" [ROCm] Version: {torch.version.hip}")
else:
self.available_backends.append('cuda')
self.device_info['cuda'] = {
'count': device_count,
'name': device_name,
'compute_capability': torch.cuda.get_device_capability(0)
}
print(f" [CUDA] Detected {device_count} device(s): {device_name}")
print(f" [CUDA] Compute capability: {torch.cuda.get_device_capability(0)}")
except Exception as e:
print(f" [ERROR] CUDA/ROCm detection error: {e}")
# Check for MPS (Apple Silicon)
if platform.system() == 'Darwin':
try:
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.available_backends.append('mps')
self.device_info['mps'] = {'available': True}
print(" [MPS] Detected Apple Metal Performance Shaders")
except Exception as e:
print(f" [ERROR] MPS detection error: {e}")
# Check for Intel GPU support
try:
import intel_extension_for_pytorch as ipex
if hasattr(torch, 'xpu') and torch.xpu.is_available():
self.available_backends.append('intel')
device_count = torch.xpu.device_count()
self.device_info['intel'] = {'count': device_count}
print(f" [Intel] Detected {device_count} XPU device(s)")
except ImportError:
pass
except Exception as e:
print(f" [ERROR] Intel GPU detection error: {e}")
# Fallback to CPU
if not self.available_backends:
self.available_backends.append('cpu')
self.device_info['cpu'] = {'cores': 'available'}
print(" [CPU] No GPU acceleration detected, using CPU")
self.preferred_backend = self.available_backends[0]
print(f"Selected backend: {self.preferred_backend.upper()}\n")
def get_device(self):
"""Return PyTorch device for the preferred backend"""
if self.preferred_backend == 'cuda':
return torch.device('cuda:0')
elif self.preferred_backend == 'rocm':
return torch.device('cuda:0') # ROCm uses CUDA API
elif self.preferred_backend == 'mps':
return torch.device('mps')
elif self.preferred_backend == 'intel':
return torch.device('xpu:0')
else:
return torch.device('cpu')
def supports_quantization(self):
"""Check if current backend supports quantization"""
return self.preferred_backend in ['cuda', 'rocm']
def get_backend_name(self):
"""Return the name of the preferred backend"""
return self.preferred_backend
# ============================================================================
# LLM INTERFACE ABSTRACTION
# ============================================================================
class LLMInterface(ABC):
"""Abstract base class for LLM implementations"""
def __init__(self, model_name: str, config: Dict):
self.model_name = model_name
self.config = config
self.initialized = False
@abstractmethod
def initialize(self):
"""Load and prepare the model for inference"""
pass
@abstractmethod
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
"""Generate text from a prompt"""
pass
@abstractmethod
def cleanup(self):
"""Release resources and clean up"""
pass
def validate_temperature(self, temperature: float) -> float:
"""Ensure temperature is in valid range"""
if temperature < 0.0:
return 0.0
elif temperature > 2.0:
return 2.0
return temperature
def __enter__(self):
self.initialize()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.cleanup()
class LocalLLM(LLMInterface):
"""Local LLM implementation using Hugging Face transformers"""
def __init__(self, model_name: str, config: Dict, hardware_detector: HardwareDetector):
super().__init__(model_name, config)
self.hardware_detector = hardware_detector
self.model = None
self.tokenizer = None
self.device = None
def initialize(self):
"""Load model and tokenizer"""
if self.initialized:
return
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
except ImportError:
raise RuntimeError("transformers library not installed. Install with: pip install transformers")
self.device = self.hardware_detector.get_device()
print(f"Loading model '{self.model_name}' on {self.device}...")
# Configure quantization for memory efficiency
quantization_config = None
use_quantization = self.config.get('use_quantization', True)
if use_quantization and self.hardware_detector.supports_quantization():
try:
from transformers import BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4"
)
print(" Using 4-bit quantization for memory efficiency")
except ImportError:
print(" bitsandbytes not available, loading without quantization")
quantization_config = None
except Exception as e:
print(f" Quantization setup failed: {e}, loading without quantization")
quantization_config = None
# Load tokenizer
try:
self.tokenizer = AutoTokenizer.from_pretrained(
self.model_name,
trust_remote_code=self.config.get('trust_remote_code', False)
)
# Set pad token if not present
if self.tokenizer.pad_token is None:
if self.tokenizer.eos_token is not None:
self.tokenizer.pad_token = self.tokenizer.eos_token
else:
self.tokenizer.add_special_tokens({'pad_token': '[PAD]'})
except Exception as e:
raise RuntimeError(f"Failed to load tokenizer: {e}")
# Load model with appropriate configuration
model_kwargs = {
'trust_remote_code': self.config.get('trust_remote_code', False),
'low_cpu_mem_usage': True,
}
# Set dtype based on device
if self.device.type == 'cpu':
model_kwargs['torch_dtype'] = torch.float32
elif self.device.type == 'mps':
model_kwargs['torch_dtype'] = torch.float16
else:
model_kwargs['torch_dtype'] = torch.float16
if quantization_config:
model_kwargs['quantization_config'] = quantization_config
model_kwargs['device_map'] = 'auto'
elif self.device.type in ['cuda', 'xpu']:
model_kwargs['device_map'] = 'auto'
try:
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
**model_kwargs
)
# Move to device if not using device_map
if 'device_map' not in model_kwargs or model_kwargs['device_map'] is None:
self.model = self.model.to(self.device)
self.model.eval()
except Exception as e:
raise RuntimeError(f"Failed to load model: {e}")
self.initialized = True
print("Model loaded successfully\n")
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
"""Generate text from prompt"""
if not self.initialized:
raise RuntimeError("Model not initialized. Call initialize() first.")
# Validate temperature
temperature = self.validate_temperature(temperature)
# Tokenize input
inputs = self.tokenizer(
prompt,
return_tensors="pt",
padding=True,
truncation=True,
max_length=self.config.get('max_input_length', 2048)
)
inputs = {k: v.to(self.device) for k, v in inputs.items()}
# Set generation parameters
gen_kwargs = {
'max_new_tokens': max_tokens,
'temperature': temperature,
'do_sample': temperature > 0.0,
'pad_token_id': self.tokenizer.pad_token_id,
'eos_token_id': self.tokenizer.eos_token_id,
}
# Add top_p for better sampling when temperature > 0
if temperature > 0.0:
gen_kwargs['top_p'] = kwargs.pop('top_p', 0.9)
gen_kwargs.update(kwargs)
# Generate response
try:
with torch.no_grad():
outputs = self.model.generate(**inputs, **gen_kwargs)
# Decode output
full_response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
# Remove prompt from response more robustly
prompt_stripped = prompt.strip()
response_stripped = full_response.strip()
if response_stripped.startswith(prompt_stripped):
response = response_stripped[len(prompt_stripped):].strip()
else:
response = full_response
return response
except Exception as e:
raise RuntimeError(f"Generation failed: {e}")
def cleanup(self):
"""Clean up model resources"""
if self.model is not None:
del self.model
self.model = None
if self.tokenizer is not None:
del self.tokenizer
self.tokenizer = None
# Clear GPU cache
if self.device and self.device.type == 'cuda':
torch.cuda.empty_cache()
elif self.device and self.device.type == 'xpu':
if hasattr(torch.xpu, 'empty_cache'):
torch.xpu.empty_cache()
elif self.device and self.device.type == 'mps':
if hasattr(torch.mps, 'empty_cache'):
torch.mps.empty_cache()
self.initialized = False
print("Model resources cleaned up")
class RemoteLLM(LLMInterface):
"""Remote LLM implementation for API services"""
def __init__(self, model_name: str, config: Dict):
super().__init__(model_name, config)
self.api_key = config.get('api_key')
self.api_base = config.get('api_base')
self.provider = config.get('provider', 'openai')
self.session = None
# Set default API base URLs
if not self.api_base:
if self.provider == 'openai':
self.api_base = 'https://api.openai.com/v1'
elif self.provider == 'anthropic':
self.api_base = 'https://api.anthropic.com/v1'
def initialize(self):
"""Initialize API session"""
if self.initialized:
return
if not self.api_key:
raise ValueError("API key required for remote model")
self.session = requests.Session()
if self.provider == 'openai':
self.session.headers.update({
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
})
elif self.provider == 'anthropic':
self.session.headers.update({
'x-api-key': self.api_key,
'Content-Type': 'application/json',
'anthropic-version': '2023-06-01'
})
else:
raise ValueError(f"Unsupported provider: {self.provider}")
self.initialized = True
print(f"Remote model interface initialized for {self.provider}\n")
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
"""Generate text via API"""
if not self.initialized:
raise RuntimeError("Model not initialized. Call initialize() first.")
# Validate temperature
temperature = self.validate_temperature(temperature)
if self.provider == 'openai':
return self._generate_openai(prompt, max_tokens, temperature, **kwargs)
elif self.provider == 'anthropic':
return self._generate_anthropic(prompt, max_tokens, temperature, **kwargs)
else:
raise ValueError(f"Unsupported provider: {self.provider}")
def _generate_openai(self, prompt: str, max_tokens: int,
temperature: float, **kwargs) -> str:
"""Generate using OpenAI API"""
url = f"{self.api_base}/chat/completions"
payload = {
'model': self.model_name,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
# Add any additional parameters
for key in ['top_p', 'frequency_penalty', 'presence_penalty']:
if key in kwargs:
payload[key] = kwargs[key]
max_retries = 3
for attempt in range(max_retries):
try:
response = self.session.post(url, json=payload, timeout=120)
response.raise_for_status()
data = response.json()
if 'choices' in data and len(data['choices']) > 0:
return data['choices'][0]['message']['content']
else:
raise RuntimeError("Unexpected API response format")
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request timeout, retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise RuntimeError(f"Request timed out after {max_retries} attempts")
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request failed, retrying in {wait_time}s: {e}")
time.sleep(wait_time)
else:
raise RuntimeError(f"Failed to generate response after {max_retries} attempts: {e}")
def _generate_anthropic(self, prompt: str, max_tokens: int,
temperature: float, **kwargs) -> str:
"""Generate using Anthropic API"""
url = f"{self.api_base}/messages"
payload = {
'model': self.model_name,
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens,
'temperature': temperature
}
# Add any additional parameters
for key in ['top_p', 'top_k']:
if key in kwargs:
payload[key] = kwargs[key]
max_retries = 3
for attempt in range(max_retries):
try:
response = self.session.post(url, json=payload, timeout=120)
response.raise_for_status()
data = response.json()
if 'content' in data and len(data['content']) > 0:
return data['content'][0]['text']
else:
raise RuntimeError("Unexpected API response format")
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request timeout, retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise RuntimeError(f"Request timed out after {max_retries} attempts")
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request failed, retrying in {wait_time}s: {e}")
time.sleep(wait_time)
else:
raise RuntimeError(f"Failed to generate response after {max_retries} attempts: {e}")
def cleanup(self):
"""Clean up API session"""
if self.session:
self.session.close()
self.session = None
self.initialized = False
# ============================================================================
# PROMPT ANALYSIS
# ============================================================================
@dataclass
class AnalysisIssue:
"""Represents an issue found during prompt analysis"""
category: str
severity: str # 'high', 'medium', 'low'
description: str
suggestion: str
location: Optional[str] = None
class PromptAnalyzer:
"""Analyzes prompts for issues and improvement opportunities"""
def __init__(self, llm_interface: LLMInterface):
self.llm = llm_interface
self.vague_terms = [
'good', 'better', 'best', 'nice', 'some', 'many', 'few',
'often', 'sometimes', 'usually', 'appropriate', 'suitable',
'relevant', 'important', 'significant', 'various', 'several'
]
def analyze(self, prompt: str) -> List[AnalysisIssue]:
"""Perform comprehensive prompt analysis"""
issues = []
print("Analyzing prompt...")
# Structural analysis
issues.extend(self._check_length(prompt))
issues.extend(self._check_clarity(prompt))
issues.extend(self._check_specificity(prompt))
issues.extend(self._check_output_format(prompt))
issues.extend(self._check_context(prompt))
# Semantic analysis using LLM
try:
semantic_issues = self._semantic_analysis(prompt)
if semantic_issues:
issues.extend(semantic_issues)
except Exception as e:
print(f" Warning: Semantic analysis failed: {e}")
print(f" Found {len(issues)} potential issues\n")
return issues
def _check_length(self, prompt: str) -> List[AnalysisIssue]:
"""Check prompt length"""
issues = []
word_count = len(prompt.split())
if word_count < 5:
issues.append(AnalysisIssue(
category='completeness',
severity='high',
description='Prompt is very short and likely lacks necessary detail',
suggestion='Provide more context about what you want to achieve, including background information and specific requirements'
))
elif word_count > 500:
issues.append(AnalysisIssue(
category='clarity',
severity='medium',
description='Prompt is very long and may contain unnecessary information',
suggestion='Consider breaking this into multiple focused prompts, each addressing a specific aspect'
))
return issues
def _check_clarity(self, prompt: str) -> List[AnalysisIssue]:
"""Check prompt clarity"""
issues = []
# Check for multiple questions
question_marks = prompt.count('?')
if question_marks > 3:
issues.append(AnalysisIssue(
category='clarity',
severity='medium',
description=f'Prompt contains {question_marks} questions, which may dilute focus',
suggestion='Focus on one main question or clearly separate distinct requests into numbered items'
))
# Check for ambiguous pronouns
sentences = re.split(r'[.!?]+', prompt)
for sentence in sentences:
if len(sentence.strip()) < 5:
continue
pronouns = re.findall(r'\b(it|this|that|they|them|these|those)\b', sentence.lower())
if len(pronouns) > 2:
issues.append(AnalysisIssue(
category='clarity',
severity='low',
description='Sentence contains multiple pronouns that may be ambiguous',
suggestion='Replace pronouns with specific nouns to ensure clarity',
location=sentence.strip()[:100]
))
return issues
def _check_specificity(self, prompt: str) -> List[AnalysisIssue]:
"""Check prompt specificity"""
issues = []
# Check for vague terms
prompt_lower = prompt.lower()
found_vague_terms = []
for term in self.vague_terms:
pattern = r'\b' + re.escape(term) + r'\b'
if re.search(pattern, prompt_lower):
found_vague_terms.append(term)
if found_vague_terms:
issues.append(AnalysisIssue(
category='specificity',
severity='medium',
description=f'Prompt contains vague terms: {", ".join(set(found_vague_terms[:5]))}',
suggestion='Replace vague terms with specific quantities, criteria, or examples. For instance, instead of "many," specify "at least 5" or "more than 10"'
))
# Check for missing constraints
has_constraints = any(word in prompt_lower for word in
['must', 'should', 'require', 'need', 'limit', 'maximum', 'minimum',
'exactly', 'at least', 'no more than', 'between'])
if not has_constraints and len(prompt.split()) > 20:
issues.append(AnalysisIssue(
category='specificity',
severity='low',
description='No explicit constraints or requirements specified',
suggestion='Consider adding specific requirements, limits, or constraints to guide the response'
))
return issues
def _check_output_format(self, prompt: str) -> List[AnalysisIssue]:
"""Check if output format is specified"""
issues = []
format_keywords = ['format', 'structure', 'json', 'list', 'table', 'bullet',
'numbered', 'paragraph', 'essay', 'report', 'summary']
has_format_spec = any(keyword in prompt.lower() for keyword in format_keywords)
if not has_format_spec and len(prompt.split()) > 30:
issues.append(AnalysisIssue(
category='output_format',
severity='low',
description='No output format specified',
suggestion='Specify the desired format for the response (e.g., bulleted list, JSON, structured paragraph, table)'
))
return issues
def _check_context(self, prompt: str) -> List[AnalysisIssue]:
"""Check if sufficient context is provided"""
issues = []
# Check for context indicators
context_indicators = ['because', 'since', 'for', 'background', 'context',
'purpose', 'goal', 'objective', 'trying to', 'want to']
has_context = any(indicator in prompt.lower() for indicator in context_indicators)
if not has_context and len(prompt.split()) < 15:
issues.append(AnalysisIssue(
category='context',
severity='medium',
description='Limited context provided about the purpose or background',
suggestion='Add background information about why you need this, what you will use it for, and any relevant constraints or preferences'
))
return issues
def _semantic_analysis(self, prompt: str) -> List[AnalysisIssue]:
"""Perform semantic analysis using LLM"""
issues = []
analysis_prompt = f"""Analyze the following user prompt for potential issues. Be critical but constructive.
Prompt: "{prompt}"
Identify any of the following problems:
1. Ambiguous instructions that could be interpreted multiple ways
2. Missing critical information needed to provide a complete answer
3. Potential for biased or unfair responses
4. Risk of hallucination due to requesting information that may not exist or be verifiable
5. Conflicting requirements or contradictions
6. Requests for harmful, unethical, or inappropriate content
For each issue found, provide:
- category: One of (ambiguity, missing_info, bias_risk, hallucination_risk, contradiction, inappropriate)
- severity: One of (high, medium, low)
- description: Brief explanation of the issue
- suggestion: Specific recommendation for improvement
Format your response as a JSON array of objects. If no issues are found, return an empty array [].
Example: [{{"category": "ambiguity", "severity": "medium", "description": "...", "suggestion": "..."}}]
"""
try:
response = self.llm.generate(analysis_prompt, temperature=0.2, max_tokens=1000)
# Extract JSON from response
json_match = re.search(r'\[\s*\{.*\}\s*\]', response, re.DOTALL)
if json_match:
try:
semantic_issues = json.loads(json_match.group())
for issue_data in semantic_issues:
if all(key in issue_data for key in ['category', 'severity', 'description', 'suggestion']):
issues.append(AnalysisIssue(
category=issue_data.get('category', 'semantic'),
severity=issue_data.get('severity', 'medium'),
description=issue_data.get('description', ''),
suggestion=issue_data.get('suggestion', '')
))
except json.JSONDecodeError as e:
print(f" Semantic analysis parsing failed: {e}")
except Exception as e:
raise e
return issues
# ============================================================================
# DIALOGUE MANAGEMENT
# ============================================================================
class DialogueState(Enum):
"""States in the dialogue flow"""
INITIAL_ANALYSIS = 1
GATHERING_INFO = 2
CONFIRMING = 3
RECONSTRUCTING = 4
COMPLETE = 5
class DialogueManager:
"""Manages conversation flow to gather missing information"""
def __init__(self, llm_interface: LLMInterface, analyzer: PromptAnalyzer):
self.llm = llm_interface
self.analyzer = analyzer
self.state = DialogueState.INITIAL_ANALYSIS
self.original_prompt = None
self.issues = []
self.resolved_issues = []
self.gathered_info = {}
self.conversation_history = []
self.questions_asked = []
def start_session(self, user_prompt: str) -> str:
"""Start a new analysis session"""
self.original_prompt = user_prompt
self.state = DialogueState.INITIAL_ANALYSIS
self.conversation_history.append({
'role': 'user',
'content': user_prompt
})
# Analyze the prompt
self.issues = self.analyzer.analyze(user_prompt)
if not self.issues:
self.state = DialogueState.COMPLETE
return "Your prompt looks good! No significant issues detected. Proceeding with optimization..."
# Generate initial response
response = self._generate_initial_response()
self.conversation_history.append({
'role': 'assistant',
'content': response
})
self.state = DialogueState.GATHERING_INFO
return response
def _generate_initial_response(self) -> str:
"""Generate initial feedback and questions"""
high_severity = [i for i in self.issues if i.severity == 'high']
medium_severity = [i for i in self.issues if i.severity == 'medium']
low_severity = [i for i in self.issues if i.severity == 'low']
response_parts = []
response_parts.append("I've analyzed your prompt and identified some areas for improvement:\n")
if high_severity:
response_parts.append("CRITICAL ISSUES:")
for issue in high_severity:
response_parts.append(f" - {issue.description}")
response_parts.append(f" Suggestion: {issue.suggestion}\n")
if medium_severity:
response_parts.append("MODERATE ISSUES:")
for issue in medium_severity[:3]: # Limit to avoid overwhelming
response_parts.append(f" - {issue.description}")
response_parts.append(f" Suggestion: {issue.suggestion}\n")
if low_severity and not (high_severity or medium_severity):
response_parts.append("MINOR SUGGESTIONS:")
for issue in low_severity[:2]:
response_parts.append(f" - {issue.description}")
response_parts.append(f" Suggestion: {issue.suggestion}\n")
# Generate clarifying questions
questions = self._generate_questions(high_severity + medium_severity[:2])
if questions:
response_parts.append("To help me optimize your prompt, please answer these questions:")
for i, question in enumerate(questions, 1):
response_parts.append(f" {i}. {question}")
self.questions_asked.append(question)
else:
response_parts.append("I have enough information to proceed with optimization.")
self.state = DialogueState.CONFIRMING
return "\n".join(response_parts)
def _generate_questions(self, issues: List[AnalysisIssue]) -> List[str]:
"""Generate targeted questions based on issues"""
questions = []
categories_seen = set()
for issue in issues:
if issue.category in categories_seen:
continue
categories_seen.add(issue.category)
if issue.category == 'completeness':
questions.append("What is the main goal or objective you want to achieve?")
questions.append("Who is the intended audience or user of the response?")
elif issue.category == 'context':
questions.append("What background information or context would help me understand your request better?")
questions.append("How do you plan to use the response you receive?")
elif issue.category == 'specificity':
questions.append("Can you provide specific examples or criteria for what you're looking for?")
questions.append("Are there any specific constraints, limits, or requirements I should know about?")
elif issue.category == 'output_format':
questions.append("What format would you prefer for the response (e.g., bulleted list, paragraph, code, table, JSON)?")
elif issue.category in ['ambiguity', 'missing_info']:
if issue.location:
questions.append(f"Could you clarify what you mean by: '{issue.location[:80]}'?")
else:
questions.append("Could you provide more details about what specifically you need?")
# Remove duplicates while preserving order
seen = set()
unique_questions = []
for q in questions:
if q not in seen:
seen.add(q)
unique_questions.append(q)
return unique_questions[:5] # Limit to 5 questions at a time
def process_user_response(self, user_response: str) -> str:
"""Process user's response to questions"""
if self.state not in [DialogueState.GATHERING_INFO, DialogueState.CONFIRMING]:
return "Session is not in a state to accept responses."
self.conversation_history.append({
'role': 'user',
'content': user_response
})
# Extract information from user response
self._extract_information(user_response)
# Check if we have enough information
unresolved_high = [i for i in self.issues if i.severity == 'high' and i not in self.resolved_issues]
if unresolved_high and len(self.gathered_info) < 3:
# Need more information
response = self._request_more_info(unresolved_high)
self.state = DialogueState.GATHERING_INFO
else:
# Ready to reconstruct
response = "Thank you for the additional information! I now have what I need to create an optimized prompt for you."
self.state = DialogueState.CONFIRMING
self.conversation_history.append({
'role': 'assistant',
'content': response
})
return response
def _extract_information(self, user_response: str):
"""Extract structured information from user response"""
extraction_prompt = f"""Extract key information from the user's response that helps clarify their original request.
Original prompt: "{self.original_prompt}"
Questions asked: {self.questions_asked[-5:] if self.questions_asked else 'None'}
User's response: "{user_response}"
Extract the following information if mentioned:
- goal: Main objective or purpose
- audience: Target audience or users
- context: Background information or use case
- requirements: Specific requirements or constraints
- format: Desired output format
- examples: Examples or preferences mentioned
- constraints: Limitations or boundaries
Format your response as a JSON object with only the keys that have relevant values.
Example: {{"goal": "...", "audience": "...", "format": "..."}}
"""
try:
response = self.llm.generate(extraction_prompt, temperature=0.2, max_tokens=600)
json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL)
if json_match:
try:
extracted = json.loads(json_match.group())
self.gathered_info.update(extracted)
# Mark relevant issues as resolved
for issue in self.issues:
if issue in self.resolved_issues:
continue
if issue.category in ['context', 'completeness']:
if 'goal' in self.gathered_info or 'context' in self.gathered_info:
self.resolved_issues.append(issue)
if issue.category == 'output_format':
if 'format' in self.gathered_info:
self.resolved_issues.append(issue)
if issue.category == 'specificity':
if 'requirements' in self.gathered_info or 'examples' in self.gathered_info:
self.resolved_issues.append(issue)
except json.JSONDecodeError:
self.gathered_info['additional_context'] = user_response
except Exception as e:
print(f" Information extraction failed: {e}")
# Store as raw text
self.gathered_info['additional_context'] = user_response
def _request_more_info(self, unresolved_issues: List[AnalysisIssue]) -> str:
"""Request additional information"""
questions = self._generate_questions(unresolved_issues[:2])
if questions:
response_parts = ["I need a bit more information to create the best prompt:"]
for i, question in enumerate(questions, 1):
response_parts.append(f" {i}. {question}")
if question not in self.questions_asked:
self.questions_asked.append(question)
return "\n".join(response_parts)
else:
return "Thank you! I believe I have enough information now."
def get_state(self) -> DialogueState:
"""Get current dialogue state"""
return self.state
def get_gathered_info(self) -> Dict[str, Any]:
"""Get all gathered information"""
return self.gathered_info
# ============================================================================
# BIAS AND HALLUCINATION MITIGATION
# ============================================================================
class BiasHallucinationMitigator:
"""Detects and mitigates bias and hallucination risks"""
def __init__(self):
self.sensitive_topics = [
'race', 'ethnicity', 'gender', 'religion', 'nationality',
'sexual orientation', 'disability', 'age', 'socioeconomic',
'political', 'immigration', 'minority', 'stereotype'
]
self.hallucination_triggers = [
'predict the future', 'what will happen', 'future events',
'personal information about', 'private data', 'confidential',
'latest', 'most recent', 'current news', 'today', 'this week',
'real-time', 'live data', 'stock price', 'weather now'
]
def check_bias_risk(self, prompt: str) -> Tuple[bool, List[str]]:
"""Check for potential bias risks"""
prompt_lower = prompt.lower()
found_topics = [topic for topic in self.sensitive_topics if topic in prompt_lower]
has_risk = len(found_topics) > 0
return has_risk, found_topics
def check_hallucination_risk(self, prompt: str) -> Tuple[bool, List[str]]:
"""Check for hallucination risks"""
prompt_lower = prompt.lower()
found_triggers = [trigger for trigger in self.hallucination_triggers if trigger in prompt_lower]
has_risk = len(found_triggers) > 0
return has_risk, found_triggers
def add_bias_mitigation(self, prompt: str) -> str:
"""Add bias mitigation instructions"""
mitigation = """
IMPORTANT - Fairness and Bias Considerations:
- Consider multiple perspectives and avoid stereotypes or generalizations about any group
- Ensure your response treats all individuals and groups fairly and respectfully
- Acknowledge the complexity and diversity of human experiences
- If discussing sensitive topics, be especially careful to avoid perpetuating biases
- Present balanced viewpoints when discussing controversial subjects
"""
return prompt + "\n" + mitigation
def add_hallucination_mitigation(self, prompt: str, triggers: List[str]) -> str:
"""Add hallucination mitigation instructions"""
mitigation_parts = []
if any('future' in t or 'predict' in t for t in triggers):
mitigation_parts.append("- Do not make specific predictions about future events. Discuss possibilities based on current trends and historical patterns, clearly marking these as speculative.")
if any('latest' in t or 'recent' in t or 'current' in t or 'today' in t for t in triggers):
mitigation_parts.append("- My knowledge has a cutoff date. Clearly state if information may be outdated and suggest authoritative sources for current information.")
if any('personal' in t or 'private' in t or 'confidential' in t for t in triggers):
mitigation_parts.append("- Do not provide or speculate about private, personal, or confidential information about individuals or organizations.")
mitigation_parts.append("- If uncertain about any information, explicitly state your uncertainty rather than guessing.")
mitigation_parts.append("- Distinguish clearly between facts, informed opinions, and speculation.")
mitigation = "\nIMPORTANT - Accuracy and Reliability:\n" + "\n".join(mitigation_parts)
return prompt + mitigation
# ============================================================================
# PROMPT RECONSTRUCTION
# ============================================================================
class PromptReconstructor:
"""Reconstructs optimized prompts from analysis and gathered information"""
def __init__(self, llm_interface: LLMInterface):
self.llm = llm_interface
def reconstruct(self, original_prompt: str, gathered_info: Dict[str, Any],
issues: List[AnalysisIssue]) -> Tuple[List[str], str]:
"""Reconstruct optimized prompt(s)"""
print("Reconstructing optimized prompt(s)...")
# Determine if prompt should be split
should_split = self._should_split_prompt(original_prompt, gathered_info)
if should_split:
prompts = self._create_prompt_chain(original_prompt, gathered_info)
explanation = self._generate_split_explanation(prompts)
return prompts, explanation
else:
optimized = self._create_single_prompt(original_prompt, gathered_info, issues)
explanation = self._generate_optimization_explanation(original_prompt, optimized, issues)
return [optimized], explanation
def _should_split_prompt(self, original_prompt: str, gathered_info: Dict[str, Any]) -> bool:
"""Determine if prompt should be split into multiple prompts"""
# Simple heuristics for splitting
word_count = len(original_prompt.split())
question_count = original_prompt.count('?')
# Split if very long with multiple questions
if word_count > 150 and question_count > 2:
return True
# Use LLM for more nuanced analysis
analysis_prompt = f"""Analyze whether this request should be split into multiple sequential prompts.
Original request: "{original_prompt}"
Additional context: {json.dumps(gathered_info, indent=2)}
A prompt should be split if:
1. It requests multiple distinct outputs or tasks that don't depend on each other
2. Later tasks explicitly depend on the results of earlier tasks
3. The scope is very broad and would benefit from focused sub-tasks
4. Different parts require fundamentally different approaches
Should this be split? Respond with JSON: {{"should_split": true/false, "reason": "brief explanation", "num_prompts": 2-4}}
"""
try:
response = self.llm.generate(analysis_prompt, temperature=0.2, max_tokens=400)
json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL)
if json_match:
try:
result = json.loads(json_match.group())
return result.get('should_split', False)
except json.JSONDecodeError:
pass
except Exception as e:
print(f" Split analysis failed: {e}")
return False
def _create_prompt_chain(self, original_prompt: str, gathered_info: Dict[str, Any]) -> List[str]:
"""Create a chain of sequential prompts"""
chain_prompt = f"""Break down this complex request into a logical sequence of 2-4 focused prompts.
Original request: "{original_prompt}"
Context: {json.dumps(gathered_info, indent=2)}
Create prompts that:
1. Each focus on a specific, well-defined sub-task
2. Build logically on previous results where appropriate
3. Together accomplish the original goal completely
4. Are clear, specific, and actionable
Format as a JSON array of prompt strings.
Example: ["First prompt focusing on X", "Second prompt building on X to address Y", ...]
"""
try:
response = self.llm.generate(chain_prompt, temperature=0.3, max_tokens=1200)
json_match = re.search(r'\[.*\]', response, re.DOTALL)
if json_match:
try:
prompts = json.loads(json_match.group())
# Enhance each prompt
enhanced = [self._enhance_prompt(p, gathered_info, i+1, len(prompts))
for i, p in enumerate(prompts)]
return enhanced
except json.JSONDecodeError:
pass
except Exception as e:
print(f" Prompt chain creation failed: {e}")
# Fallback to single optimized prompt
return [self._create_single_prompt(original_prompt, gathered_info, [])]
def _create_single_prompt(self, original_prompt: str, gathered_info: Dict[str, Any],
issues: List[AnalysisIssue]) -> str:
"""Create a single optimized prompt"""
components = []
# Role definition
if 'audience' in gathered_info:
audience = gathered_info['audience']
components.append(f"You are an expert assistant helping {audience}.")
elif 'goal' in gathered_info:
components.append("You are a knowledgeable and helpful assistant.")
# Context and background
context_parts = []
if 'context' in gathered_info:
context_parts.append(gathered_info['context'])
if 'background' in gathered_info:
context_parts.append(gathered_info['background'])
if context_parts:
components.append(f"CONTEXT: {' '.join(context_parts)}")
# Main task
task_parts = []
if 'goal' in gathered_info:
task_parts.append(f"GOAL: {gathered_info['goal']}")
task_parts.append(f"TASK: {original_prompt}")
components.append("\n".join(task_parts))
# Requirements and constraints
req_parts = []
if 'requirements' in gathered_info:
req_parts.append(f"Requirements: {gathered_info['requirements']}")
if 'constraints' in gathered_info:
req_parts.append(f"Constraints: {gathered_info['constraints']}")
if req_parts:
components.append("REQUIREMENTS:\n" + "\n".join(f"- {r}" for r in req_parts))
# Examples
if 'examples' in gathered_info:
components.append(f"EXAMPLES: {gathered_info['examples']}")
# Output format
if 'format' in gathered_info:
components.append(f"OUTPUT FORMAT: {gathered_info['format']}")
else:
components.append("OUTPUT FORMAT: Provide a clear, well-structured response with appropriate formatting.")
# Quality guidelines
quality_guidelines = [
"Be specific and concrete in your response",
"Use clear, professional language",
"Organize information logically"
]
if any(issue.category == 'specificity' for issue in issues):
quality_guidelines.append("Avoid vague terms; use specific quantities and criteria")
components.append("QUALITY GUIDELINES:\n" + "\n".join(f"- {g}" for g in quality_guidelines))
# Accuracy safeguards
components.append("\nIMPORTANT: Only provide information you are confident about. If uncertain about any aspect, clearly state your uncertainty rather than guessing or making assumptions.")
return "\n\n".join(components)
def _enhance_prompt(self, prompt: str, info: Dict[str, Any],
prompt_num: int, total_prompts: int) -> str:
"""Enhance a prompt in a chain"""
enhanced_parts = []
# Add sequence information
if total_prompts > 1:
enhanced_parts.append(f"[STEP {prompt_num} of {total_prompts}]")
# Add context if available
if 'context' in info and prompt_num == 1:
enhanced_parts.append(f"Context: {info['context']}")
# Main prompt
enhanced_parts.append(prompt)
# Add format if specified
if 'format' in info:
enhanced_parts.append(f"Format: {info['format']}")
# Add quality reminder
enhanced_parts.append("\nBe specific, accurate, and acknowledge any limitations or uncertainties.")
return "\n\n".join(enhanced_parts)
def _generate_split_explanation(self, prompts: List[str]) -> str:
"""Generate explanation for split prompts"""
explanation_parts = [
"I've split your request into multiple focused prompts for better results.",
"Execute these in sequence, using the output of each as context for the next:\n"
]
for i, prompt in enumerate(prompts, 1):
explanation_parts.append(f"{'='*70}")
explanation_parts.append(f"PROMPT {i} of {len(prompts)}")
explanation_parts.append(f"{'='*70}")
explanation_parts.append(prompt)
explanation_parts.append("")
return "\n".join(explanation_parts)
def _generate_optimization_explanation(self, original: str, optimized: str,
issues: List[AnalysisIssue]) -> str:
"""Generate explanation of optimizations"""
explanation_parts = [
"I've optimized your prompt with the following improvements:\n",
f"{'='*70}",
"ORIGINAL PROMPT",
f"{'='*70}",
original,
"",
f"{'='*70}",
"OPTIMIZED PROMPT",
f"{'='*70}",
optimized,
"",
f"{'='*70}",
"KEY ENHANCEMENTS",
f"{'='*70}"
]
enhancements = [
"Added clear structure with labeled sections (Context, Task, Requirements, etc.)",
"Made instructions more specific and actionable",
"Specified output format and quality expectations",
"Included safeguards against hallucination and inaccuracy"
]
if any(issue.category == 'context' for issue in issues):
enhancements.append("Added missing context and background information")
if any(issue.category == 'specificity' for issue in issues):
enhancements.append("Replaced vague terms with specific criteria")
if any(issue.category in ['bias_risk', 'fairness'] for issue in issues):
enhancements.append("Added fairness and bias mitigation guidelines")
for enhancement in enhancements:
explanation_parts.append(f"- {enhancement}")
return "\n".join(explanation_parts)
# ============================================================================
# MAIN CRITIQUE SYSTEM
# ============================================================================
class CritiqueSystem:
"""Main system integrating all components"""
def __init__(self, llm_config: Dict[str, Any]):
"""Initialize the Critique system"""
print("="*70)
print("CRITIQUE - LLM-Based Prompt Analyzer and Optimizer")
print("="*70)
print()
# Initialize hardware detection for local models
self.hardware_detector = None
if llm_config.get('mode') == 'local':
self.hardware_detector = HardwareDetector()
# Initialize LLM interface
print("Initializing LLM interface...")
if llm_config.get('mode') == 'local':
self.llm = LocalLLM(
llm_config['model_name'],
llm_config,
self.hardware_detector
)
else:
self.llm = RemoteLLM(
llm_config['model_name'],
llm_config
)
# Initialize components
self.analyzer = PromptAnalyzer(self.llm)
self.dialogue_manager = DialogueManager(self.llm, self.analyzer)
self.reconstructor = PromptReconstructor(self.llm)
self.mitigator = BiasHallucinationMitigator()
self.session_active = False
print("System initialized successfully!\n")
def start(self, user_prompt: str) -> str:
"""Start a new optimization session"""
if not self.llm.initialized:
self.llm.initialize()
print("\n" + "="*70)
print("STARTING NEW SESSION")
print("="*70 + "\n")
self.session_active = True
response = self.dialogue_manager.start_session(user_prompt)
# If no issues, proceed directly to reconstruction
if self.dialogue_manager.get_state() == DialogueState.COMPLETE:
return response + "\n\n" + self._perform_reconstruction()
return response
def continue_dialogue(self, user_response: str) -> str:
"""Continue the dialogue with user response"""
if not self.session_active:
return "No active session. Please start with a new prompt using start()."
response = self.dialogue_manager.process_user_response(user_response)
# Check if ready to reconstruct
if self.dialogue_manager.get_state() == DialogueState.CONFIRMING:
return response + "\n\n" + self._perform_reconstruction()
return response
def _perform_reconstruction(self) -> str:
"""Perform prompt reconstruction"""
print("\n" + "="*70)
print("RECONSTRUCTING OPTIMIZED PROMPT(S)")
print("="*70 + "\n")
original = self.dialogue_manager.original_prompt
info = self.dialogue_manager.get_gathered_info()
issues = self.dialogue_manager.issues
# Check for bias and hallucination risks
has_bias_risk, bias_topics = self.mitigator.check_bias_risk(original)
has_halluc_risk, halluc_triggers = self.mitigator.check_hallucination_risk(original)
if has_bias_risk:
print(f"Detected sensitivity to: {', '.join(bias_topics)}")
print("Adding bias mitigation guidelines...\n")
if has_halluc_risk:
print(f"Detected hallucination risks: {', '.join(halluc_triggers[:3])}")
print("Adding accuracy safeguards...\n")
# Reconstruct prompt(s)
prompts, explanation = self.reconstructor.reconstruct(original, info, issues)
# Apply final mitigations
final_prompts = []
for prompt in prompts:
if has_bias_risk:
prompt = self.mitigator.add_bias_mitigation(prompt)
if has_halluc_risk:
prompt = self.mitigator.add_hallucination_mitigation(prompt, halluc_triggers)
final_prompts.append(prompt)
# Format final output
result_parts = [
"="*70,
"OPTIMIZATION COMPLETE",
"="*70,
""
]
if len(final_prompts) > 1:
result_parts.append(f"Your request has been split into {len(final_prompts)} sequential prompts:\n")
for i, prompt in enumerate(final_prompts, 1):
result_parts.append("="*70)
result_parts.append(f"OPTIMIZED PROMPT {i} of {len(final_prompts)}")
result_parts.append("="*70)
result_parts.append(prompt)
result_parts.append("")
else:
result_parts.append("="*70)
result_parts.append("OPTIMIZED PROMPT")
result_parts.append("="*70)
result_parts.append(final_prompts[0])
result_parts.append("")
self.session_active = False
print("Session complete!\n")
return "\n".join(result_parts)
def shutdown(self):
"""Shutdown the system and cleanup resources"""
print("\nShutting down Critique system...")
if self.llm:
self.llm.cleanup()
print("Shutdown complete.")
# ============================================================================
# COMMAND LINE INTERFACE
# ============================================================================
def main():
"""Main entry point for command line usage"""
parser = argparse.ArgumentParser(
description='Critique: LLM-Based Prompt Analyzer and Optimizer',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Use local model (requires GPU)
python critique.py --mode local --model "meta-llama/Llama-2-7b-chat-hf"
# Use OpenAI API
python critique.py --mode remote --provider openai --model "gpt-4" --api-key YOUR_KEY
# Use Anthropic API
python critique.py --mode remote --provider anthropic --model "claude-3-opus-20240229" --api-key YOUR_KEY
"""
)
parser.add_argument('--mode', choices=['local', 'remote'], required=True,
help='Use local model or remote API')
parser.add_argument('--model', required=True,
help='Model name (HuggingFace model ID or API model name)')
parser.add_argument('--provider', choices=['openai', 'anthropic'],
help='API provider (required for remote mode)')
parser.add_argument('--api-key',
help='API key (required for remote mode)')
parser.add_argument('--api-base',
help='API base URL (optional, for custom endpoints)')
parser.add_argument('--no-quantization', action='store_true',
help='Disable quantization for local models')
parser.add_argument('--interactive', action='store_true',
help='Run in interactive mode')
args = parser.parse_args()
# Validate arguments
if args.mode == 'remote':
if not args.provider:
parser.error("--provider is required for remote mode")
if not args.api_key:
parser.error("--api-key is required for remote mode")
# Build configuration
llm_config = {
'mode': args.mode,
'model_name': args.model,
'use_quantization': not args.no_quantization
}
if args.mode == 'remote':
llm_config['provider'] = args.provider
llm_config['api_key'] = args.api_key
if args.api_base:
llm_config['api_base'] = args.api_base
# Initialize system
try:
critique = CritiqueSystem(llm_config)
except Exception as e:
print(f"Failed to initialize Critique system: {e}")
return 1
# Interactive mode
if args.interactive:
print("\nEntering interactive mode. Type 'quit' to exit.\n")
while True:
print("="*70)
user_prompt = input("Enter your prompt (or 'quit' to exit):\n> ")
if user_prompt.lower() in ['quit', 'exit', 'q']:
break
if not user_prompt.strip():
continue
try:
# Start session
response = critique.start(user_prompt)
print("\n" + response + "\n")
# Continue dialogue if needed
while critique.session_active:
user_input = input("\nYour response:\n> ")
if user_input.lower() in ['skip', 's']:
print("\nSkipping to optimization...\n")
critique.dialogue_manager.state = DialogueState.CONFIRMING
response = critique._perform_reconstruction()
print(response)
break
response = critique.continue_dialogue(user_input)
print("\n" + response + "\n")
except Exception as e:
print(f"\nError during processing: {e}\n")
critique.session_active = False
else:
# Single prompt mode - read from stdin
print("\nEnter your prompt (Ctrl+D when done):")
user_prompt = sys.stdin.read().strip()
if not user_prompt:
print("No prompt provided.")
return 1
try:
response = critique.start(user_prompt)
print("\n" + response + "\n")
# If dialogue needed, prompt for responses
while critique.session_active:
print("\nEnter your response (Ctrl+D when done, or type 'skip' to proceed):")
user_input = sys.stdin.read().strip()
if user_input.lower() in ['skip', 's']:
critique.dialogue_manager.state = DialogueState.CONFIRMING
response = critique._perform_reconstruction()
print(response)
break
response = critique.continue_dialogue(user_input)
print("\n" + response + "\n")
except Exception as e:
print(f"\nError during processing: {e}")
return 1
# Cleanup
critique.shutdown()
return 0
if __name__ == '__main__':
sys.exit(main())
