Qwen3 Embedding + Milvus 知识库基线：从可追踪切块到带引用回答

几十行代码就能把文档、向量数据库和大模型连成一个 RAG Demo，但企业知识库真正困难的部分不是“跑起来”，而是回答三个问题：为什么召回了这些内容？回答是否真的有依据？系统变更之后，质量到底变好还是变坏？

本文构建一个以 Qwen3 Embedding 和 Milvus Lite 为基础的知识库参考基线。重点不是追求组件数量，而是把数据处理、检索、重排、生成和评估拆开，让每一层都可以独立检查。本文提供的是可连接自有文档与模型服务的实现骨架，不宣称已经给出适用于所有企业数据的质量结果。

目标与边界

我们希望得到的不是一个聊天页面，而是一条可重复运行的流水线：

1. 从 Markdown 或纯文本读取文档；
2. 保留文档来源、标题和更新时间；
3. 使用结构化切块生成检索单元；
4. 用 Qwen3 Embedding 编码并写入 Milvus；
5. 召回较大的候选集，再进行重排；
6. 把最终上下文交给生成模型；
7. 使用固定问题集评估每次变更。

示例使用 Milvus Lite，适合本地开发和质量验证。大规模数据、并发服务、权限隔离和高可用需要切换到 Milvus 服务端部署，不能直接用 Lite 的结果推断生产性能。

环境准备

Qwen3 Embedding 有不同参数规模。本文以 0.6B 版本说明流程，它更容易在开发机上运行；批量建库仍建议使用 GPU。以下示例要求 Python 3.10 或更高版本。

Shell

python3 -m venv .venv
source .venv/bin/activate
pip install \
  "pymilvus[milvus_lite]>=2.5,<3" \
  "sentence-transformers>=2.7,<6" \
  "transformers>=4.51,<5" \
  "torch>=2.4,<3" \
  "openai>=1.60,<3"

锁定实际安装版本并保存：

Shell

pip freeze > requirements-lock.txt

模型接口和推荐提示词可能更新，运行前应查看 Qwen3 Embedding 模型卡 与当前版本说明。

第一步：先准备可追踪的文档

不要只保存一段 text。知识库至少应保留：

• document_id：稳定的文档标识；
• chunk_id：块标识；
• title：标题或章节；
• source：原始文件或 URL；
• updated_at：更新时间；
• text：检索内容；
• 权限或租户字段。

下面是一个面向 Markdown 的最小切块函数。它优先按二级标题切分，再对过长章节做窗口切分。

Python

import hashlib
import re
from dataclasses import dataclass


@dataclass
class Chunk:
    document_id: str
    chunk_id: str
    title: str
    text: str
    source: str
    updated_at: str
    access_scope: str


def split_markdown(
    source: str,
    content: str,
    updated_at: str,
    access_scope: str,
    document_id: str | None = None,
    max_chars: int = 1800,
):
    # 生产系统应从内容管理系统传入永久 document_id；这里只用来源地址派生。
    document_id = document_id or hashlib.sha256(source.encode()).hexdigest()[:16]
    sections = re.split(r"(?=^##\s+)", content, flags=re.MULTILINE)
    chunks = []
    chunk_occurrences = {}
    for section in sections:
        section = section.strip()
        if not section:
            continue
        first_line, *_ = section.splitlines()
        title = first_line.removeprefix("## ").strip()
        for window_index, start in enumerate(range(0, len(section), max_chars)):
            text = section[start : start + max_chars].strip()
            content_hash = hashlib.sha256(
                f"{title}\n{window_index}\n{text}".encode()
            ).hexdigest()[:16]
            occurrence = chunk_occurrences.get(content_hash, 0)
            chunk_occurrences[content_hash] = occurrence + 1
            chunk_hash = (
                content_hash
                if occurrence == 0
                else hashlib.sha256(
                    f"{content_hash}:{occurrence}".encode()
                ).hexdigest()[:16]
            )
            chunks.append(
                Chunk(
                    document_id=document_id,
                    chunk_id=f"{document_id}:{chunk_hash}",
                    title=title,
                    text=text,
                    source=source,
                    updated_at=updated_at,
                    access_scope=access_scope,
                )
            )
    return chunks

这只是基线。正式项目还要处理表格、代码块、列表、PDF 页码和标题继承。切块的优劣必须通过检索评估决定，而不是凭经验选一个固定长度。

第二步：区分查询和文档编码

部分检索模型对查询和文档使用不同提示词。Qwen3 Embedding 的 SentenceTransformer 接口可以为查询使用检索提示。

Python

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")


def encode_documents(texts: list[str]) -> list[list[float]]:
    vectors = model.encode(
        texts,
        normalize_embeddings=True,
        batch_size=16,
        show_progress_bar=True,
    )
    return vectors.tolist()


def encode_query(query: str) -> list[float]:
    vector = model.encode(
        [query],
        prompt_name="query",
        normalize_embeddings=True,
    )[0]
    return vector.tolist()

