Add kn-style calibration workflow

README.md

@@ -12,4 +12,49 @@ license: mit
 short_description: 'an AIGC_detector by model: yuchuantian/AIGC_detector_zhv2'
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# AIGC Detector (论文AIGC风险 + 知网对齐预测)
+
+这是一个可直接部署到 Hugging Face Spaces 的 Gradio 项目。
+
+## 功能
+
+- 段落级 AIGC 风险分析
+- 综合 AI 风险率
+- 预测知网 AIGC 率（校准模式）
+- 无校准文件时自动回退到原始风险率模式
+
+## Space 直接部署
+
+1. 上传本仓库到 Hugging Face Space（SDK 选 Gradio）。
+2. Space 会自动读取 `requirements.txt` 安装依赖。
+3. 默认即可运行，不需要额外环境变量。
+
+## 知网对齐校准（推荐）
+
+`calibration/model.json` 是校准模型文件。
+
+- 默认提供的是透传占位模型（预测率=综合风险率）。
+- 你可以用真实数据重新训练并覆盖它。
+
+### 训练数据格式（CSV）
+
+需要列：
+
+- `overall`
+- `p90`
+- `high_ratio`
+- `mid_ratio`
+- `std`
+- `kn_rate`（知网AIGC率，0~1）
+
+### 训练命令
+
+```bash
+python3 scripts/train_calibration.py --input your_dataset.csv --target kn_rate --out calibration/model.json
+```
+
+训练完成后，把新的 `calibration/model.json` 提交到 Space，即可自动启用“知网对齐预测率”。
+
+## 说明
+
+本项目输出用于研究与预筛查，不代表任何官方系统结论。

app.py

@@ -1,104 +1,203 @@
-import gradio as gr
-import fitz
-import torch
-import numpy as np
+import json
+import os
 import re
+from pathlib import Path
+from typing import Dict, List
 
-from transformers import (
-    AutoTokenizer,
-    AutoModelForSequenceClassification
-)
+import fitz
+import gradio as gr
+import numpy as np
+import torch
+from transformers import AutoModelForSequenceClassification, AutoTokenizer
 
 MODEL_NAME = "yuchuantian/AIGC_detector_zhv2"
 
+MAX_PAGES = 120
+MAX_PARAGRAPHS = 120
+MIN_PARAGRAPH_CHARS = 80
+WINDOW_MAX_LENGTH = 512
+WINDOW_STRIDE = 128
+
+CALIBRATION_PATH = Path("calibration/model.json")
+
+
 tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
 model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
+model.eval()
+
+
+def load_calibration_model() -> Dict:
+    if not CALIBRATION_PATH.exists():
+        return {}
+
+    try:
+        data = json.loads(CALIBRATION_PATH.read_text(encoding="utf-8"))
+    except Exception:
+        return {}
+
+    if data.get("model_type") != "linear":
+        return {}
 
-def split_paragraphs(text):
-    paragraphs = re.split(r'\n+', text)
+    required = {"feature_order", "coef", "intercept"}
+    if not required.issubset(data.keys()):
+        return {}
+
+    return data
+
+
+CALIBRATION_MODEL = load_calibration_model()
+
+
+def normalize_text(text: str) -> str:
+    text = re.sub(r"\s+", " ", text)
+    return text.strip()
+
+
+def split_paragraphs(text: str) -> List[str]:
+    paragraphs = re.split(r"\n{2,}", text)
+    return [p.strip() for p in paragraphs if len(normalize_text(p)) >= MIN_PARAGRAPH_CHARS]
+
+
+def should_skip_paragraph(text: str) -> bool:
+    t = normalize_text(text)
+    if not t:
+        return True
+
+    if re.search(r"(参考文献|致谢|附录|作者简介)", t[:40], flags=re.IGNORECASE):
+        return True
+
+    cn_chars = len(re.findall(r"[\u4e00-\u9fff]", t))
+    digit_punc = len(re.findall(r"[\d\W_]", t))
+    if cn_chars < 20 or digit_punc > len(t) * 0.75:
+        return True
+
+    return False
+
+
+def extract_pdf_text(pdf_file) -> str:
+    doc = fitz.open(pdf_file.name)
+    chunks = []
 
-    return [
-        p.strip()
-        for p in paragraphs
-        if len(p.strip()) > 50
-    ]
+    for page_idx, page in enumerate(doc):
+        if page_idx >= MAX_PAGES:
+            break
 
