推荐语

RAG系统分词难题？Milvus Analyzer帮你精准切分专有名词，避免"武汉市长江大桥"变"武汉市长"的尴尬！

核心内容：
1. RAG系统中分词错误的典型案例与危害
2. Milvus Analyzer的架构与核心组件解析
3. 不同场景下的Tokenizer选型策略与实战技巧

杨芳贤

53AI创始人/腾讯云(TVP)最具价值专家

analyzer如何避免大模型把《无线电法国别研究》理解成无线电，法国别研究？

正文开始前，我们先复习个RAG人与向量数据库er的噩梦

Milvus 宣称2.5 版本就已经引入了全文检索（Full-text Search）

结果你搭了个RAG后发现

地名、人名、专有词汇全！都！检！索！不！出！来！

比如，《鲁迅全集》中，能检索到“藤野先生”却检索不到“藤野”；做半导体术语，能搜“EUV”能搜“光刻机”就是搜不到“EUV光刻机”

那是embedding模型选错了？

向量检索阈值设得太高了？

还是Milvus垃圾？（绝对不可能！）

或许，最大的可能是你在分词这一步，就把 Analyzer 选错了

把武汉市长江大桥 分成了 武汉市长、江大桥

把霍格沃兹魔法学院 分成了 霍格沃兹魔、法学院

以及这样

（给所有广州朋友滑跪道歉）

当然，在分词环节，常见的问题除了过度分词之外，可能还会出现分词不足、语言不匹配等等情况。

而要解决这些问题，实现高效的全文检索，最重要的就是选对合适的 Analyzer。在文本处理中，Analyzer 可以将原始文本转换为结构化、可搜索的格式，它的选择直接关乎最终的查询质量。

那么，Analyzer是如何工作的？不同场景如何对其选型？我们又该如何将其落地生产实践？

本文将带来重点解读。

01

什么是 Milvus Analyzer？

一句话来说，Milvus Analyzer 是 Milvus 提供的文本预处理与分词工具，用来将原始文本拆解为 token，并对其进行标准化和清洗，从而更好地支持全文检索和 text match。

下面这张架构图展示了 Milvus Analyzer 的整体结构：

Analyzer 的核心组件有两个，Tokenizer（分词器）与Filter（过滤器）。它们共同将输入文本转换为词元（token），并对这些词元进行优化，以便为高效的索引和检索做好准备。

• Tokenizer（分词器）：负责把文本切分成基础的 token，例如按空格切分（Whitespace）、中文分词（Jieba）、多语言分词（ICU）等。

• Filter（过滤器）：对 token 进行特定的处理方法，Milvus 内置了丰富的 filter，例如统一大小写（Lowercase）、去掉标点（Removepunct）、停用词过滤（Stop）、词干提取（Stemmer）、正则匹配（Regex）等。Milvus 支持设置多个 filter 按顺序处理，可以满足复杂的 token 处理需求。

(1)Tokenizer

Tokenizer 是 Milvus Analyzer 的第一步处理工具，它的任务是将一段原始文本切分成更小的 token（词或子词）。不同语言、不同场景需要使用不同的 Tokenizer。Milvus 目前支持以下几类 Tokenizer：

在 Milvus 中，Tokenizer 是在创建 Collection 的 Schema 时配置的，具体是在定义 VARCHAR 字段时，通过 analyzer_params 指定。也就是说，Tokenizer 并不是一个单独的对象，而是绑定在字段级别的配置里，这样 Milvus 在插入数据时就会自动进行分词和预处理。

