RAG chunk 元数据管理：给每个片段打上时间戳、权限和来源标记-拓冰建站

RAG chunk 元数据管理：给每个片段打上时间戳、权限和来源标记

一、深度引言与场景痛点

大家好，我是赵咕咕。

你辛辛苦苦搭了一套 RAG 系统，用户问"公司最近的报销政策是什么？"，系统给它返回了2022年的旧政策——因为向量检索只看语义相似度，完全不关心文档的时效性。

这就是 RAG 系统中一个容易被忽略的关键组件：chunk 元数据管理。

向量检索解决的是"语义匹配"，但它天然缺失了对时间、权限、来源这些结构性约束的支持。如果你不给每个 chunk 打上元数据标签，检索结果可能给用户返回过期的、无权限的、来源不明的信息。

这篇文章我们来探讨：如何设计一套生产级的 chunk 元数据管理系统，让检索结果不仅语义相关，而且在时间、权限和来源维度上也是可靠的。

二、底层机制与原理深度剖析

RAG 的检索本质是：query_vector → top_k nearest neighbors in vector space。这个过程中，向量距离是唯一的排序依据。但实际业务中，我们需要多维度的过滤条件。

元数据管理的核心思想是：在向量检索之上叠加结构化过滤层。检索分两步：

预过滤：用结构化条件缩小候选集（如时间范围、权限列表）
向量检索：在筛后的候选集中做语义匹配
后过滤：对返回结果做二次校验（如来源优先级排序）

架构如下：

flowchart TB A[用户 Query] --> B[元数据提取器<br/>提取时间/权限/来源约束] B --> C{预过滤层<br/>结构化过滤} C --> D[(向量数据库<br/>Filtered Index)] D --> E[向量语义检索<br/>ANN Search] E --> F[后过滤层<br/>二次校验与排序] F --> G[最终结果<br/>Top-K Chunks] B -.-> H[时间过滤器<br/>时间窗口过滤] B -.-> I[权限过滤器<br/>ACL 过滤] B -.-> J[来源过滤器<br/>source_type 过滤] H --> C I --> C J --> C style C fill:#fff3e0 style D fill:#e8f5e9 style F fill:#e3f2fd

元数据字段设计是最关键的一步。一个 chunk 至少需要以下字段：

created_at：文档创建时间（用于时效性过滤）
updated_at：最后更新时间（用于检测是否过时）
expires_at：过期时间（可选，政策类文档需要）
access_level：权限等级（如 public → internal → restricted → confidential）
allowed_users或allowed_roles：允许访问的用户/角色列表
source_type：来源类型（wiki、api_doc、code_repo、ticket）
source_url：来源链接（用于溯源和引用）
version：文档版本号

三、生产级代码实现

下面是完整的 chunk 元数据管理实现：

