基于 Qwen3 和 Milvus 的 RAG 实战：多语言、指令调优与精细排序

当你在关注embedding模型领域时，Alibaba 最近发布的 Qwen3 embedding 系列无疑会吸引你的目光。该系列不仅推出了 embedding 模型，还发布了reranking 模型，各有三种尺寸（0.6B、4B、8B），它们都基于 Qwen3 基础模型构建，并专门针对检索任务进行了优化。本文将深入探讨如何使用 Qwen3-Embedding-0.6B 和 Qwen3-Reranker-0.6B 结合 Milvus 搭建一个完整的 RAG 系统，体验其多语言、指令调优以及通过智能 reranking 实现的性能提升。

Qwen3 系列：不仅仅是开源，更是生产力

Qwen3 系列有几个显著的特点，使其在众多 embedding 模型中脱颖而出：

• 多语言 embeddings：Qwen3 声称拥有跨越 100 多种语言的统一语义空间。这意味着你可以使用相同的模型来处理不同语言的文本，而无需为每种语言单独训练模型。
• 指令提示 (Instruction prompting)：你可以传递自定义指令来修改 embedding 行为。这对于针对特定领域进行调优非常有用，例如，你可以指示模型更关注技术术语或特定情感色彩。
• 可变维度：通过 Matryoshka Representation Learning 支持不同的 embedding 大小。这允许你根据你的具体需求在性能和资源消耗之间进行权衡。
• 32K 上下文长度：可以处理更长的输入序列，这对于处理长文档或复杂的查询非常有用。
• 标准的双/交叉编码器设置：embedding 模型使用双编码器，reranker 使用交叉编码器。这种架构设计既保证了检索效率，又提升了排序精度。

从基准测试结果来看，Qwen3-Embedding-8B 在 MTEB 多语言排行榜上取得了 70.58 的分数，超越了 BGE、E5 甚至 Google Gemini。Qwen3-Reranker-8B 在多语言排序任务中也达到了 69.02。这些成绩表明，Qwen3 系列不仅仅是“开源模型中表现良好”，而是全面地匹配甚至超越了主流商业 API。尤其在 RAG 检索、跨语言搜索和代码搜索系统中，尤其是在中文语境下，这些模型已经具备了生产级别的能力。

RAG 系统架构：分层检索，精益求精

本文将构建的 RAG 系统采用两阶段检索流水线：

1. 密集检索 (Dense retrieval)：使用 Qwen3 embedding 快速选择候选文档。
2. 重排序 (Reranking)：使用 Qwen3 交叉编码器进行精度优化。
3. 生成 (Generation)：使用 OpenAI 的 GPT-4 来生成最终响应。

这种架构结合了速度和准确性，通过 reranking 过滤掉初始检索结果中的噪声，从而提高最终答案的质量。

环境搭建：确保兼容性是关键

首先，我们需要安装必要的依赖项。请注意，最低版本要求非常重要，以确保兼容性：

pip install --upgrade pymilvus openai requests tqdm sentence-transformers transformers

要求 transformers>=4.51.0 和 sentence-transformers>=2.7.0。

我们还将使用 OpenAI 作为生成模型。设置你的 API 密钥：

import os
os.environ["OPENAI_API_KEY"] = "sk-***********"

数据准备：Milvus 文档作为知识库

我们将使用 Milvus 文档作为知识库。它包含了丰富的技术内容，可以很好地测试检索和生成质量。

下载并解压文档：

! wget https://github.com/milvus-io/milvus-docs/releases/download/v2.4.6-preview/milvus_docs_2.4.x_en.zip
! unzip -q milvus_docs_2.4.x_en.zip -d milvus_docs

加载和分块 Markdown 文件。这里我们使用一个简单的基于标题的分割策略。对于生产系统，可以考虑更复杂的分块方法，例如使用递归字符文本分割器（RecursiveCharacterTextSplitter），它能够根据指定的字符列表（例如：[“\n\n”, “\n”, ” “, “”]）递归地将文本分割成更小的块，并且可以设置块的大小和重叠度，以优化 RAG 系统的检索效果。

