混合搜索法律笔记本：Google Colab 教程

本教程面向有编程基础的开发者，深入讲解如何结合 Gemini API 与 Qdrant 向量数据库构建法律领域的混合检索管道。通过实际代码演示，您将学习使用 gemini-embedding-001 的 Matryoshka 表示进行高效检索与重排序，并利用 Qdrant 的 miniCOIL 实现关键词与语义搜索的融合。教程涵盖环境配置、数据预处理、集合创建、批量嵌入与上传，以及基于 hit@1 指标的实验评估，帮助您理解 RAG 系统中检索质量的关键因素。

技术背景

在法律 AI 应用中，如 RAG 聊天机器人，检索质量直接决定回答的准确性与事实性。传统关键词搜索（如 BM25）无法理解语义，而纯向量搜索可能忽略精确匹配。混合搜索结合两者优势，可显著提升检索效果。Gemini 嵌入模型支持 Matryoshka 表示，允许截断向量以降低存储与计算成本，同时保持较高精度。Qdrant 作为高性能向量数据库，原生支持混合查询与多阶段检索，适合生产级应用。

核心原理

核心原理包括：1) 使用 Gemini 嵌入模型生成 3072 维向量，其前 768 维已包含主要语义信息，可用于快速检索；2) 利用 Qdrant 的稀疏向量（miniCOIL）实现基于 BM25 的关键词匹配，通过 IDF 加权提升关键词重要性；3) 混合搜索通过 RRF 融合语义与关键词结果，兼顾语义理解与精确匹配；4) 重排序阶段使用完整向量对候选集重新排序，提高精度。

架构分析

系统架构分为数据层、索引层、检索层和评估层。数据层使用 Hugging Face 的 LegalQAEval 数据集，去重后存储文本块。索引层在 Qdrant 中创建集合，配置两个稠密向量（检索用 768 维、重排用 3072 维）和一个稀疏向量（miniCOIL）。检索层通过 Qdrant 客户端执行三种查询：基础检索、重排序、混合搜索。评估层使用 hit@1 指标比较不同策略。关键设计决策包括：使用 Matryoshka 截断向量降低内存占用，对重排向量禁用索引（m=0）以节省资源，以及使用流式上传处理大规模数据。

完整实战教程

##### 版权所有 2026 Google LLC.

# @title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

在法律AI中使用Gemini API与Qdrant向量搜索进行混合检索

<!-- 社区贡献者徽章 -->

<table>

<tr>

<!-- 作者头像单元格 -->

<td bgcolor="#d7e6ff">

<a href="https://github.com/mrscoopers" target="_blank" title="在 GitHub 上查看 Jenny 的个人资料">

</a>

</td>

<!-- 文本内容单元格 -->

<td bgcolor="#d7e6ff">

<h2><font color='black'>本笔记本由 <a href="https://github.com/mrscoopers" target="_blank"><font color='#217bfe'><strong>Jenny</strong></font></a> 贡献。</font></h2>

<h5><font color='black'><a href="https://www.linkedin.com/in/evgeniya-sukhodolskaya/"><font color="#078efb">Jenny的LinkedIn</font></a></h5></font><br>

<!-- 页脚 -->

<font color='black'><small><em>有很酷的Gemini示例？欢迎<a href="https://github.com/google-gemini/cookbook/blob/main/CONTRIBUTING.md" target="_blank"><font color="#078efb">分享它</font></a>！</em></small></font>

</td>

</tr>

</table>

注意： 此笔记本需要付费层级速率限制才能正常运行。（更多详情请参阅定价）。

概述

Austronaut judge

在法律领域，准确性和事实正确性至关重要。

一家与 Qdrant 合作的 Legal AI 初创公司概述了在 Legal AI 应用中确保这两点的方法（例如，基于检索增强生成（RAG）或代理式）：

*“把一切都变成检索问题，检索真实依据。如果这样设定，你就不必担心幻觉，因为给用户的一切都基于有效文档的某一部分。”*

确实，许多 Legal AI 企业在其应用中需要高质量的检索。要实现这一点，你需要：

本笔记本

在本笔记本中，您将学习如何将 gemini-embedding-001 与 Qdrant 向量搜索引擎提供的工具相结合，构建一个法律问答检索管道。

您将学习如何：

设置

安装 SDK

%pip install -q -U "google-genai>=2.9.0" qdrant-client[fastembed] datasets