归一化之后可以使用余弦相似度。不要混用“未归一化向量 + 内积”和“归一化向量 + 余弦”的测试结果。

第三步：写入 Milvus Lite

Milvus Lite 把数据保存在本地文件中，适合复现教程。

Python

from pymilvus import MilvusClient

client = MilvusClient(uri="./knowledge_base.db")
collection = "knowledge_chunks"
dimension = model.get_sentence_embedding_dimension()

if client.has_collection(collection):
    client.drop_collection(collection)

client.create_collection(
    collection_name=collection,
    dimension=dimension,
    metric_type="COSINE",
    consistency_level="Strong",
)

with open("handbook.md", encoding="utf-8") as source_file:
    chunks = split_markdown(
        source="handbook.md",
        content=source_file.read(),
        updated_at="2026-07-28T00:00:00Z",
        access_scope="employees",
        document_id="employee-handbook",
    )
vectors = encode_documents([chunk.text for chunk in chunks])

client.insert(
    collection_name=collection,
    data=[
        {
            "id": index,
            "vector": vector,
            "document_id": chunk.document_id,
            "chunk_id": chunk.chunk_id,
            "title": chunk.title,
            "source": chunk.source,
            "updated_at": chunk.updated_at,
            "access_scope": chunk.access_scope,
            "text": chunk.text,
        }
        for index, (chunk, vector) in enumerate(zip(chunks, vectors))
    ],
)

生产系统应使用显式 schema，并根据权限、租户、更新时间等过滤需求建立标量索引。动态字段方便 Demo，但不应替代正式的数据建模。

第四步：召回候选，而不是直接生成

Embedding 检索的职责是高召回。先取 10～30 个候选，再由更昂贵的重排模型选择最终上下文。示例把调用方已经获准访问的 access_scope 作为必填参数，并用 JSON 编码生成安全的字符串字面量，在检索阶段完成过滤；生产系统还应把用户身份映射为完整的租户、角色和文档 ACL 条件。

Python

import json


def retrieve(query: str, access_scope: str, limit: int = 20):
    if not access_scope:
        raise ValueError("access_scope 不能为空")
    access_scope_literal = json.dumps(access_scope, ensure_ascii=False)
    result = client.search(
        collection_name=collection,
        data=[encode_query(query)],
        filter=f"access_scope == {access_scope_literal}",
        limit=limit,
        output_fields=[
            "document_id",
            "chunk_id",
            "title",
            "source",
            "updated_at",
            "access_scope",
            "text",
        ],
        search_params={"metric_type": "COSINE", "params": {}},
    )[0]
    return [
        {
            "score": hit["distance"],
            **hit["entity"],
        }
        for hit in result
    ]

如果文档中有大量产品型号、错误码和 API 名称，建议加入关键词或稀疏检索形成混合召回。纯向量检索并不擅长所有精确匹配问题。

第五步：用重排提高前几名质量

Qwen3 Reranker 可以对“查询—文档”对重新评分，但它比 Embedding 更慢，不应对全部文档运行。下面按官方模型卡的 yes/no 概率方法给出一个最小实现；CPU 可以运行，实际服务应加入批处理、显存预算与超时控制。

Python

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

reranker_name = "Qwen/Qwen3-Reranker-0.6B"
reranker_tokenizer = AutoTokenizer.from_pretrained(
    reranker_name,
    padding_side="left",
)
reranker = AutoModelForCausalLM.from_pretrained(reranker_name).eval()
reranker_device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
reranker = reranker.to(reranker_device)

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)
yes_token_id = reranker_tokenizer.convert_tokens_to_ids("yes")
no_token_id = reranker_tokenizer.convert_tokens_to_ids("no")
max_length = 8192


@torch.inference_mode()
def score_pairs(pairs: list[tuple[str, str]]) -> list[float]:
    task = "根据用户问题，判断文档是否包含能够支持回答的相关信息"
    texts = [
        f"<Instruct>: {task}\n<Query>: {query}\n<Document>: {document}"
        for query, document in pairs
    ]
    encoded = reranker_tokenizer(
        texts,
        padding=False,
        truncation="longest_first",
        max_length=max_length - len(prefix_tokens) - len(suffix_tokens),
        return_attention_mask=False,
    )
    for index, token_ids in enumerate(encoded["input_ids"]):
        encoded["input_ids"][index] = prefix_tokens + token_ids + suffix_tokens
    inputs = reranker_tokenizer.pad(encoded, padding=True, return_tensors="pt")
    inputs = {name: value.to(reranker_device) for name, value in inputs.items()}
    next_token_logits = reranker(**inputs).logits[:, -1, :]
    binary_logits = torch.stack(
        [next_token_logits[:, no_token_id], next_token_logits[:, yes_token_id]],
        dim=1,
    )
    return binary_logits.softmax(dim=1)[:, 1].cpu().tolist()


