كشف سلوك الذكاء الاصطناعي غير المرغوب فيه: التقنيات والعلامات الحمراء

from transformers import GPT2LMHeadModel, GPT2Tokenizer
import torch
import numpy as np

def calculate_perplexity(text, model_name='gpt2'):
    """Calculate perplexity of text using a reference model"""
    tokenizer = GPT2Tokenizer.from_pretrained(model_name)
    model = GPT2LMHeadModel.from_pretrained(model_name)
    
    encodings = tokenizer(text, return_tensors='pt')
    max_length = model.config.n_positions
    stride = 512
    
    nlls = []
    for i in range(0, encodings.input_ids.size(1), stride):
        begin_loc = max(i + stride - max_length, 0)
        end_loc = min(i + stride, encodings.input_ids.size(1))
        trg_len = end_loc - i
        
        input_ids = encodings.input_ids[:, begin_loc:end_loc]
        target_ids = input_ids.clone()
        target_ids[:, :-trg_len] = -100
        
        with torch.no_grad():
            outputs = model(input_ids, labels=target_ids)
            neg_log_likelihood = outputs.loss * trg_len
        
        nlls.append(neg_log_likelihood)
    
    ppl = torch.exp(torch.stack(nlls).sum() / end_loc)
    return ppl.item()

# Usage
text_sample = "Your text to analyze goes here..."
perplexity_score = calculate_perplexity(text_sample)
print(f"Perplexity: {perplexity_score:.2f}")

# Lower perplexity (< 50) suggests AI generation
# Higher perplexity (> 100) suggests human writing

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

def classify_text(text, model_name='roberta-base-openai-detector'):
    """
    Classify text as human or AI-generated using a fine-tuned transformer.
    Note: Replace model_name with actual detector model from HuggingFace
    """
    tokenizer = AutoTokenizer.from_pretrained('roberta-base')
    # In practice, use a model specifically trained for detection
    model = AutoModelForSequenceClassification.from_pretrained('roberta-base', num_labels=2)
    
    inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=512)
    
    with torch.no_grad():
        outputs = model(**inputs)
        predictions = torch.softmax(outputs.logits, dim=1)
    
    ai_probability = predictions[0][1].item()
    human_probability = predictions[0][0].item()
    
    return {
        'ai_probability': ai_probability,
        'human_probability': human_probability,
        'predicted_class': 'AI' if ai_probability > 0.5 else 'Human',
        'confidence': max(ai_probability, human_probability)
    }

# Example usage
text = "The implementation of advanced algorithms..."
result = classify_text(text)
print(f"Prediction: {result['predicted_class']} (confidence: {result['confidence']:.2%})")

import hashlib
import numpy as np

class SimpleWatermarkDetector:
    """
    Simplified watermark detector based on vocabulary partitioning.
    Production systems would use more sophisticated approaches.
    """
    def __init__(self, key: str, vocab_size: int = 50000, green_fraction: float = 0.5):
        self.key = key
        self.vocab_size = vocab_size
        self.green_fraction = green_fraction
    
    def _get_green_list(self, prefix: str) -> set:
        """Generate green list based on prefix and secret key"""
        hash_input = f"{self.key}:{prefix}".encode()
        hash_output = hashlib.sha256(hash_input).digest()
        
        # Use hash to seed RNG for deterministic green list
        rng = np.random.RandomState(int.from_bytes(hash_output[:4], 'big'))
        green_size = int(self.vocab_size * self.green_fraction)
        green_tokens = set(rng.choice(self.vocab_size, green_size, replace=False))
        
        return green_tokens
    
    def score_text(self, tokens: list) -> dict:
        """Calculate watermark score for token sequence"""
        green_count = 0
        
        for i, token in enumerate(tokens):
            prefix = "".join(map(str, tokens[:i])) if i > 0 else ""
            green_list = self._get_green_list(prefix)
            
            if token in green_list:
                green_count += 1
        
        n = len(tokens)
        expected_green = n * self.green_fraction
        std_dev = np.sqrt(n * self.green_fraction * (1 - self.green_fraction))
        z_score = (green_count - expected_green) / std_dev if std_dev > 0 else 0
        
        return {
            'green_count': green_count,
            'total_tokens': n,
            'z_score': z_score,
            'is_watermarked': z_score > 2.0,  # Threshold for detection
            'p_value': 1 - 0.5 * (1 + np.tanh(z_score / np.sqrt(2)))
        }