设置你的 API 密钥：

要运行以下单元格，你的 API 密钥必须存储在 Colab 的 Secret 标签页中。

from google.colab import userdata

GEMINI_API_KEY = userdata.get('GEMINI_API_KEY')
QDRANT_API_KEY = userdata.get('QDRANT_API_KEY')
QDRANT_URL = userdata.get('QDRANT_URL')

第一步：下载数据集

你将使用 Isaacus 提供的 Hugging Face 数据集 之一，Isaacus 是一家法律人工智能研究公司。

法律 AI 中的一个常见用例是检索增强生成（RAG）聊天机器人。为了评估此类应用的检索性能，你需要一个问答（QA）数据集。

选择数据集

LegalQAEval

该数据集包含约2400个问答对，包括：

加载法律问答语料库；你将使用所有可用的数据划分。

from datasets import load_dataset, concatenate_datasets

corpus = concatenate_datasets(load_dataset('isaacus/LegalQAEval', split=['val', 'test']))

文本块去重

由于数据集中可能包含多个问题对应的text块，因此首先对text字段进行去重，以避免多次存储相同的信息。

import pandas as pd
import datasets

# Convert the Hugging Face dataset to a pandas DataFrame
df = corpus.to_pandas()

# Group by 'text' and aggregate 'id' into a list
grouped_corpus = df.groupby('text')['id'].apply(list).reset_index().rename(columns={'id': 'ids'})

corpus_deduplicated = datasets.Dataset.from_pandas(grouped_corpus)

步骤 2：定义用例配置

在典型的法律聊天机器人场景中，用户提出问题，LLM 根据相关的文本块生成答案。

为了模拟这一过程，你需要在 Qdrant 中存储 text 块的数值表示（嵌入）。

在检索期间，question 将被转换为同一嵌入空间中的数值表示。然后，将在向量索引中找到（近似）最近的 text 块。

Gemini 嵌入模型支持 RAG 风格的问答检索（任务类型 QUESTION_ANSWERING）。

现在，为了完全定义我们的存储配置，让我们考虑与法律 AI 领域常见 RAG 用例相关的几个因素。

成本与准确性：套娃表示

Gemini gemini-embedding-001 嵌入是 3072 维的。

在拥有约 100 万个块的 RAG 设置中，将这些嵌入存储在 RAM 中（用于快速检索）将需要约 12 GB。

Gemini 嵌入模型支持一种平衡检索成本与准确性的方法。它使用 Matryoshka 表示学习 (MRL) 进行训练，这意味着编码文本的最重要信息存储在嵌入的前几个维度中。

因此，例如，您可以：

融合两者优势的准确性：混合搜索

在法律用例中，通常将以下两种搜索方式的优势结合起来会更有益：

在 Qdrant 中，这两种方法可以通过混合查询和多阶段查询结合使用。

对于基于关键词的部分，Qdrant 支持多种选项，从传统的 BM25 到稀疏神经检索器（如 SPLADE）。在这些选项中，有我们自定义改进的 BM25，称为 miniCOIL，您将在本笔记本中使用它。

在 Qdrant 中，基于关键词的检索通过稀疏向量实现。

集合配置

为法律问答检索管道配置一个 Qdrant 集合。

from qdrant_client import QdrantClient, models

qdrant_client = QdrantClient(  # Initializing Qdrant client.
    url=QDRANT_URL,
    api_key=QDRANT_API_KEY,
)

COLLECTION_NAME = "legal_AI_QA"
GEMINI_EMBEDDING_RETRIEVAL_SIZE = 768
GEMINI_EMBEDDING_FULL_SIZE = 3072

if not qdrant_client.collection_exists(collection_name=COLLECTION_NAME):
    qdrant_client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config={
            "gemini_embedding_retrieve": models.VectorParams(
                size=GEMINI_EMBEDDING_RETRIEVAL_SIZE,  # Smaller embeddings for faster retrieval.
                distance=models.Distance.COSINE,
            ),
            "gemini_embedding_rerank": models.VectorParams(
                size=GEMINI_EMBEDDING_FULL_SIZE,  # Full-sized embeddings for precision-boosting reranking.
                distance=models.Distance.COSINE,
                hnsw_config=models.HnswConfigDiff(
                    m=0  # Since these embeddings aren't used for retrieval, you don't need to spend resources on building a vector index.
                ),
                on_disk=True,  # To save on RAM used for retrieval.
            ),
        },
        sparse_vectors_config={
            "miniCOIL": models.SparseVectorParams(
                modifier=models.Modifier.IDF  # Inverse Document Frequency statistic, computed on the Qdrant side.
            )
        },
    )