-def calc_repetition(text):
-    words = text.split()
+        blocks = page.get_text("blocks")
+        blocks_sorted = sorted(blocks, key=lambda b: (round(b[1], 1), round(b[0], 1)))
+        page_text = "\n".join(b[4].strip() for b in blocks_sorted if b[4].strip())
 
-    if len(words) == 0:
-        return 0
+        if page_text:
+            chunks.append(page_text)
 
-    unique_ratio = len(set(words)) / len(words)
+    return "\n\n".join(chunks)
 
-    return 1 - unique_ratio
 
-def calc_sentence_variance(text):
-    sents = re.split(r'[。！？]', text)
+def calc_repetition(text: str) -> float:
+    t = normalize_text(text)
+    if not t:
+        return 0.0
 
+    grams = [t[i : i + 2] for i in range(len(t) - 1)]
+    if not grams:
+        return 0.0
+
+    unique_ratio = len(set(grams)) / len(grams)
+    return max(0.0, 1.0 - unique_ratio)
+
+
+def calc_sentence_variance(text: str) -> float:
+    sents = re.split(r"[。！？!?]", text)
     lens = [len(s.strip()) for s in sents if len(s.strip()) > 0]
 
     if len(lens) < 2:
-        return 0
+        return 0.0
+
+    return float(min(np.var(lens) / 900.0, 1.0))
 
-    return np.var(lens) / 1000
 
-def detector_score(text):
+def detector_score(text: str) -> float:
     inputs = tokenizer(
         text,
-        return_tensors="pt",
         truncation=True,
-        max_length=512
+        max_length=WINDOW_MAX_LENGTH,
+        stride=WINDOW_STRIDE,
+        return_overflowing_tokens=True,
+        padding=True,
+        return_tensors="pt",
     )
 
     with torch.no_grad():
         outputs = model(**inputs)
 
     probs = torch.softmax(outputs.logits, dim=-1)
+    ai_probs = probs[:, 1].cpu().numpy()
 
-    return probs[0][1].item()
+    return float(0.75 * np.mean(ai_probs) + 0.25 * np.max(ai_probs))
 
-def analyze_paragraph(text):
-    detector = detector_score(text)
 
+def analyze_paragraph(text: str):
+    detector = detector_score(text)
     repetition = calc_repetition(text)
-
     variance = calc_sentence_variance(text)
 
-    risk = (
-        detector * 0.7
-        + repetition * 0.15
-        + (1 - min(variance, 1)) * 0.15
-    )
+    stable_style = 1 - variance
+    risk = detector * 0.78 + repetition * 0.12 + stable_style * 0.10
+    risk = float(min(max(risk, 0.0), 1.0))
 
     return {
         "detector": detector,
         "repetition": repetition,
         "variance": variance,
-        "risk": risk
+        "risk": risk,
     }
 
-def analyze_pdf(pdf_file):
-    doc = fitz.open(pdf_file.name)
 
-    text = ""
+def clip01(v: float) -> float:
+    return float(min(max(v, 0.0), 1.0))
+
+
+def build_doc_features(risks: List[float]) -> Dict[str, float]:
+    arr = np.array(risks, dtype=float)
+    return {
+        "overall": clip01(float(np.mean(arr))),
+        "p90": clip01(float(np.percentile(arr, 90))),
+        "high_ratio": clip01(float(np.mean(arr > 0.75))),
+        "mid_ratio": clip01(float(np.mean((arr > 0.55) & (arr <= 0.75)))),
+        "std": clip01(float(np.std(arr))),
+    }
+
+
+def predict_kn_like_rate(features: Dict[str, float]) -> float:
+    if not CALIBRATION_MODEL:
+        return features["overall"]
+
+    order = CALIBRATION_MODEL["feature_order"]
+    coef = CALIBRATION_MODEL["coef"]
+    intercept = float(CALIBRATION_MODEL["intercept"])
 
-    for page in doc:
-        text += page.get_text()
+    x = np.array([features.get(name, 0.0) for name in order], dtype=float)
+    y = float(np.dot(x, np.array(coef, dtype=float)) + intercept)
 
+    return clip01(y)
+
+
+def analyze_pdf(pdf_file):
+    text = extract_pdf_text(pdf_file)
     paragraphs = split_paragraphs(text)
+    paragraphs = [p for p in paragraphs if not should_skip_paragraph(p)]
+    paragraphs = paragraphs[:MAX_PARAGRAPHS]
 