# Example usage
detector = SimpleWatermarkDetector(key="secret_key_123")
token_sequence = [1234, 5678, 9012, 3456, 7890]  # Example token IDs
result = detector.score_text(token_sequence)
print(f"Watermarked: {result['is_watermarked']} (z-score: {result['z_score']:.2f})")

import re
from collections import Counter

def analyze_ai_patterns(text: str) -> dict:
    """Detect linguistic patterns common in AI-generated text"""
    
    # Common AI hedge phrases
    hedge_phrases = [
        r'\bit[\'']s worth noting',
        r'\bit[\'']s important to',
        r'\barguably\b',
        r'\bto some extent\b',
        r'\bin many ways\b',
        r'\bit depends\b',
        r'\bvarious factors\b',
        r'\bwhile .*? may vary\b',
    ]
    
    # Analyze sentence starts
    sentences = re.split(r'[.!?]+', text)
    sentence_starts = [s.strip().split()[:3] for s in sentences if s.strip()]
    start_patterns = Counter([' '.join(start[:2]) for start in sentence_starts if len(start) >= 2])
    
    # Count hedging
    hedge_count = sum(len(re.findall(pattern, text, re.IGNORECASE)) for pattern in hedge_phrases)
    
    # Check for list formatting
    has_numbered_lists = bool(re.search(r'\n\d+\.', text))
    has_bullet_points = bool(re.search(r'\n[\-\*]', text))
    
    # Calculate metrics
    word_count = len(text.split())
    hedge_density = hedge_count / (word_count / 100) if word_count > 0 else 0
    
    # Detect repetitive starts
    max_repeat_ratio = max(count / len(sentence_starts) 
                          for count in start_patterns.values()) if sentence_starts else 0
    
    return {
        'hedge_density': hedge_density,  # Hedges per 100 words
        'max_repeat_ratio': max_repeat_ratio,  # Most common sentence start ratio
        'has_list_formatting': has_numbered_lists or has_bullet_points,
        'sentence_count': len([s for s in sentences if s.strip()]),
        'suspicion_score': (hedge_density * 0.4 + max_repeat_ratio * 60),
        'top_repeated_starts': start_patterns.most_common(3)
    }

# Example
text_sample = """
It's worth noting that AI detection is complex. It's important to remember that 
multiple methods work best. Arguably, no single approach is perfect.
"""

analysis = analyze_ai_patterns(text_sample)
print(f"Suspicion Score: {analysis['suspicion_score']:.1f}")
print(f"Hedge Density: {analysis['hedge_density']:.2f} per 100 words")

import requests
import os