from glob import glob

text_lines = []
for file_path in glob("milvus_docs/en/faq/*.md", recursive=True):
    with open(file_path, "r") as file:
        file_text = file.read()
    text_lines += file_text.split("# ")

模型配置：轻量级模型，性能与资源兼顾

现在我们初始化模型。我们使用轻量级的 0.6B 版本，它在性能和资源需求之间提供了良好的平衡：

from openai import OpenAI
from sentence_transformers import SentenceTransformer
import torch
from transformers import AutoModel, AutoTokenizer, AutoModelForCausalLM

# 初始化 OpenAI 客户端，用于 LLM 生成
openai_client = OpenAI()

# 加载 Qwen3-Embedding-0.6B 模型，用于文本 embedding
embedding_model = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")

# 加载 Qwen3-Reranker-0.6B 模型，用于 reranking
reranker_tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-Reranker-0.6B", padding_side='left')
reranker_model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-Reranker-0.6B").eval()

# Reranker 配置
token_false_id = reranker_tokenizer.convert_tokens_to_ids("no")
token_true_id = reranker_tokenizer.convert_tokens_to_ids("yes")
max_reranker_length = 8192
prefix = "<|im_start|>system\nJudge whether the Document meets the requirements based on the Query and the Instruct provided. Note that the answer can only be \"yes\" or \"no\".<|im_end|>\n<|im_start|>user\n"
suffix = "<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
prefix_tokens = reranker_tokenizer.encode(prefix, add_special_tokens=False)
suffix_tokens = reranker_tokenizer.encode(suffix, add_special_tokens=False)

Embedding 函数：指令调优的关键

使用 Qwen3 embeddings 的关键在于能够为查询和文档使用不同的提示。这个看似微小的细节可以显著提高检索性能。例如，对于查询，我们可以使用一个更注重信息检索的提示；而对于文档，我们可以使用一个更注重语义理解的提示。

def emb_text(text, is_query=False):
    """
    使用 Qwen3-Embedding-0.6B 模型生成文本 embeddings。

    Args:
        text: 要 embed 的输入文本
        is_query: 是否为查询 (True) 或文档 (False)

    Returns:
        embedding 值列表
    """
    if is_query:
        # 对于查询，使用 "query" 提示以获得更好的检索性能
        embeddings = embedding_model.encode([text], prompt_name="query")
    else:
        # 对于文档，使用默认编码
        embeddings = embedding_model.encode([text])
    return embeddings[0].tolist()

让我们测试一下 embedding 函数，并检查输出维度：

test_embedding = emb_text("This is a test")
embedding_dim = len(test_embedding)
print(f"Embedding dimension: {embedding_dim}")
print(f"First 10 values: {test_embedding[:10]}")

预期输出：

Embedding dimension: 1024
First 10 values: [-0.009923271834850311, -0.030248118564486504, -0.011494234204292297, ...]

Reranking 实现：精细排序，提升相关性

Reranker 使用交叉编码器架构来评估查询-文档对。与双编码器 embedding 模型相比，这种方法计算量更大，但可以提供更细致的相关性评分。交叉编码器会同时考虑查询和文档的上下文信息，从而更准确地判断它们之间的相关性。

以下是完整的 reranking 流水线：

def format_instruction(instruction, query, doc):
    """格式化 reranker 输入的指令"""
    if instruction is None:
        instruction = 'Given a web search query, retrieve relevant passages that answer the query'
    output = "<Instruct>: {instruction}\n<Query>: {query}\n<Document>: {doc}".format(
        instruction=instruction, query=query, doc=doc
    )
    return output

