AI 学术污染治理

# 检测问卷回复中可能的 LLM 生成文本
import openai
from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np

def detect_llm_patterns(responses):
    """检测回复中的 LLM 典型模式"""
    patterns = {
        "repetitive_structure": [],
        "overly_polite": [],
        "hedging_language": [],
        "perfect_grammar": [],
    }
    
    hedging_phrases = [
        "it's important to note", "it depends", "generally speaking",
        "in many cases", "on the other hand", "however", "moreover"
    ]
    
    results = []
    for i, resp in enumerate(responses):
        score = 0
        for phrase in hedging_phrases:
            if phrase.lower() in resp.lower():
                score += 1
        if abs(len(resp) - np.mean([len(r) for r in responses])) < 50:
            score += 1
        results.append({
            "response_id": i,
            "llm_probability": min(score / 5, 1.0),
            "flagged": score >= 3
        })
    
    flagged_count = sum(1 for r in results if r["flagged"])
    print(f"检测到 {flagged_count}/{len(results)} 条回复疑似 LLM 生成")
    return results

# 检测文献引用中的幻觉引用
import requests
from concurrent.futures import ThreadPoolExecutor

def verify_citation(title, doi=None):
    """验证文献引用是否真实存在"""
    if doi:
        url = f"https://api.crossref.org/works/{doi}"
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            return {"valid": True, "doi": doi}
        return {"valid": False, "reason": "DOI not found"}
    
    url = f"https://api.crossref.org/works?query.title={title}&rows=1"
    response = requests.get(url, timeout=10)
    if response.status_code == 200:
        data = response.json()
        if data["message"]["total-results"] > 0:
            return {"valid": True, "matched": data["message"]["items"][0]["title"]}
    return {"valid": False, "reason": "Title not found"}

citations = [
    {"title": "Large Language Models and Social Science", "doi": "10.1038/s41586-024-xxxxx"},
    {"title": "AI Bias in Survey Responses", "doi": None},
]