FieldSchema(    name="text",    dtype=DataType.VARCHAR,    max_length=512,    analyzer_params={"tokenizer": "standard"   # 这里配置 Tokenizer    })

（2）Filter

如果说 Tokenizer 是把文本切开的刀，那 Filter 就是刀后面的精修工序。在 Milvus Analyzer 中，Filter 的作用是对切分后的 token 进行进一步的标准化、清洗或改造，让最终的 token 更适合用于检索。

例如，统一大小写、去掉停用词（如 “the”、“and”）、去除标点、词干提取（如 running → run）等，都是典型的 Filter 工作。

Milvus 内置了多种常用的 Filter，可以满足大部分语言处理需求：

使用 Filter 的好处是，你可以根据业务场景灵活组合不同的清洗规则。例如在英文搜索中，常见的组合是 Lowercase + Stop + Stemmer，这样可以保证大小写统一、去掉无意义词汇，并把不同形态的词统一成词干。

在中文搜索中，通常会结合 Cncharonly + Stop，让分词结果更简洁、更精准。在 Milvus 中，Filter 与 Tokenizer 一样，是通过 analyzer_params 配置在 FieldSchema 里的。举个例子：

FieldSchema(    name="text",    dtype=DataType.VARCHAR,    max_length=512,    analyzer_params={        "tokenizer": "standard","filter": ["lowercase",            {"type": "stop", # Specifies the filter type as stop"stop_words": ["of", "to", "_english_"], # Defines custom stop words and includes the English stop word list            },            {"type": "stemmer",  # Specifies the filter type as stemmer"language": "english"            }],    })

02

Analyer 类型

合适的 Analyzer 可以让我们的检索变得更高效与低成本。为了满足不同场景的需求，Milvus 提供了三类 Analyzer：

• 内置（Built-in）的 Standard/English/Chinese 三种 Analyzer

• 基于用户自定义的 Tokenizer 和 Filter 组成的 Custom Analyzer

• 在多语言文档场景中非常实用的 Multi-language Analyzer。

（1）Built-in Analyzer

内置 Analyzer 是 Milvus 自带的标准配置，开箱即用，适合大多数常见场景。它们已经预定义好 Tokenizer 和 Filter 的组合：

如果你只是需要最常见的英文或中文搜索，可以直接使用内置 Analyzer，而无需额外配置。

这里需要注意一点，Standard Analyzer 默认是处理英文文档的，如果中文使用 Standard Analyzer，后续就会出现全文搜索没有结果的问题，社区里已经碰到不少朋友踩到这坑里。

（2）Multi-language Analyzer

在跨语言的文本库中，单一的分词器往往无法覆盖所有语种。为此，Milvus 提供了 Multi-language Analyzer，它会根据文本的语言自动选择合适的分词器。

不同语言使用的 Tokenizer 对照表：

这意味着，如果你的数据集同时包含英文、中文、日文、韩文甚至阿拉伯文，Milvus 可以在同一个字段里进行灵活处理，大大减少了手工预处理的复杂度。

（3）Custom Analyzer

如果内置或多语言 Analyzer 不能完全满足需求，Milvus 还支持用户自定义 Analyzer。你可以自由组合 Tokenizer 和 Filter，从而形成一个符合业务特点的 Analyzer。

例如：

FieldSchema(        name="text",        dtype=DataType.VARCHAR,        max_length=512,        analyzer_params={"tokenizer": "jieba",  "filter": ["cncharonly", "stop"]  # 自定义组合，比如中英混合语料中只搜中文，且去掉中文停用词，比如“的”、“了”、“在”        }    )

03

代码实践

下面我们通过 Python SDK 演示如何在 Milvus 中使用 Analyzer。我们会分别展示普通 Analyzer 和多语言 Analyzer 的用法。

本文使用的 Milvus 版本为 v2.6.1，Pymilvus 版本为 v2.6.1。

（1）普通 Analyzer

假设我们要建立一个英文文本搜索的 Collection，并在插入数据时自动完成分词和预处理。我们选用内置的 English Analyzer（相当于 standard + lowercase + stop + stemmer 的组合）。

from pymilvus import MilvusClient, DataType, Function, FunctionTypeclient = MilvusClient(    uri="http://localhost:19530",)schema = client.create_schema()schema.add_field(    field_name="id",                  # Field name    datatype=DataType.INT64,          # Integer data type    is_primary=True,                  # Designate as primary key    auto_id=True                      # Auto-generate IDs (recommended))schema.add_field(    field_name='text',    datatype=DataType.VARCHAR,    max_length=1000,    enable_analyzer=True,    analyzer_params={            "tokenizer": "standard",            "filter": [            "lowercase",            {            "type": "stop", # Specifies the filter type as stop            "stop_words": ["of", "to", "_english_"], # Defines custom stop words and includes the English stop word list            },            {                "type": "stemmer",  # Specifies the filter type as stemmer                "language": "english"            }],        },    enable_match=True,)schema.add_field(    field_name="sparse",                   # Field name    datatype=DataType.SPARSE_FLOAT_VECTOR  # Sparse vector data type)bm25_function = Function(    name="text_to_vector",            # Descriptive function name    function_type=FunctionType.BM25,  # Use BM25 algorithm    input_field_names=["text"],       # Process text from this field    output_field_names=["sparse"]     # Store vectors in this field)schema.add_function(bm25_function)index_params = client.prepare_index_params()index_params.add_index(    field_name="sparse",        # Field to index (our vector field)    index_type="AUTOINDEX",     # Let Milvus choose optimal index type    metric_type="BM25"          # Must be BM25 for this feature)COLLECTION_NAME = "english_demo"if client.has_collection(COLLECTION_NAME):    client.drop_collection(COLLECTION_NAME)      print(f"Dropped existing collection: {COLLECTION_NAME}")client.create_collection(    collection_name=COLLECTION_NAME,       # Collection name    schema=schema,                         # Our schema    index_params=index_params              # Our search index configuration)print(f"成功创建集合: {COLLECTION_NAME}")# 准备示例数据sample_texts = [    "The quick brown fox jumps over the lazy dog",    "Machine learning algorithms are revolutionizing artificial intelligence",      "Python programming language is widely used for data science projects",    "Natural language processing helps computers understand human languages",    "Deep learning models require large amounts of training data",    "Search engines use complex algorithms to rank web pages",    "Text analysis and information retrieval are important NLP tasks",    "Vector databases enable efficient similarity searches",    "Stemming reduces words to their root forms for better searching",    "Stop words like 'the', 'and', 'of' are often filtered out"]# 插入数据print("\n正在插入数据...")data = [{"text": text} for text in sample_texts]client.insert(    collection_name=COLLECTION_NAME,    data=data)print(f"成功插入 {len(sample_texts)} 条数据")# 演示分词器效果print("\n" + "="*60)print("分词器分析演示")print("="*60)test_text = "The running dogs are jumping over the lazy cats"print(f"\n原始文本: '{test_text}'")# 使用 run_analyzer 展示分词结果analyzer_result = client.run_analyzer(    texts=test_text,    collection_name=COLLECTION_NAME,    field_name="text")print(f"分词结果: {analyzer_result}")print("\n分析说明:")print("- lowercase: 将所有字母转换为小写")print("- stop words: 过滤掉停用词 ['of', 'to'] 和英语常见停用词")  print("- stemmer: 将词汇还原为词干形式 (running -> run, jumping -> jump)")# 全文检索演示print("\n" + "="*60)print("全文检索演示")print("="*60)# 等待数据索引完成import timetime.sleep(2)# 搜索查询示例search_queries = [    "jump",           # 测试词干匹配 (应该匹配 "jumps")    "algorithm",      # 测试精确匹配    "python program", # 测试多词查询    "learn"          # 测试词干匹配 (应该匹配 "learning")]for i, query in enumerate(search_queries, 1):    print(f"\n查询 {i}: '{query}'")    print("-" * 40)    # 执行全文检索    search_results = client.search(        collection_name=COLLECTION_NAME,        data=[query],                    # 查询文本        search_params={"metric_type": "BM25"},        output_fields=["text"],         # 返回原始文本        limit=3                         # 返回前3个结果    )    if search_results and len(search_results[0]) > 0:        for j, result in enumerate(search_results[0], 1):            score = result["distance"]            text = result["entity"]["text"]            print(f"  结果 {j} (相关度: {score:.4f}): {text}")    else:        print("  未找到相关结果")print("\n" + "="*60)print("检索完成！")print("="*60)
结果输出：

Dropped existing collection: english_demo成功创建集合: english_demo正在插入数据...成功插入 10 条数据============================================================分词器分析演示============================================================原始文本: 'The running dogs are jumping over the lazy cats'分词结果: ['run', 'dog', 'jump', 'over', 'lazi', 'cat']分析说明:- lowercase: 将所有字母转换为小写- stop words: 过滤掉停用词 ['of', 'to'] 和英语常见停用词- stemmer: 将词汇还原为词干形式 (running -> run, jumping -> jump)============================================================全文检索演示============================================================查询 1: 'jump'----------------------------------------  结果 1 (相关度: 2.0040): The quick brown fox jumps over the lazy dog查询 2: 'algorithm'----------------------------------------  结果 1 (相关度: 1.5819): Machine learning algorithms are revolutionizing artificial intelligence  结果 2 (相关度: 1.4086): Search engines use complex algorithms to rank web pages查询 3: 'python program'----------------------------------------  结果 1 (相关度: 3.7884): Python programming language is widely used for data science projects查询 4: 'learn'----------------------------------------  结果 1 (相关度: 1.5819): Machine learning algorithms are revolutionizing artificial intelligence  结果 2 (相关度: 1.4086): Deep learning models require large amounts of training data============================================================检索完成！============================================================
（2）多语言 Analyzer

如果数据集中同时包含多种语言，比如英文、中文和日文，那么我们就可以启用 Multi-language Analyzer。这样 Milvus 会根据文本语言自动选择合适的分词器。

from pymilvus import MilvusClient, DataType, Function, FunctionTypeimport time# 配置连接client = MilvusClient(    uri="http://localhost:19530",)COLLECTION_NAME = "multilingual_demo"# 删除已存在的集合if client.has_collection(COLLECTION_NAME):    client.drop_collection(COLLECTION_NAME)# 创建schemaschema = client.create_schema()# 添加主键字段schema.add_field(    field_name="id",    datatype=DataType.INT64,    is_primary=True,    auto_id=True)# 添加语言标识字段schema.add_field(    field_name="language",    datatype=DataType.VARCHAR,    max_length=50)# 添加文本字段，配置多语言分析器multi_analyzer_params = {    "by_field": "language",  # 根据language字段选择分析器    "analyzers": {        "en": {            "type": "english"  # 英语分析器        },        "zh": {            "type": "chinese"  # 中文分析器        },        "jp": {            "tokenizer": "icu",  # 日语使用ICU分词器            "filter": [                "lowercase",                {                    "type": "stop",                    "stop_words": ["は", "が", "の", "に", "を", "で", "と"]                }            ]        },        "default": {            "tokenizer": "icu"  # 默认使用ICU通用分词器        }    },    "alias": {        "english": "en",        "chinese": "zh",         "japanese": "jp",        "中文": "zh",        "英文": "en",        "日文": "jp"    }}schema.add_field(    field_name="text",    datatype=DataType.VARCHAR,    max_length=2000,    enable_analyzer=True,    multi_analyzer_params=multi_analyzer_params)# 添加稀疏向量字段用于BM25schema.add_field(    field_name="sparse_vector",    datatype=DataType.SPARSE_FLOAT_VECTOR)# 定义BM25函数bm25_function = Function(    name="text_bm25",    function_type=FunctionType.BM25,    input_field_names=["text"],    output_field_names=["sparse_vector"])schema.add_function(bm25_function)# 准备索引参数index_params = client.prepare_index_params()index_params.add_index(    field_name="sparse_vector",    index_type="AUTOINDEX",    metric_type="BM25")# 创建集合client.create_collection(    collection_name=COLLECTION_NAME,    schema=schema,    index_params=index_params)# 准备多语言测试数据multilingual_data = [    # 英文数据    {"language": "en", "text": "Artificial intelligence is revolutionizing technology industries worldwide"},    {"language": "en", "text": "Machine learning algorithms process large datasets efficiently"},    {"language": "en", "text": "Vector databases provide fast similarity search capabilities"},    # 中文数据      {"language": "zh", "text": "人工智能正在改变世界各行各业"},    {"language": "zh", "text": "机器学习算法能够高效处理大规模数据集"},    {"language": "zh", "text": "向量数据库提供快速的相似性搜索功能"},    # 日文数据    {"language": "jp", "text": "人工知能は世界中の技術産業に革命をもたらしています"},    {"language": "jp", "text": "機械学習アルゴリズムは大量のデータセットを効率的に処理します"},    {"language": "jp", "text": "ベクトルデータベースは高速な類似性検索機能を提供します"},]client.insert(    collection_name=COLLECTION_NAME,    data=multilingual_data)# 等待BM25函数生成向量print("等待BM25向量生成...")client.flush(COLLECTION_NAME)time.sleep(5)client.load_collection(COLLECTION_NAME)# 演示分词器效果print("\n分词器分析:")test_texts = {    "en": "The running algorithms are processing data efficiently",    "zh": "这些运行中的算法正在高效地处理数据",     "jp": "これらの実行中のアルゴリズムは効率的にデータを処理しています"}for lang, text in test_texts.items():    print(f"{lang}: {text}")    try:        analyzer_result = client.run_analyzer(            texts=text,            collection_name=COLLECTION_NAME,            field_name="text",            analyzer_names=[lang]        )        print(f"  → {analyzer_result}")    except Exception as e:        print(f"  → 分析失败: {e}")# 多语言检索演示print("\n检索测试:")search_cases = [    ("zh", "人工智能"),    ("jp", "機械学習"),    ("en", "algorithm"),]for lang, query in search_cases:    print(f"\n{lang} '{query}':")    try:        search_results = client.search(            collection_name=COLLECTION_NAME,            data=[query],            search_params={"metric_type": "BM25"},            output_fields=["language", "text"],            limit=3,            filter=f'language == "{lang}"'        )        if search_results and len(search_results[0]) > 0:            for result in search_results[0]:                score = result["distance"]                text = result["entity"]["text"]                print(f"  {score:.3f}: {text}")        else:            print("  无结果")    except Exception as e:        print(f"  错误: {e}")print("\n完成")
结果输出：

等待BM25向量生成...分词器分析:en: The running algorithms are processing data efficiently  → ['run', 'algorithm', 'process', 'data', 'effici']zh: 这些运行中的算法正在高效地处理数据  → ['这些', '运行', '中', '的', '算法', '正在', '高效', '地', '处理', '数据']jp: これらの実行中のアルゴリズムは効率的にデータを処理しています  → ['これらの', '実行', '中の', 'アルゴリズム', '効率', '的', 'データ', '処理', 'し', 'てい', 'ます']检索测试:zh '人工智能':  3.300: 人工智能正在改变世界各行各业jp '機械学習':  3.649: 機械学習アルゴリズムは大量のデータセットを効率的に処理しますen 'algorithm':  2.096: Machine learning algorithms process large datasets efficiently完成

另外，Milvus 目前也支持使用 language_identifier 分词器进行搜索，它的好处是不用手动告诉系统这段文本是什么语言，Milvus 会自己识别。相应的，语言字段（language）也不是必须的。

社区之前的官宣：Milvus 2.6引入多语言分析器，全文搜索再升级，助力业务全球化博客中已经做了详细介绍，这里就不再赘述。