-    results = []
+    if not paragraphs:
+        return "未提取到可分析正文。请尝试文本层可复制的 PDF，或调整排版后再试。"
 
+    results = []
     risks = []
 
-    for p in paragraphs[:80]:
+    for p in paragraphs:
         score = analyze_paragraph(p)
-
         risks.append(score["risk"])
 
         level = "🟢"
-
         if score["risk"] > 0.75:
             level = "🔴"
         elif score["risk"] > 0.55:
@@ -110,29 +209,45 @@ def analyze_pdf(pdf_file):
 
 Detector: {score['detector']:.2%}
 重复度: {score['repetition']:.2%}
-句式稳定性: {1-score['variance']:.2%}
+句式稳定性: {1 - score['variance']:.2%}
 
 {p[:400]}
 """
         )
 
-    overall = np.mean(risks)
+    features = build_doc_features(risks)
+    overall = features["overall"]
+    high_ratio = features["high_ratio"]
+    mid_ratio = features["mid_ratio"]
+    kn_like_rate = predict_kn_like_rate(features)
+
+    mode_line = "当前模式: 原始风险率（未加载校准模型）"
+    if CALIBRATION_MODEL:
+        mode_line = "当前模式: 知网对齐预测率（已加载校准模型）"
 
     return f"""
 # 综合AI风险率: {overall:.2%}
+# 预测知网AIGC率: {kn_like_rate:.2%}
+高风险段落占比: {high_ratio:.2%}
+中风险段落占比: {mid_ratio:.2%}
+有效段落数: {len(paragraphs)}
 
-（说明：该结果为“LLM生成风险分析”，并非绝对判定）
+{mode_line}
+
+（说明：该结果为“风险分析与校准预测”，并非官方系统结果）
 
 ---
 
 """ + "\n\n---\n\n".join(results)
 
+
 demo = gr.Interface(
     fn=analyze_pdf,
     inputs=gr.File(file_types=[".pdf"]),
     outputs="markdown",
     title="论文AIGC风险检测系统",
-    description="模拟知网/维普风格论文AI率分析"
+    description="论文AIGC风险分析（含知网对齐预测）",
 )
 
-demo.launch()
+
+demo.launch()

calibration/model.json

@@ -0,0 +1,7 @@
+{
+  "model_type": "linear",
+  "feature_order": ["overall", "p90", "high_ratio", "mid_ratio", "std"],
+  "coef": [1.0, 0.0, 0.0, 0.0, 0.0],
+  "intercept": 0.0,
+  "note": "default passthrough calibration; replace with trained calibration model"
+}

scripts/train_calibration.py

@@ -0,0 +1,62 @@
+import argparse
+import json
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+from sklearn.linear_model import LinearRegression
+from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
+
+FEATURE_ORDER = ["overall", "p90", "high_ratio", "mid_ratio", "std"]
+
+
+def clip01(x):
+    return np.clip(x, 0.0, 1.0)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Train kn-like calibration model")
+    parser.add_argument("--input", required=True, help="CSV with feature columns + target")
+    parser.add_argument("--target", default="kn_rate", help="target column name")
+    parser.add_argument("--out", default="calibration/model.json", help="output model json")
+    args = parser.parse_args()
+
+    df = pd.read_csv(args.input)
+
+    missing = [c for c in FEATURE_ORDER + [args.target] if c not in df.columns]
+    if missing:
+        raise ValueError(f"Missing columns: {missing}")
+
+    X = df[FEATURE_ORDER].astype(float).values
+    y = df[args.target].astype(float).values
+
+    model = LinearRegression()
+    model.fit(X, y)
+
+    pred = clip01(model.predict(X))
+
+    metrics = {
+        "mae": float(mean_absolute_error(y, pred)),
+        "rmse": float(np.sqrt(mean_squared_error(y, pred))),
+        "r2": float(r2_score(y, pred)),
+        "n": int(len(df)),
+    }
+
+    payload = {
+        "model_type": "linear",
+        "feature_order": FEATURE_ORDER,
+        "coef": [float(v) for v in model.coef_.tolist()],
+        "intercept": float(model.intercept_),
+        "train_metrics": metrics,
+    }
+
+    out = Path(args.out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    print("Saved:", out)
+    print("Metrics:", metrics)
+
+
+if __name__ == "__main__":
+    main()