for cite in citations:
    result = verify_citation(cite["title"], cite.get("doi"))
    print(f"{cite['title']}: {'✅' if result['valid'] else '❌'} - {result.get('reason', '')}

# 语义多样性分析：计算回复的词汇丰富度
from collections import Counter
import math

def lexical_diversity(text):
    """计算文本的词法多样性（Type-Token Ratio）"""
    words = text.lower().split()
    if not words:
        return 0
    unique_words = len(set(words))
    total_words = len(words)
    return unique_words / total_words

def shannon_entropy(text):
    """计算文本的信息熵"""
    words = text.lower().split()
    if not words:
        return 0
    freq = Counter(words)
    total = len(words)
    entropy = -sum((count/total) * math.log2(count/total) 
                   for count in freq.values())
    return entropy

real_responses = ["问卷回复 1", "问卷回复 2"]
llm_responses = ["LLM 生成 1", "LLM 生成 2"]

for name, responses in [("真实", real_responses), ("LLM", llm_responses)]:
    diversities = [lexical_diversity(r) for r in responses]
    entropies = [shannon_entropy(r) for r in responses]
    print(f"{name}回复 - 平均词法多样性: {sum(diversities)/len(diversities):.3f}")
    print(f"{name}回复 - 平均信息熵: {sum(entropies)/len(entropies):.3f}")

# 学术论文 AI 检测的多维度评分框架
class AIDetectionScorer:
    def __init__(self, weights=None):
        self.weights = weights or {
            "perplexity": 0.25,
            "burstiness": 0.20,
            "citation_validity": 0.20,
            "structural_pattern": 0.15,
            "semantic_coherence": 0.20,
        }
    
    def calculate_perplexity_score(self, text):
        """计算文本的困惑度"""
        import torch
        from transformers import AutoModelForCausalLM, AutoTokenizer
        model = AutoModelForCausalLM.from_pretrained("gpt2")
        tokenizer = AutoTokenizer.from_pretrained("gpt2")
        encoded = tokenizer(text, return_tensors="pt")
        with torch.no_grad():
            outputs = model(**encoded, labels=encoded["input_ids"])
        perplexity = torch.exp(outputs.loss).item()
        return max(0, 1 - (perplexity / 100))
    
    def score(self, text, citations=None):
        scores = {
            "perplexity": self.calculate_perplexity_score(text),
            "burstiness": self._calculate_burstiness(text),
            "citation_validity": self._verify_citations(citations or []),
            "structural_pattern": self._check_structure(text),
            "semantic_coherence": self._semantic_coherence(text),
        }
        total = sum(self.weights[k] * scores[k] for k in scores)
        return {"overall_ai_probability": total, "breakdown": scores}

# 构建文献污染传播网络
import networkx as nx

class CitationPollutionTracker:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.polluted_papers = set()
    
    def add_paper(self, paper_id, title, citations, is_polluted=False):
        self.graph.add_node(paper_id, title=title, polluted=is_polluted)
        for cite in citations:
            self.graph.add_edge(cite, paper_id)
        if is_polluted:
            self.polluted_papers.add(paper_id)
    
    def find_pollution_cascade(self):
        affected = set()
        for polluted in self.polluted_papers:
            for successor in nx.descendants(self.graph, polluted):
                affected.add(successor)
        return affected
    
    def get_pollution_depth(self, paper_id):
        min_depth = float('inf')
        for polluted in self.polluted_papers:
            try:
                depth = nx.shortest_path_length(self.graph, polluted, paper_id)
                min_depth = min(min_depth, depth)
            except nx.NetworkXNoPath:
                pass
        return min_depth if min_depth != float('inf') else -1

# 多层次 AI 学术污染检测管道
class MultiLayerPollutionDetector:
    def __init__(self):
        self.layers = {
            "text_analysis": TextAnalysisLayer(),
            "citation_verification": CitationVerificationLayer(),
            "data_integrity": DataIntegrityLayer(),
            "methodology_check": MethodologyCheckLayer(),
        }
    
    def evaluate_paper(self, paper):
        results = {}
        for layer_name, layer in self.layers.items():
            results[layer_name] = layer.analyze(paper)
        risk_score = self._aggregate_risk(results)
        return {
            "overall_risk": risk_score,
            "layer_results": results,
            "recommendation": self._recommendation(risk_score),
        }
    
    def _aggregate_risk(self, results):
        weights = {
            "text_analysis": 0.3,
            "citation_verification": 0.25,
            "data_integrity": 0.25,
            "methodology_check": 0.2,
        }
        return sum(weights[k] * results[k]["risk"] for k in weights)
    
    def _recommendation(self, risk_score):
        if risk_score > 0.7:
            return "🔴 高风险 - 建议深入调查"
        elif risk_score > 0.4:
            return "🟡 中等风险 - 建议人工审查"
        else:
            return "🟢 低风险 - 通过常规流程"

# AI 模拟 vs 真实人类数据的偏差分析
import numpy as np
from scipy import stats

def compare_ai_vs_human(ai_responses, human_responses):
    ks_stat, ks_pvalue = stats.ks_2samp(ai_responses, human_responses)
    cohen_d = (np.mean(ai_responses) - np.mean(human_responses)) / \
              np.sqrt((np.std(ai_responses)**2 + np.std(human_responses)**2) / 2)
    
    print(f"KS 检验: 统计量={ks_stat:.3f}, p值={ks_pvalue:.4f}")
    print(f"效应量 (Cohen's d): {cohen_d:.3f}")
    
    if ks_pvalue < 0.05:
        print("⚠️ AI 模拟与真实人类数据存在显著差异")
    if abs(cohen_d) > 0.8:
        print("🔴 大效应量——不可直接替代人类数据")
    return {"ks_stat": ks_stat, "p_value": ks_pvalue, "cohen_d": cohen_d}

# 中文 AI 文本检测的差异化策略
class ChineseAIDetector:
    """针对中文 LLM 生成内容的检测工具"""
    
    def __init__(self):
        self.chinese_features = {
            "idiom_usage": 0.15,       # 成语使用频率
            "classical_reference": 0.10, # 古典引用
            "sentence_rhythm": 0.15,    # 句式节奏
            "colloquialism": 0.10,      # 口语化表达
            "regional_dialect": 0.10,   # 方言特征
            "emotional_authenticity": 0.15,  # 情感真实性
            "logical_coherence": 0.15,  # 逻辑连贯性
            "citation_pattern": 0.10,   # 引用模式
        }
    
    def analyze(self, text):
        """分析中文文本的 AI 生成概率"""
        scores = {}
        for feature, weight in self.chinese_features.items():
            scores[feature] = self._evaluate_feature(text, feature)
        
        # 加权综合
        total = sum(scores[f] * self.chinese_features[f] for f in scores)
        return {
            "ai_probability": total,
            "feature_scores": scores,
        }
    
    def _evaluate_feature(self, text, feature):
        # 根据不同特征进行评估
        if feature == "idiom_usage":
            return self._check_idiom_naturalness(text)
        elif feature == "sentence_rhythm":
            return self._check_sentence_rhythm(text)
        return 0.5