def process_inputs(pairs):
    """处理 reranker 的输入"""
    inputs = reranker_tokenizer(
        pairs, padding=False, truncation='longest_first',
        return_attention_mask=False, max_length=max_reranker_length - len(prefix_tokens) - len(suffix_tokens)
    )
    for i, ele in enumerate(inputs['input_ids']):
        inputs['input_ids'][i] = prefix_tokens + ele + suffix_tokens
    inputs = reranker_tokenizer.pad(inputs, padding=True, return_tensors="pt", max_length=max_reranker_length)
    for key in inputs:
        inputs[key] = inputs[key].to(reranker_model.device)
    return inputs

@torch.no_grad()
def compute_logits(inputs, **kwargs):
    """使用 reranker 计算相关性得分"""
    batch_scores = reranker_model(**inputs).logits[:, -1, :]
    true_vector = batch_scores[:, token_true_id]
    false_vector = batch_scores[:, token_false_id]
    batch_scores = torch.stack([false_vector, true_vector], dim=1)
    batch_scores = torch.nn.functional.log_softmax(batch_scores, dim=1)
    scores = batch_scores[:, 1].exp().tolist()
    return scores

def rerank_documents(query, documents, task_instruction=None):
    """
    使用 Qwen3-Reranker 基于查询相关性对文档进行 reranking。

    Args:
        query: 搜索查询
        documents: 要 rerank 的文档列表
        task_instruction: reranking 的任务指令

    Returns:
        按相关性得分排序的 (document, score) 元组列表
    """
    if task_instruction is None:
        task_instruction = 'Given a web search query, retrieve relevant passages that answer the query'

    # 格式化 reranker 的输入
    pairs = [format_instruction(task_instruction, query, doc) for doc in documents]

    # 处理输入并计算得分
    inputs = process_inputs(pairs)
    scores = compute_logits(inputs)

    # 将文档与得分结合，并按得分降序排序
    doc_scores = list(zip(documents, scores))
    doc_scores.sort(key=lambda x: x[1], reverse=True)

    return doc_scores

Milvus 向量数据库：存储 Embedding，加速检索

现在我们设置向量数据库。为了简单起见，我们使用 Milvus Lite，但相同的代码也适用于完整的 Milvus 部署：

from pymilvus import MilvusClient

milvus_client = MilvusClient(uri="./milvus_demo.db")
collection_name = "my_rag_collection"

部署选项：

• 本地文件 (例如 ./milvus.db)：使用 Milvus Lite，非常适合开发。
• Docker/Kubernetes：使用服务器 URI，例如 http://localhost:19530，用于生产环境。
• Zilliz Cloud：使用云端点和 API 密钥获取托管服务。

清理任何现有集合并创建一个新集合：

# 如果存在，删除现有集合
if milvus_client.has_collection(collection_name):
    milvus_client.drop_collection(collection_name)

# 使用我们的 embedding 维度创建新集合
milvus_client.create_collection(
    collection_name=collection_name,
    dimension=embedding_dim,  # Qwen3-Embedding-0.6B 为 1024
    metric_type="IP",  # 内积用于相似性
    consistency_level="Strong",  # 确保数据一致性
)

数据加载：将文档向量化并存入 Milvus

现在我们处理文档并将它们插入到向量数据库中：

from tqdm import tqdm

data = []
for i, line in enumerate(tqdm(text_lines, desc="Creating embeddings")):
    data.append({"id": i, "vector": emb_text(line), "text": line})

milvus_client.insert(collection_name=collection_name, data=data)

预期输出：

Creating embeddings: 100%|████████████| 72/72 [00:08<00:00, 8.68it/s]
Inserted 72 documents

RAG 系统集成：检索、重排序与生成

现在到了激动人心的时刻——将所有内容组合成一个完整的检索增强生成系统。

步骤 1：查询和初始检索

让我们用一个关于 Milvus 的常见问题进行测试：

question = "How is data stored in milvus?"

# 执行初始密集检索以获得前几个候选文档
search_res = milvus_client.search(
    collection_name=collection_name,
    data=[emb_text(question, is_query=True)],  # 使用查询提示
    limit=10,  # 获取前 10 个候选文档进行 reranking
    search_params={"metric_type": "IP", "params": {}},
    output_fields=["text"],  # 返回实际文本内容
)