步骤 3：将文本嵌入并索引到 Qdrant

为了加快数据转换过程，你需要：

  1. 使用 get_embeddings_batch 函数批量嵌入所有 text 块。
  2. 将结果批量上传到 Qdrant。

    3. Qdrant Python 客户端提供了 upload_collection 和 upload_points 函数。这些函数处理批处理、重试和并行化。它们接受生成器作为输入，因此你需要为此创建一个生成器函数 qdrant_points_stream。

注意： 如果你的集合中的距离函数设置为 COSINE（余弦相似度），Qdrant 会自动规范化上传的嵌入。这意味着你不需要预先规范化截断的 Gemini Matryoshka 嵌入，正如 Gemini 文档中所建议的。
GEMINI_MODEL_ID = "gemini-embedding-001" # @param ["gemini-embedding-001"] {"allow-input":true, isTemplate: true}

from google import genai
from google.genai import types
from google.api_core import retry
import uuid

google_client = genai.Client(api_key=GEMINI_API_KEY)

@retry.Retry(timeout=300)
def get_embeddings_batch(texts, task_type: str = "RETRIEVAL_DOCUMENT"):
    """Generates embeddings for a batch of texts.

    Args:
        texts: A list of strings to embed.
        task_type: The task type for the embedding model.

    Returns:
        A list of embedding vectors.

    Raises:
        Exception: If an error occurs during embedding generation.
    """
    try:
        res = google_client.models.embed_content(
            model=GEMINI_MODEL_ID,
            contents=texts,
            config=types.EmbedContentConfig(task_type=task_type),
        )
        return [e.values for e in res.embeddings]
    except Exception as e:
        print(f"An error occurred while getting embeddings: {e}")
        raise


def qdrant_points_stream(corpus, avg_corpus_text_length, gemini_batch_size: int = 8):
    """Streams Qdrant points with embeddings for a given corpus.

    Args:
        corpus: The dataset to process.
        avg_corpus_text_length: The average text length for miniCOIL (based on BM25 formula).
        gemini_batch_size: The batch size for Gemini embedding requests.

    Yields:
        Qdrant PointStruct objects.
    """
    for start in range(0, len(corpus), gemini_batch_size):  # Iterate over the dataset in batches.
        end = min(start + gemini_batch_size, len(corpus))
        batch = corpus.select(range(start, end))  # Current batch slice.

        gemini_embeddings_full = get_embeddings_batch(
            [row["text"] for row in batch], task_type="RETRIEVAL_DOCUMENT"
            )  # Generate embeddings for this batch.

        for batch_item, gemini_embedding_full in zip(batch, gemini_embeddings_full):
            yield models.PointStruct(
                id=str(uuid.uuid4()),  # Unique ID (string UUID or integer supported by Qdrant).
                payload={  # Metadata stored alongside the vector.
                    "text": batch_item["text"],  # Raw text for users/LLMs.
                    "ids": batch_item["ids"],  # IDs of the QA pairs related to this `text` (for later evaluation).
                },
                vector={  # Embeddings.
                    "gemini_embedding_rerank": gemini_embedding_full,  # Full Gemini embedding for reranking.
                    "gemini_embedding_retrieve": gemini_embedding_full[:768],  # Truncated Gemini embedding for retrieval.
                    "miniCOIL": models.Document(  # Custom Qdrant-optimized BM25 replacement.
                        text=batch_item["text"],
                        model="Qdrant/minicoil-v1",
                        options={"avg_len": avg_corpus_text_length, "k": 0.9, "b": 0.4},  # Corpus avg length, k_1 & b from BM25 formula.
                    ),
                },
            )

现在，你将嵌入数据并上传嵌入向量。

尝试在生成嵌入向量并上传到 Qdrant 时使用不同的批处理大小。

最快的设置通常取决于你的网络速度和内存/CPU/GPU，并且请记住，嵌入推理并不是一个非常快速的过程。

Qdrant 用于混合搜索中基于关键词的检索部分的表示是由 Qdrant 生成的。

在 Colab 中，Qdrant 会在你首次使用时下载所需的模型（在我们的案例中，是 Qdrant/minicoil-v1），因为需要它们将 text 块转换为稀疏表示。

