RAG 数据清理

# PDF
import fitz  # PyMuPDF
def extract_pdf(path: str) -> list[dict]:
    doc = fitz.open(path)
    pages = []
    for i, page in enumerate(doc):
        text = page.get_text("text")          # 优先 text 层
        if len(text.strip()) < 50:            # 疑似扫描件
            text = page.get_text("blocks")    # fallback 块模式
        pages.append({
            "text": clean_noise(text),
            "meta": {"source": path, "page": i + 1}
        })
    return pages

# HTML（网页）
from trafilatura import extract
def extract_html(html: str, url: str) -> dict:
    text = extract(html, include_tables=True, no_fallback=False)
    return {"text": text or "", "meta": {"source": url}}

# Excel / CSV
import pandas as pd
def extract_table(path: str) -> list[dict]:
    df = pd.read_excel(path)  # 或 read_csv
    rows = []
    for _, row in df.iterrows():
        rows.append({
            "text": " | ".join(f"{k}: {v}" for k, v in row.items() if pd.notna(v)),
            "meta": {"source": path, "row": _}
        })
    return rows

import re
def clean_noise(text: str) -> str:
    text = re.sub(r'\f', '\n', text)                    # 换页符
    text = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f]', '', text)  # 控制字符
    text = re.sub(r'第 \d+ 页.*?共 \d+ 页', '', text)   # 页眉/页脚
    text = re.sub(r'[ \t]{3,}', ' ', text)              # 多余空白
    text = re.sub(r'\n{3,}', '\n\n', text)              # 多余空行
    return text.strip()

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # token 数（用 tiktoken 计算）
    chunk_overlap=64,     # 重叠防止语义断裂
    separators=["\n\n", "\n", "。", "；", " "],  # 优先按自然段
)
chunks = splitter.split_text(text)

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("BAAI/bge-small-zh-v1.5")

def semantic_chunk(sentences: list[str], threshold: float = 0.8) -> list[str]:
    embs = model.encode(sentences, normalize_embeddings=True)
    chunks, cur = [], [sentences[0]]
    for i in range(1, len(sentences)):
        sim = float(np.dot(embs[i-1], embs[i]))
        if sim < threshold:       # 语义跳跃 → 切块
            chunks.append("".join(cur))
            cur = []
        cur.append(sentences[i])
    if cur:
        chunks.append("".join(cur))
    return chunks

# 伪代码
parent_chunks = split_by_heading(doc)  # 按 H2/H3 切
for p in parent_chunks:
    child_chunks = fixed_split(p, size=256, overlap=32)
    for c in child_chunks:
        c["parent_id"] = p["id"]       # 关联回 parent

chunk_doc = {
    "id": sha256(text + source),       # 全局唯一 ID
    "text": chunk_text,
    "metadata": {
        "source": "docs/faq.pdf",
        "page": 3,
        "section": "安装指南",
        "created_at": "2024-06-01",
        "doc_type": "pdf",             # 供检索时 filter
        "language": "zh",
    }
}

results = collection.query(
    vector=query_emb,
    filter={"doc_type": "pdf", "created_at": {"gte": "2024-01-01"}},
    limit=5
)

import hashlib, redis

r = redis.Redis()
def dedup_exact(text: str) -> bool:
    h = hashlib.md5(text.encode()).hexdigest()
    if r.sismember("seen_hashes", h):
        return True   # 重复
    r.sadd("seen_hashes", h)
    return False

from datasketch import MinHash, MinHashLSH

lsh = MinHashLSH(threshold=0.85, num_perm=128)

def get_minhash(text: str) -> MinHash:
    m = MinHash(num_perm=128)
    for word in text.split():
        m.update(word.encode())
    return m

def dedup_semantic(doc_id: str, text: str) -> bool:
    m = get_minhash(text)
    result = lsh.query(m)
    if result:
        return True   # 近似重复
    lsh.insert(doc_id, m)
    return False

def quality_filter(chunk: dict) -> bool:
    text = chunk["text"]
    # 1. 长度过滤
    if len(text) < 50 or len(text) > 2000:
        return False
    # 2. 中文字符比例（防止乱码/纯英文/代码块混入正文）
    zh_ratio = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') / len(text)
    if zh_ratio < 0.1:
        return False
    # 3. 重复字符检测（如 "。。。。。。"）
    if max(text.count(c) for c in set(text)) / len(text) > 0.3:
        return False
    # 4. 困惑度过滤（可选，用小型 LM 打分）
    # ppl = compute_perplexity(text)
    # if ppl > 500: return False
    return True