class ContentDetectionPipeline:
    """Integrate multiple detection services for robust checking"""
    
    def __init__(self, api_keys: dict):
        self.api_keys = api_keys
        self.results_cache = {}
    
    def check_gptzero(self, text: str) -> dict:
        """Check using GPTZero API"""
        url = "https://api.gptzero.me/v2/predict/text"
        headers = {
            "Authorization": f"Bearer {self.api_keys.get('gptzero')}",
            "Content-Type": "application/json"
        }
        data = {"document": text}
        
        try:
            response = requests.post(url, json=data, headers=headers)
            response.raise_for_status()
            result = response.json()
            
            return {
                'service': 'gptzero',
                'ai_probability': result.get('documents', [{}])[0].get('completely_generated_prob', 0),
                'confidence': result.get('documents', [{}])[0].get('average_generated_prob', 0)
            }
        except Exception as e:
            return {'service': 'gptzero', 'error': str(e)}
    
    def check_originality(self, text: str) -> dict:
        """Check using Originality.ai API"""
        url = "https://api.originality.ai/api/v1/scan/ai"
        headers = {"X-OAI-API-KEY": self.api.keys.get('originality')}
        data = {"content": text}
        
        try:
            response = requests.post(url, data=data, headers=headers)
            response.raise_for_status()
            result = response.json()
            
            return {
                'service': 'originality',
                'ai_probability': result.get('score', {}).get('ai', 0),
                'human_probability': result.get('score', {}).get('original', 0)
            }
        except Exception as e:
            return {'service': 'originality', 'error': str(e)}
    
    def aggregate_results(self, results: list) -> dict:
        """Combine multiple detection results"""
        valid_results = [r for r in results if 'error' not in r]
        
        if not valid_results:
            return {'error': 'All services failed', 'verdict': 'UNKNOWN'}
        
        avg_ai_prob = sum(r.get('ai_probability', 0) for r in valid_results) / len(valid_results)
        
        return {
            'ai_probability': avg_ai_prob,
            'verdict': 'AI' if avg_ai_prob > 0.7 else 'HUMAN' if avg_ai_prob < 0.3 else 'UNCERTAIN',
            'confidence': abs(avg_ai_prob - 0.5) * 2,  # Distance from uncertain
            'individual_results': results
        }
    
    def analyze(self, text: str) -> dict:
        """Run full detection pipeline"""
        results = []
        
        # Check with available services
        if self.api_keys.get('gptzero'):
            results.append(self.check_gptzero(text))
        
        if self.api_keys.get('originality'):
            results.append(self.check_originality(text))
        
        return self.aggregate_results(results)

# Usage example
pipeline = ContentDetectionPipeline({
    'gptzero': os.getenv('GPTZERO_API_KEY'),
    'originality': os.getenv('ORIGINALITY_API_KEY')
})

content = "Your content to check goes here..."
result = pipeline.analyze(content)
print(f"Verdict: {result['verdict']} (confidence: {result['confidence']:.2%})")

import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import numpy as np

class SelfHostedDetector:
    """Self-hosted detection combining multiple approaches"""
    
    def __init__(self, model_path: str = 'roberta-base'):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_path, 
            num_labels=2
        )
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.model.to(self.device)
        self.model.eval()
    
    def detect_with_classifier(self, text: str) -> dict:
        """Use transformer classifier"""
        inputs = self.tokenizer(
            text, 
            return_tensors='pt', 
            truncation=True, 
            max_length=512,
            padding=True
        ).to(self.device)
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=1)
        
        return {
            'ai_score': probs[0][1].item(),
            'human_score': probs[0][0].item()
        }
    
    def detect_with_perplexity(self, text: str) -> dict:
        """Use perplexity-based detection"""
        from transformers import GPT2LMHeadModel, GPT2Tokenizer
        
        gpt2_tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
        gpt2_model = GPT2LMHeadModel.from_pretrained('gpt2').to(self.device)
        
        encodings = gpt2_tokenizer(text, return_tensors='pt').to(self.device)
        
        with torch.no_grad():
            outputs = gpt2_model(**encodings, labels=encodings['input_ids'])
            perplexity = torch.exp(outputs.loss).item()
        
        # Lower perplexity suggests AI generation
        ai_score = 1 / (1 + np.exp((perplexity - 50) / 20))  # Sigmoid normalization
        
        return {
            'perplexity': perplexity,
            'ai_score': ai_score
        }
    
    def detect_with_patterns(self, text: str) -> dict:
        """Use linguistic pattern detection"""
        analysis = analyze_ai_patterns(text)  # From previous section
        
        # Normalize suspicion score to 0-1 range
        ai_score = min(analysis['suspicion_score'] / 100, 1.0)
        
        return {
            'pattern_ai_score': ai_score,
            'details': analysis
        }
    
    def detect(self, text: str, methods: list = None) -> dict:
        """Run detection with specified or all methods"""
        if methods is None:
            methods = ['classifier', 'perplexity', 'patterns']
        
        results = {}
        
        if 'classifier' in methods:
            results['classifier'] = self.detect_with_classifier(text)
        
        if 'perplexity' in methods:
            results['perplexity'] = self.detect_with_perplexity(text)
        
        if 'patterns' in methods:
            results['patterns'] = self.detect_with_patterns(text)
        
        # Aggregate scores
        ai_scores = []
        if 'classifier' in results:
            ai_scores.append(results['classifier']['ai_score'])
        if 'perplexity' in results:
            ai_scores.append(results['perplexity']['ai_score'])
        if 'patterns' in results:
            ai_scores.append(results['patterns']['pattern_ai_score'])
        
        final_score = np.mean(ai_scores) if ai_scores else 0
        
        return {
            'final_ai_score': final_score,
            'verdict': 'AI' if final_score > 0.7 else 'HUMAN' if final_score < 0.3 else 'UNCERTAIN',
            'confidence': abs(final_score - 0.5) * 2,
            'method_results': results
        }

