RAG 存量数据清理

from qdrant_client import QdrantClient

client = QdrantClient(host="localhost", port=6333)

# 统计各来源文档的 chunk 数量和时间分布
def audit_metadata(collection: str):
    result = {}
    offset = None
    while True:
        points, offset = client.scroll(
            collection_name=collection,
            scroll_filter=None,
            limit=1000,
            offset=offset,
            with_payload=True,
            with_vectors=False,   # 只读元数据，省内存
        )
        for p in points:
            src = p.payload.get("source", "unknown")
            ts  = p.payload.get("created_at", "")
            result.setdefault(src, {"count": 0, "oldest": ts})
            result[src]["count"] += 1
            if ts < result[src]["oldest"]:
                result[src]["oldest"] = ts
        if offset is None:
            break
    return result

import os

def find_orphan_chunks(audit: dict, doc_root: str) -> list[str]:
    """返回源文件已不存在的 source 列表"""
    orphans = []
    for src in audit:
        if src.startswith("http"):
            continue  # URL 类另行处理
        if not os.path.exists(os.path.join(doc_root, src)):
            orphans.append(src)
    return orphans

from datetime import datetime, timedelta

def find_stale_chunks(audit: dict, ttl_days: int = 180) -> list[str]:
    cutoff = datetime.now() - timedelta(days=ttl_days)
    stale = []
    for src, info in audit.items():
        try:
            ts = datetime.fromisoformat(info["oldest"])
            if ts < cutoff:
                stale.append(src)
        except Exception:
            pass
    return stale

import numpy as np
from sklearn.cluster import DBSCAN

def dedup_by_vector(collection: str, eps: float = 0.08, min_samples: int = 2):
    """
    DBSCAN 聚类：同一簇内的 chunk 语义高度相似 → 保留最新/最长，删其余。
    eps 对应余弦距离阈值（0.08 ≈ 余弦相似度 0.92）
    """
    # 1. 批量拉取所有向量
    points, _ = client.scroll(collection, limit=10000, with_vectors=True)
    ids    = [p.id for p in points]
    vecs   = np.array([p.vector for p in points])
    metas  = [p.payload for p in points]

    # 2. 余弦距离矩阵（大规模用 FAISS 替代）
    norms  = np.linalg.norm(vecs, axis=1, keepdims=True)
    normed = vecs / (norms + 1e-9)
    dist   = 1 - normed @ normed.T   # 余弦距离

    # 3. DBSCAN 聚类
    labels = DBSCAN(eps=eps, min_samples=min_samples, metric="precomputed").fit_predict(dist)

    # 4. 每个簇保留最新的 chunk，其余标记删除
    to_delete = []
    for cluster_id in set(labels):
        if cluster_id == -1:
            continue  # 噪声点（唯一 chunk），保留
        cluster_idx = [i for i, l in enumerate(labels) if l == cluster_id]
        # 按 created_at 降序，保留最新
        cluster_idx.sort(
            key=lambda i: metas[i].get("created_at", ""),
            reverse=True
        )
        to_delete.extend(ids[i] for i in cluster_idx[1:])  # 保留第一个

    return to_delete

# 执行删除（分批，避免大事务）
def batch_delete(collection: str, ids: list, batch_size: int = 200):
    for i in range(0, len(ids), batch_size):
        client.delete(
            collection_name=collection,
            points_selector=ids[i:i+batch_size],
        )
        print(f"deleted {min(i+batch_size, len(ids))}/{len(ids)}")

import asyncio
from sentence_transformers import SentenceTransformer

new_model = SentenceTransformer("BAAI/bge-large-zh-v1.5")  # 新模型

async def rebuild_collection(old_col: str, new_col: str, batch_size: int = 64):
    """异步批量重建，后台运行，不影响线上查询"""
    # 先建新集合（新向量维度可能不同）
    client.recreate_collection(
        collection_name=new_col,
        vectors_config={"size": 1024, "distance": "Cosine"},
    )
    offset = None
    total = 0
    while True:
        points, offset = client.scroll(
            collection_name=old_col,
            limit=batch_size,
            offset=offset,
            with_payload=True,
            with_vectors=False,   # 拿文本重 embed，不复用旧向量
        )
        if not points:
            break
        texts  = [p.payload["text"] for p in points]
        new_vecs = new_model.encode(texts, normalize_embeddings=True).tolist()
        client.upsert(
            collection_name=new_col,
            points=[
                {"id": p.id, "vector": v, "payload": p.payload}
                for p, v in zip(points, new_vecs)
            ]
        )
        total += len(points)
        print(f"rebuilt {total} chunks...")
        await asyncio.sleep(0.01)  # 让出 IO，避免打满 CPU
        if offset is None:
            break
    print(f"✅ rebuild done: {total} chunks → {new_col}")

# 批量给某 source 的所有 chunk 打上新字段
def patch_metadata(collection: str, source: str, patch: dict):
    from qdrant_client.models import Filter, FieldCondition, MatchValue, SetPayload

    client.set_payload(
        collection_name=collection,
        payload=patch,
        points=Filter(
            must=[FieldCondition(key="source", match=MatchValue(value=source))]
        ),
    )

# 示例：给旧文档标记为 deprecated
patch_metadata(
    "knowledge_base",
    source="docs/old-api-v1.pdf",
    patch={"status": "deprecated", "deprecated_at": "2025-06-01"},
)