def rerank(query: str, candidates: list[dict]) -> list[dict]:
    if not candidates:
        return []
    scores = score_pairs([(query, item["text"]) for item in candidates])
    ranked = [
        {**candidate, "rerank_score": float(score)}
        for candidate, score in zip(candidates, scores)
    ]
    return sorted(ranked, key=lambda item: item["rerank_score"], reverse=True)

正式服务应把模型加载、批处理和显存配置封装在独立模块中，避免检索逻辑与推理框架耦合。模型卡更新时也要重新验证前后缀与判定 token。

第六步：生成回答时保留证据

最终提示词至少包含三个约束：只能根据上下文回答、证据不足时明确拒答、回答中返回来源标识。下面使用 OpenAI Python SDK 连接任意兼容的聊天模型服务：

Shell

export OPENAI_API_KEY="替换为实际密钥"
export OPENAI_BASE_URL="https://你的兼容服务/v1"
export CHAT_MODEL="替换为支持的模型名"

Python

import os
import re

from openai import OpenAI

llm = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL"),
)


def answer(query: str, access_scope: str, top_k: int = 5) -> dict:
    candidates = rerank(
        query,
        retrieve(query, access_scope=access_scope, limit=20),
    )[:top_k]
    if not candidates:
        return {
            "answer": "现有权限范围内没有检索到可用资料。",
            "sources": [],
        }
    context = "\n\n".join(
        f'[来源: {item["chunk_id"]}]\n{item["text"]}' for item in candidates
    )
    response = llm.chat.completions.create(
        model=os.environ["CHAT_MODEL"],
        temperature=0,
        messages=[
            {
                "role": "system",
                "content": (
                    "你是企业知识库助手。只能根据提供的上下文回答。"
                    "证据不足时回答‘现有资料无法确认’。"
                    "每个关键结论后必须标注 [来源: chunk_id]。"
                ),
            },
            {
                "role": "user",
                "content": f"问题：{query}\n\n上下文：\n{context}",
            },
        ],
    )
    answer_text = response.choices[0].message.content or ""
    allowed_sources = {item["chunk_id"] for item in candidates}
    cited_sources = set(re.findall(r"\[来源:\s*([^\]]+)\]", answer_text))
    unknown_sources = cited_sources - allowed_sources
    if unknown_sources:
        raise ValueError(f"回答引用了未提供的来源：{sorted(unknown_sources)}")
    return {
        "answer": answer_text,
        "sources": [
            {"chunk_id": item["chunk_id"], "source": item["source"]}
            for item in candidates
        ],
    }

这段校验只能发现“引用了不存在的块”，不能证明引用内容真的支持结论；后者仍要通过固定评估集和人工抽检判断。检索结果必须携带 chunk_id 和 source，否则无法进行任何引用追踪。

第七步：建立固定评估集

至少准备四类问题：

1. 能在单个块中直接回答的问题；
2. 需要组合多个块的问题；
3. 包含缩写、同义词或跨语言表达的问题；
4. 知识库中没有答案的问题。

检索阶段建议评估 Recall@5、Recall@20 和 MRR。生成阶段至少检查：

• 回答正确性；
• 上下文相关性；
• 引用是否支持结论；
• 无答案问题是否拒答；
• P50/P95 延迟与 Token 消耗。

不要只用同一个大模型同时生成答案和评判答案。模型评审可以提高效率，但应使用固定校验集并抽样人工复核。

四组最值得做的对照实验

切块大小

比较 300、600、1000 token 以及结构化切块。记录召回、重排后排名和上下文 Token 数。块越大不一定越好。

向量、混合与重排

至少比较：

• 纯向量 Top 5；
• 纯向量 Top 20 + 重排 Top 5；
• 混合召回 Top 20 + 重排 Top 5。

这样才能判断质量提升来自模型、召回方式还是重排。

权限与元数据过滤

构造两个用户只能访问不同文档的场景，确保过滤在检索阶段完成。生成后再删除内容无法防止敏感信息进入上下文。

增量更新与删除

更新原文后重新写入向量，检查旧块是否被删除、引用是否指向新版本。知识库最常见的生产问题之一，是搜索结果仍然命中过期内容。

从本地基线走向生产

上线前还需要补齐：

• 文档解析失败与重试；
• 增量更新和幂等写入；
• 权限与租户隔离；
• 检索、重排和生成的分层监控；
• 固定回归集与 CI 检查；
• 模型与提示词版本管理；
• 超时、限流和降级。

RAG 的质量不是一个模型决定的，而是数据、切块、检索、重排、生成与评估共同决定的。先建立一个能测量的基线，才能知道下一次优化是否真的有效。

参考资料

• Qwen3 Embedding 0.6B 模型卡
• Qwen3 Reranker 0.6B 模型卡
• Milvus Lite 文档
• 使用 Milvus 构建 RAG
• Zilliz：Qwen3 Embedding + Milvus 教程