# Usage
detector = SelfHostedDetector()
text = "Your content to analyze..."
result = detector.detect(text)
print(f"Verdict: {result['verdict']} ({result['final_ai_score']:.2%} AI probability)")

class RobustDetectionSystem:
    """Production-ready detection with fallbacks and human review queue"""
    
    def __init__(self, confidence_threshold: float = 0.8):
        self.detector = SelfHostedDetector()
        self.confidence_threshold = confidence_threshold
        self.review_queue = []
    
    def classify_with_context(self, text: str, metadata: dict = None) -> dict:
        """Classify considering text and contextual signals"""
        
        # Primary detection
        detection_result = self.detector.detect(text)
        
        # Add context scoring
        context_signals = self._analyze_context(metadata or {})
        
        # Combine text and context
        combined_score = (
            detection_result['final_ai_score'] * 0.7 + 
            context_signals['suspicion_score'] * 0.3
        )
        
        confidence = detection_result['confidence']
        
        # Route based on confidence
        if confidence < self.confidence_threshold:
            self._add_to_review_queue(text, detection_result, metadata)
            verdict = 'NEEDS_REVIEW'
        else:
            verdict = 'AI' if combined_score > 0.7 else 'HUMAN'
        
        return {
            'verdict': verdict,
            'ai_score': combined_score,
            'confidence': confidence,
            'needs_review': confidence < self.confidence_threshold,
            'detection_details': detection_result,
            'context_signals': context_signals
        }
    
    def _analyze_context(self, metadata: dict) -> dict:
        """Analyze non-textual signals"""
        suspicion_factors = []
        
        # Check posting velocity
        if metadata.get('posts_last_hour', 0) > 10:
            suspicion_factors.append(0.3)
        
        # Check account age vs. content volume
        if metadata.get('account_age_days', 365) < 7 and metadata.get('total_posts', 0) > 50:
            suspicion_factors.append(0.4)
        
        # Check response time (very fast responses suspicious)
        if metadata.get('response_time_seconds', 60) < 10:
            suspicion_factors.append(0.2)
        
        suspicion_score = min(sum(suspicion_factors), 1.0) if suspicion_factors else 0
        
        return {
            'suspicion_score': suspicion_score,
            'factors': suspicion_factors
        }
    
    def _add_to_review_queue(self, text: str, result: dict, metadata: dict):
        """Add borderline cases to human review"""
        self.review_queue.append({
            'text': text[:500],  # Preview only
            'detection_result': result,
            'metadata': metadata,
            'timestamp': __import__('datetime').datetime.now().isoformat()
        })
    
    def get_review_queue(self, limit: int = 10) -> list:
        """Get items needing human review"""
        return self.review_queue[:limit]

# Usage
system = RobustDetectionSystem(confidence_threshold=0.75)

result = system.classify_with_context(
    text="Content to check...",
    metadata={
        'account_age_days': 5,
        'posts_last_hour': 15,
        'response_time_seconds': 8
    }
)

print(f"Verdict: {result['verdict']}")
if result['needs_review']:
    print("Flagged for human review")