RAG 前置过滤：元数据关键词匹配#

1
用户查询 → 预处理 → 元数据过滤（前置） → 向量检索 → 排序 → 生成

1
    User Query
2
        │
3
        ▼
4
  ┌──────────────┐
5
  │ 分词/词干提取 │ → 提取 query 中的关键词
6
  └──────┬───────┘
7
         ▼
8
  ┌──────────────┐
9
  │ BM25 评分    │ → 计算每个文档与关键词的相关性
10
  └──────┬───────┘
11
         ▼
12
  ┌──────────────┐
13
  │ 阈值筛选 Top-N │ → 保留 BM25 分数 > 阈值 或 Top-K 文档
14
  └──────┬───────┘
15
         ▼
16
    进入向量检索阶段

1
  User Query + 元数据条件
2
      │
3
      ▼
4
  ┌──────────────┐
5
  │ 解析元数据条件 │ → {category: "技术", date: >"2024-01"}
6
  └──────┬───────┘
7
         ▼
8
  ┌──────────────┐
9
  │ 数据库/索引过滤 │ → 只保留符合条件的文档 ID
10
  └──────┬───────┘
11
         │
12
         ▼
13
      向量检索只在过滤后的 ID 子集上进行

1
┌──────────────────────────────────────────────────────────┐
2
│             向量数据库 (Milvus/Qdrant/Weaviate)           │
3
├──────────────────────────────────────────────────────────┤
4
│  同时存储: 向量 + 元数据字段 (category, date, author)      │
5
│                                                          │
6
│  检索时: filter="category == '技术' && date > '2024-01'"  │
7
└──────────────────────────────────────────────────────────┘

1
def metadata_filter(documents, metadata_conditions):
2
    """
3
    documents: List[Document]
4
    metadata_conditions: {"key": "value"} 或 {"key": ["value1", "value2"]}
5
    """
6
    filtered = []
7
    for doc in documents:
8
        match = True
9
        for key, value in metadata_conditions.items():
10
            doc_value = doc.metadata.get(key)
11

12
            # 精确匹配
13
            if isinstance(value, str):
14
                if doc_value != value:
15
                    match = False
16

17
            # 多值匹配（OR）
18
            elif isinstance(value, list):
19
                if doc_value not in value:
20
                    match = False
21

22
        if match:
23
            filtered.append(doc)
24
    return filtered

1
# 用户查询：关于 Python RAG 的技术文档
2
user_query = "Python RAG 实现原理"
3

4
# 1. 元数据过滤（前置过滤）- 缩小范围
5
filtered_docs = collection.get(
6
    where={
7
        "$and": [
8
            {"language": "zh"},
9
            {"category": {"$in": ["tech", "tutorial"]}},
10
            {"published_year": {"$gte": 2024}}
11
        ]
12
    }
13
)
14

15
# 2. 向量检索（过滤后）
16
query_embedding = embedding_model.encode(user_query)
17
results = collection.query(
18
    query_embeddings=[query_embedding],
19
    n_results=10
20
)