import tqdm

COLLECTION_NAME = "legal_AI_QA"

# Estimating the average length of the texts in the corpus on the subsample of 1000, to use in BM25-inspired keywords-based retrieval.
SUBSET_SIZE = 1000
avg_corpus_text_length = sum(len(text.split()) for text in corpus["text"][:SUBSET_SIZE]) / SUBSET_SIZE

qdrant_client.upload_points(
    collection_name=COLLECTION_NAME,
    points=tqdm.tqdm(
        qdrant_points_stream(corpus_deduplicated,
                            avg_corpus_text_length=avg_corpus_text_length,
                            gemini_batch_size=4),
        desc="Uploading points",
    ),
    batch_size=4,
)

步骤 4：实验与评估

对于每个检索任务而言，关键在于尝试不同的工具，并基于合理的指标进行评估。

指标

在RAG中，目标通常是在非常小的N值下，于前N个检索结果中获取正确结果，因为这是LLM用来生成有依据答案的内容，并且你希望节省上下文窗口大小/降低令牌成本。

你将使用指标 hit@1，这意味着排名第一的文本块实际上就是问题的答案。

评估集

在实验中，你应当只使用 answers 字段不为 null 的问题，因为这保证了该文本块包含问题的答案。

questions = corpus.filter(lambda item: len(item['answers']) > 0)

为所有问题推断Gemini嵌入，这样你就可以自由实验，而无需花费额外的时间或金钱。

import tqdm

question_embeddings = {}

question_texts = [q['question'] for q in questions]
question_ids = [q['id'] for q in questions]
all_embeddings = []

BATCH_SIZE = 32

for i in tqdm.tqdm(range(0, len(question_texts), BATCH_SIZE), desc="Embedding questions"):
    batch_texts = question_texts[i:i + BATCH_SIZE]
    embeddings = get_embeddings_batch(batch_texts, task_type="QUESTION_ANSWERING")
    all_embeddings.extend(embeddings)

question_embeddings = {qid: emb for qid, emb in zip(question_ids, all_embeddings)}

然后随机选择一个测试子集。

questions = questions.shuffle(seed=42).select(range(500))

实验

有很多方法可以改进搜索结果。例如，仅重排序就可以使用高维嵌入（如Gemini）、多向量（如ColBERT）或交叉编码器来完成。

为简单起见，你将专注于三种简单的检索方法，这三个实验对于法律等对高精度要求高的领域来说是一个很好的起点：

实验1：基础检索

使用截断的Gemini嵌入进行基础检索。

这为你提供了一个简单的参考点，以便比较改进效果。

实验2：重排序

使用完整尺寸的Gemini嵌入对检索到的子集进行重排序。

更大的嵌入能够捕捉到检索器可能遗漏的更精细的语义细节。

实验3：混合搜索

在混合搜索中结合语义（捕捉含义）和基于关键词（确保精确匹配）的检索。

*对于这个玩具数据集，关键词匹配可能不会增加太多价值，因为所有问题都是非常“对话式”的风格，没有太多关键词，但在现实生活中的法律AI检索中，它会产生影响*

设置如下：

  1. 使用相同的查询运行两次搜索。
  2. 使用融合算法将结果合并为单个列表。这里你将使用倒数排名融合（RRF），这是一种简单且众所周知的零样本融合方法。
COLLECTION_NAME = "legal_AI_QA"
N = 1

def hit_at_n(results: list[models.ScoredPoint], question_id: str, N: int) -> int:
    """Calculates if the correct document is within the top N retrieved results.

    Args:
        results: A list of scored points from a Qdrant search.
        question_id: The ID of the question to check for.
        N: The number of top results to check.

    Returns:
        1 if the correct document is found, 0 otherwise.
    """
    for result in results:
        if question_id in result.payload["ids"]:
            return 1
    return 0

hits_baseline = 0
hits_rerank = 0
hits_hybrid = 0