print(f"Found {len(search_res[0])} initial candidates")

步骤 2：重排序以提高精度

提取候选文档并应用 reranking：

# 提取候选文档
candidate_docs = [res["entity"]["text"] for res in search_res[0]]

# 使用 Qwen3-Reranker 进行 reranking
print("Reranking documents...")
reranked_docs = rerank_documents(question, candidate_docs)

# 选择 reranking 后的前 3 个文档
top_reranked_docs = reranked_docs[:3]
print(f"Selected top {len(top_reranked_docs)} documents after reranking")

步骤 3：对比结果

让我们检查一下 reranking 如何改变结果：

print("Reranked results (top 3):")
for doc, score in top_reranked_docs:
    print([doc, score])

print("================================================================================")

print("Original embedding-based results (top 3):")
original_docs = [(res["entity"]["text"], res["distance"]) for res in search_res[0]]
for doc, score in original_docs[:3]:
    print([doc, score])

Reranking 通常显示出更高的区分性得分（对于相关文档，得分更接近 1.0），而 embedding 相似性得分则相对较低。例如，在初始检索中排名靠后的文档，经过 reranking 后可能因为更强的相关性而排名靠前。

步骤 4：生成最终响应

现在，我们使用检索到的上下文来生成一个全面的答案：

# 首先：将检索到的文档转换为字符串格式。
context = "\n".join(
    [line_with_distance[0] for line_with_distance in top_reranked_docs]
)

# 为大型语言模型提供系统提示和用户提示。此提示是从 Milvus 检索到的文档生成的。
SYSTEM_PROMPT = """Human: You are an AI assistant. You are able to find answers to the questions from the contextual passage snippets provided."""
USER_PROMPT = f"""Use the following pieces of information enclosed in <context> tags to provide an answer to the question enclosed in <question> tags.<context>{context}</context><question>{question}</question>"""

# 使用 GPT-4o 根据提示生成响应。
response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": USER_PROMPT},
    ],
)

print(response.choices[0].message.content)

预期输出：

In Milvus, data is stored in two main forms: inserted data and metadata. Inserted data, which includes vector data, scalar data, and collection-specific schema, is stored in persistent storage as incremental logs. Milvus supports multiple object storage backends for this purpose, including MinIO, AWS S3, Google Cloud Storage, Azure Blob Storage, Alibaba Cloud OSS, and Tencent Cloud Object Storage. Metadata for Milvus is generated by its various modules and stored in etcd.

总结：Qwen3 在 RAG 中的价值

本文演示了一个使用 Qwen3 的 embedding 和 reranking 模型构建的完整 RAG 实现。关键要点如下：

• 两阶段检索（密集检索 + reranking）始终优于仅使用 embedding 的方法。
• 指令提示允许在不重新训练的情况下进行特定领域的调优。
• 多语言能力可以自然地工作，无需额外的复杂性。
• 使用 0.6B 模型可以进行本地部署。

Qwen3 系列在一个轻量级的开源包中提供了可靠的性能。虽然它不是革命性的，但它提供了增量改进和有用的功能，例如指令提示，这可以在生产系统中产生真正的差异。

建议针对你的特定数据和用例测试这些模型——最佳方案始终取决于你的内容、查询模式和性能要求。例如，在处理金融领域的文档时，你可以通过指令提示引导 embedding 模型更加关注财务术语和风险指标；在处理法律领域的文档时，你可以引导模型更加关注法律条文和判例法。这些细微的调整可以显著提高 RAG 系统的准确性和相关性。

此外，值得注意的是，尽管 Qwen3 取得了显著的进展，但它仍然存在一些局限性。例如，在处理非常复杂的查询或涉及多个知识领域的查询时，Qwen3 的性能可能会下降。未来的研究方向可以包括探索更高级的 reranking 技术，例如基于 Transformer 的 reranking 模型，以及开发更有效的指令提示方法，以进一步提高 RAG 系统的性能。