from __future__ import annotations import asyncio from dataclasses import dataclass, field, asdict from typing import Optional from enum import Enum import hashlib import time class AccessLevel(Enum): """访问权限等级""" PUBLIC = "public" INTERNAL = "internal" RESTRICTED = "restricted" CONFIDENTIAL = "confidential" @classmethod def can_access( cls, doc_level: "AccessLevel", user_level: "AccessLevel" ) -> bool: """判断用户是否有权访问该文档""" levels = list(cls) return levels.index(user_level) >= levels.index(doc_level) class SourceType(Enum): """文档来源类型""" WIKI = "wiki" API_DOC = "api_doc" CODE_REPO = "code_repo" TICKET = "ticket" CHANGELOG = "changelog" MEETING_NOTE = "meeting_note" @property def priority(self) -> int: """来源优先级（数字越大越可信）""" priority_map = { SourceType.WIKI: 5, SourceType.API_DOC: 5, SourceType.CODE_REPO: 4, SourceType.CHANGELOG: 4, SourceType.TICKET: 2, SourceType.MEETING_NOTE: 1, } return priority_map.get(self, 1) @dataclass class ChunkMetadata: """Chunk 元数据""" chunk_id: str content_hash: str # 内容哈希，用于去重 source_type: SourceType source_url: str access_level: AccessLevel = AccessLevel.INTERNAL allowed_roles: list[str] = field(default_factory=list) created_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time) expires_at: Optional[float] = None # None 表示永不过期 version: int = 1 tags: list[str] = field(default_factory=list) title: str = "" @property def is_expired(self) -> bool: """判断 chunk 是否已过期""" if self.expires_at is None: return False return time.time() > self.expires_at @property def age_days(self) -> float: """chunk 年龄（天数）""" return (time.time() - self.created_at) / 86400 def time_decay_score(self) -> float: """时间衰减分数：越旧的文档分数越低""" if self.age_days < 1: return 1.0 # 指数衰减：半衰期 30 天 return max(0.1, pow(0.5, self.age_days / 30)) def to_filter_dict(self) -> dict: """转为向量数据库的 filter 格式""" return { "source_type": self.source_type.value, "access_level": self.access_level.value, "allowed_roles": {"$in": self.allowed_roles} if self.allowed_roles else None, "created_at": {"$gte": self.created_at}, "version": self.version, } class ChunkMetadataManager: """Chunk 元数据管理器""" @staticmethod def create_metadata( content: str, source_type: SourceType, source_url: str, **kwargs ) -> ChunkMetadata: """为一段内容创建元数据""" chunk_id = hashlib.md5( f"{source_url}:{content[:100]}".encode() ).hexdigest()[:16] content_hash = hashlib.sha256(content.encode()).hexdigest() return ChunkMetadata( chunk_id=chunk_id, content_hash=content_hash, source_type=source_type, source_url=source_url, **kwargs ) @staticmethod async def filter_chunks( chunks: list[tuple[str, ChunkMetadata, float]], # (content, meta, score) user_roles: list[str], time_window_days: Optional[int] = None, min_source_priority: int = 0, ) -> list[tuple[str, ChunkMetadata, float]]: """对检索结果做多维度过滤""" filtered = [] for content, meta, score in chunks: # 1. 过期检查 if meta.is_expired: continue # 2. 时间窗口检查 if time_window_days and meta.age_days > time_window_days: continue # 3. 权限检查 if not ChunkMetadataManager._check_permission(meta, user_roles): continue # 4. 来源优先级检查 if meta.source_type.priority < min_source_priority: continue # 5. 时间衰减重排序 adjusted_score = score * meta.time_decay_score() filtered.append((content, meta, adjusted_score)) # 按调整后分数排序 filtered.sort(key=lambda x: x[2], reverse=True) return filtered @staticmethod def _check_permission( meta: ChunkMetadata, user_roles: list[str] ) -> bool: """检查用户是否有权限访问""" # 公开文档任何人可访问 if meta.access_level == AccessLevel.PUBLIC: return True # 检查用户角色是否在允许列表中 if meta.allowed_roles: return any(r in meta.allowed_roles for r in user_roles) return True # 使用示例 async def main(): manager = ChunkMetadataManager() # 创建元数据 meta = manager.create_metadata( content="报销流程：员工提交申请 → 部门审批 → 财务审核 → 打款", source_type=SourceType.WIKI, source_url="https://wiki.internal/expense-policy", access_level=AccessLevel.INTERNAL, allowed_roles=["employee", "manager", "finance"], expires_at=time.time() + 86400 * 90, # 90天后过期 tags=["报销", "财务", "政策"], title="差旅报销政策 v3" ) print(f"Chunk ID: {meta.chunk_id}") print(f"是否过期: {meta.is_expired}") print(f"时间衰减分: {meta.time_decay_score():.2f}") # 模拟检索后过滤 mock_results = [ ("旧政策内容...", meta, 0.95), ] filtered = await manager.filter_chunks( mock_results, user_roles=["employee"], time_window_days=60, ) if filtered: print(f"过滤后剩余 {len(filtered)} 条结果") else: print("所有结果被过滤（权限制、过期、或超出时间窗口）") asyncio.run(main())