for question in tqdm.tqdm(questions, desc=f"Evaluating hits@{N}"):
    full_gemini_embedding = question_embeddings[question['id']] # Embedding of the question.

    ## Experiment 1: Baseline
    result_baseline = qdrant_client.query_points(
        collection_name=COLLECTION_NAME,
        query=full_gemini_embedding[:768],  # Use the first quarter of the Gemini embedding for retrieval.
        limit=N,
        using="gemini_embedding_retrieve",
        with_payload=True,
    )
    hits_baseline += hit_at_n(result_baseline.points, question_id=question['id'], N=N)

    ## Experiment 2: Reranking
    result_rerank = qdrant_client.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=models.Prefetch( # First, retrieve 50 candidates with the smaller embedding.
            query=full_gemini_embedding[:768],
            using="gemini_embedding_retrieve",
            limit=50
        ),
        query=full_gemini_embedding, # Then rerank those 50 results with the full embedding.
        limit=N,
        using="gemini_embedding_rerank",
        with_payload=True
    )
    hits_rerank += hit_at_n(result_rerank.points, question_id=question['id'], N=N)

    ## Experiment 3: Hybrid search (semantic + keyword)
    result_hybrid = qdrant_client.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=[
            models.Prefetch( # Retrieve 25 results using semantic search (Gemini truncated embeddings).
                query=full_gemini_embedding[:768],
                using="gemini_embedding_retrieve",
                limit=25
            ),
            models.Prefetch( # Retrieve 25 results using miniCOIL (Qdrant’s custom improved version of BM25-based keyword search).
                query=models.Document(
                    text=question['question'],
                    model="Qdrant/minicoil-v1"
                ),
                using="miniCOIL",
                limit=25
            )
        ],
        query=models.FusionQuery(fusion="rrf"), # Fuse the two result sets with Reciprocal Rank Fusion (RRF).
        limit=N,
        with_payload=True
    )
    hits_hybrid += hit_at_n(result_hybrid.points, question_id=question['id'], N=N)


hits_at_N_baseline = hits_baseline / len(questions) # Compute average hits@N for each experiment.
hits_at_N_rerank = hits_rerank / len(questions)
hits_at_N_hybrid = hits_hybrid / len(questions)

print("\n")
print(f"Retrieval Avg Hits@{N}: {hits_at_N_baseline:.3f} ({hits_baseline}/{len(questions)})")
print(f"Rerank Avg Hits@{N}: {hits_at_N_rerank:.3f} ({hits_rerank}/{len(questions)})")
print(f"Hybrid Search Avg Hits@{N}: {hits_at_N_hybrid:.3f} ({hits_hybrid}/{len(questions)})")

后续步骤

在本笔记本中，您使用 Qdrant 向量搜索引擎 和 Gemini Embeddings 为典型的法律 RAG 聊天机器人设置了 检索管道。

您尝试了几种检索方法，利用了 Gemini 生成 Matryoshka 表示的能力以及 Qdrant 的检索工具，包括重排序和混合搜索。

当然，法律应用需要的远不止一个简单的零样本管道。检索质量始终取决于数据集和用例，因此除了实验和迭代之外，没有银弹。

从零样本出发的下一步：

将此 notebook 作为基线，在此基础上构建并实验，以找到最适合你的方案！

Austronaut judge 2

常见问题

常见错误：1) API 密钥未正确设置，导致认证失败；2) 集合已存在导致创建冲突；3) 嵌入批次过大导致超时；4) miniCOIL 模型首次下载失败。排查方法：检查密钥是否在 Colab Secret 中；使用 collection_exists 检查；调整批次大小并启用重试；确保网络连接并手动下载模型。生产优化：使用异步客户端、增加批处理大小、调整 HNSW 参数（如 ef）、使用 GPU 加速嵌入、缓存嵌入结果、监控 Qdrant 性能指标。

企业应用扩展

后续可探索：1) 使用 ColBERT 多向量表示进行更精细的检索；2) 集成交叉编码器进行重排序；3) 实现查询扩展与过滤器提取；4) 调整 RRF 融合参数或使用加权融合；5) 针对特定法律领域微调嵌入模型；6) 部署为 REST API 服务，集成到实际应用中。

总结

本教程面向有编程基础的开发者，深入讲解如何结合 Gemini API 与 Qdrant 向量数据库构建法律领域的混合检索管道。通过实际代码演示，您将学习使用 gemini-embedding-001 的 Matryoshka 表示进行高效检索与重排序，并利用 Qdrant 的 miniCOIL 实现关键词与语义搜索的融合。教程涵盖环境配置、数据预处理、集合创建、批量嵌入与上传，以及基于 hit@1 指标的实验评估，帮助您理解 RAG 系统中检索质量的关键因素。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部RAG 与知识库教程