背景知识：词库的作用

IK 分词器是一款基于词典匹配的中文分词器，其准确性和召回率与 IK 使用的词库也有不小的关系。

这里我们先了解一下词典匹配法的作用流程：

1. 预先准备一个大规模的词典，用算法在文本中寻找词典里的最长匹配项。这种方法实现简单且速度快。
2. 但面临歧义切分和未登录词挑战：同一序列可能有不同切分方式（例如“北京大学生”可以切成“北京大学/生”或“北京/大学生”），需要规则或算法消除歧义；
3. 而词典中没有的新词（如网络流行语、人名等）无法正确切分。

可以看到词库是词元产生的比对基础，一个完善的中文词库能大大提高分词器的准确性和召回率。

IK 使用的词库是中文中常见词汇的合集，完善且丰富，ik_smart 和 ik_max_word 也能满足大部分中文分词的场景需求。但是针对一些专业的场景，比如医药这样的行业词库、电商搜索词、新闻热点词等，IK 是很难覆盖到的。这时候就需要使用者自己去维护自定义的词库了。

IK 的自定义词库加载方式

IK 本身也支持自定义词库的加载和更新的，但是只支持一个集群使用一个词库。

这里主要的制约因素是，词库对象与 ik 的中文分词器执行对象是一一对应的关系。

这导致了 IK 的词库面对不同中文分词场景时较低的灵活性，使用者并不能做到字段级别的词库加载。并且基于文件或者 http 协议的词库加载方式也需要不小的维护成本。

字段级别词库的加载

鉴于上述的背景问题，INFINI lab 加强了 IK 的词库加载逻辑，做到了字段级别的词库加载。同时将自定义词库的加载方式由外部文件/远程访问改成了内部索引查询。

主要逻辑如图：

这里 IK 多中文词库的加载优化主要基于 IK 可以加载多词类对象（即下面这段代码）的灵活性，将原来遍历一个 CJK 词类对象修改成遍历多个 CJK 词类对象，各个自定义词库可以附着在 CJK 词库对象上实现不同词库的分词。

do{
    //遍历子分词器
    for(ISegmenter segmenter : segmenters){
        segmenter.analyze(context);
    }
    //字符缓冲区接近读完，需要读入新的字符
    if(context.needRefillBuffer()){
        break;
    }
}

对默认词库的新增支持

对于默认词库的修改，新版 IK 也可以通过写入词库索引方式支持，只要将 dict_key 设置为 default 即可。

POST .analysis_ik/_doc
{
  "dict_key": "default",
  "dict_type": "main_dicts",
  "dict_content":"杨树林"
}

效率测试

测试方案 1：单条测试

测试方法：写入一条数据到默认 ik_max_word 和自定义词库，查看是否有明显的效率差距

1. 创建测试索引，自定义一个包括默认词库的 IK 分词器

PUT my-index-000001
{
  "settings": {
    "number_of_shards": 3,
    "analysis": {
      "analyzer": {
        "my_custom_analyzer": {
          "type": "custom",
          "tokenizer": "my_tokenizer"
        }
      },
      "tokenizer": {
        "my_tokenizer": {

          "type": "ik_max_word",
          "custom_dict_enable": true,
          "load_default_dicts":true,
          "lowcase_enable": true,
          "dict_key": "test_dic"
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "test_ik": {
        "type": "text",
        "analyzer": "my_custom_analyzer"
      }
    }
  }
}

2. 将该词库重复默认词库的内容

POST .analysis_ik/_doc
{
  "dict_key": "test_dic",
  "dict_type": "main_dicts",
  "dict_content":"""xxxx  #词库内容
  """
}

# debug 日志
[2025-07-09T16:37:43,112][INFO ][o.w.a.d.Dictionary       ] [ik-1] Loaded 275909 words from main_dicts dictionary for dict_key: test_dic

3. 测试默认词库和自定义词库的分词效率

GET my-index-000001/_analyze
{
  "analyzer": "my_custom_analyzer",
  "text":"自强不息，杨树林"
}

GET my-index-000001/_analyze
{
  "analyzer": "ik_max_word",
  "text":"自强不息，杨树林"
}

打开 debug 日志，可以看到自定义分词器在不同的词库找到了 2 次“自强不息”

...
[2025-07-09T16:52:22,937][INFO ][o.w.a.c.CN_QuantifierSegmenter] [ik-1] 当前扫描词元[息]不需要启动量词扫描
[2025-07-09T16:52:22,937][INFO ][o.w.a.c.CJKSegmenter     ] [ik-1] >>> WORD FOUND [自强不息] from dict [default]
[2025-07-09T16:52:22,937][INFO ][o.w.a.c.CJKSegmenter     ] [ik-1] >>> WORD FOUND [不息] from dict [default]
[2025-07-09T16:52:22,937][INFO ][o.w.a.c.CJKSegmenter     ] [ik-1] >>> WORD FOUND [自强不息] from dict [test_dic]
[2025-07-09T16:52:22,937][INFO ][o.w.a.c.CJKSegmenter     ] [ik-1] >>> WORD FOUND [不息] from dict [test_dic]
[2025-07-09T16:52:22,937][INFO ][o.w.a.c.CN_QuantifierSegmenter] [ik-1] 当前扫描词元[,]不需要启动量词扫描
...

而默认词库只有一次

...
[2025-07-09T16:54:22,618][INFO ][o.w.a.c.CN_QuantifierSegmenter] [ik-1] 当前扫描词元[息]不需要启动量词扫描
[2025-07-09T16:54:22,618][INFO ][o.w.a.c.CJKSegmenter     ] [ik-1] >>> WORD FOUND [自强不息] from dict [default]
[2025-07-09T16:54:22,618][INFO ][o.w.a.c.CJKSegmenter     ] [ik-1] >>> WORD FOUND [不息] from dict [default]
[2025-07-09T16:54:22,618][INFO ][o.w.a.c.CN_QuantifierSegmenter] [ik-1] 当前扫描词元[,]不需要启动量词扫描
...

测试方案 2:持续写入测试

测试方法：在 ik_max_word 和自定义词库的索引里，分别持续 bulk 写入，查看总体写入延迟。

测试索引：

# ik_max_word索引
PUT ik_max_test
{
  "mappings": {
    "properties": {
      "chapter": {
        "type": "keyword"
      },
      "content": {
        "type": "text",
        "analyzer": "ik_max_word"
      },
      "paragraph_id": {
        "type": "keyword"
      },
      "random_field": {
        "type": "text"
      },
      "timestamp": {
        "type": "keyword"
      },
      "word_count": {
        "type": "integer"
      }
    }
  },
  "settings": {
    "index": {
      "number_of_shards": "1",
      "number_of_replicas": "0"
    }
  }
}

# 自定义词库索引
PUT ik_custom_test
{
  "mappings": {
    "properties": {
      "chapter": {
        "type": "keyword"
      },
      "content": {
        "type": "text",
        "analyzer": "my_custom_analyzer"
      },
      "paragraph_id": {
        "type": "keyword"
      },
      "random_field": {
        "type": "text"
      },
      "timestamp": {
        "type": "keyword"
      },
      "word_count": {
        "type": "integer"
      }
    }
  },
  "settings": {
    "index": {
      "number_of_shards": "1",
      "analysis": {
        "analyzer": {
          "my_custom_analyzer": {
            "type": "custom",
            "tokenizer": "my_tokenizer"
          }
        },
        "tokenizer": {
          "my_tokenizer": {
            "load_default_dicts": "true",
            "type": "ik_max_word",
            "dict_key": "test_dic",
            "lowcase_enable": "true",
            "custom_dict_enable": "true"
          }
        }
      },
      "number_of_replicas": "0"
    }
  }
}

这里利用脚本循环写入了一段《四世同堂》的文本，比较相同次数下，两次写入的总体延迟。

测试脚本内容如下：

#!/usr/bin/env python3

# -_- coding: utf-8 -_-

"""
四世同堂中文内容随机循环写入 Elasticsearch 脚本
目标：生成指定 bulk 次数的索引内容
"""

import random
import time
import json
from datetime import datetime
import requests
import logging
import os
import argparse
import urllib3

# 配置日志

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(**name**)

class ESDataGenerator:
def **init**(self, es_host='localhost', es_port=9200, index_name='sisitontang_content',
target_bulk_count=10000, batch_size=1000, use_https=False, username=None, password=None, verify_ssl=True):
"""
初始化 ES 连接和配置
"""
protocol = 'https' if use_https else 'http'
self.es_url = f'{protocol}://{es_host}:{es_port}'
self.index_name = index_name
self.target_bulk_count = target_bulk_count # 目标 bulk 次数
self.batch_size = batch_size
self.check_interval = 1000 # 每 1000 次 bulk 检查一次进度

        # 设置认证信息
        self.auth = None
        if username and password:
            self.auth = (username, password)
            logger.info(f"使用用户名认证: {username}")

        # 设置请求会话
        self.session = requests.Session()
        if self.auth:
            self.session.auth = self.auth

        # 处理HTTPS和SSL证书验证
        if use_https:
            self.session.verify = False  # 始终禁用SSL验证以避免证书问题
            logger.info("警告：已禁用SSL证书验证（适合开发测试环境）")
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

            # 设置SSL适配器以处理连接问题
            from requests.adapters import HTTPAdapter
            from urllib3.util.retry import Retry

            # 配置重试策略
            retry_strategy = Retry(
                total=3,
                backoff_factor=1,
                status_forcelist=[429, 500, 502, 503, 504],
            )

            adapter = HTTPAdapter(max_retries=retry_strategy)
            self.session.mount("https://", adapter)

            # 设置更宽松的SSL上下文
            self.session.verify = False

        logger.info(f"ES连接地址: {self.es_url}")

        # 创建索引映射
        self.create_index()

    def create_index(self):
        """创建索引和映射"""
        mapping = {
            "mappings": {
                "properties": {
                    "chapter": {"type": "keyword"},
                    "content": {"type": "text", "analyzer": "ik_max_word"},
                    "timestamp": {"type": "date"},
                    "word_count": {"type": "integer"},
                    "paragraph_id": {"type": "keyword"},
                    "random_field": {"type": "text"}
                }
            }
        }

        try:
            # 检查索引是否存在
            response = self.session.head(f"{self.es_url}/{self.index_name}")
            if response.status_code == 200:
                logger.info(f"索引 {self.index_name} 已存在")
            else:
                # 创建索引
                response = self.session.put(
                    f"{self.es_url}/{self.index_name}",
                    headers={'Content-Type': 'application/json'},
                    json=mapping
                )
                if response.status_code in [200, 201]:
                    logger.info(f"创建索引 {self.index_name} 成功")
                else:
                    logger.error(f"创建索引失败: {response.status_code} - {response.text}")
        except Exception as e:
            logger.error(f"创建索引失败: {e}")

    def load_text_content(self, file_path='sisitontang.txt'):
        """
        从文件加载《四世同堂》的完整文本内容
        如果文件不存在，则返回扩展的示例内容
        """
        if os.path.exists(file_path):
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                    logger.info(f"从文件 {file_path} 加载了 {len(content)} 个字符的文本内容")
                    return content
            except Exception as e:
                logger.error(f"读取文件失败: {e}")

        # 如果文件不存在，返回扩展的示例内容
        logger.info("使用内置的扩展示例内容")
        return self.get_extended_sample_content()

    def get_extended_sample_content(self):
        """
        获取扩展的《四世同堂》示例内容
        """
        content = """
        小羊圈胡同是北平城里的一个小胡同。它不宽，可是很长，从东到西有一里多路。在这条胡同里，从东边数起，有个小茶馆，几个小门脸，和一群小房屋。小茶馆的斜对面是个较大的四合院，院子里有几棵大槐树。这个院子就是祁家的住所，四世同堂的大家庭就在这里度过了最困难的岁月。

        祁老人是个善良的老头儿，虽然年纪大了，可是还很有精神。他的一生见证了太多的变迁，从清朝的衰落到民国的建立，再到现在的战乱，他都以一种达观的态度面对着。他的儿子祁天佑是个教书先生，为人正直，在胡同里很有威望。祁家的儿媳妇韵梅是个贤惠的女人，把家里打理得井井有条，即使在最困难的时候，也要维持着家庭的尊严。

        钱默吟先生是个有学问的人，他的诗写得很好，可是性格有些古怪。他住在胡同深处的一个小院子里，平时很少出门，只是偶尔到祁家坐坐，和祁天佑聊聊古今。他对时局有着自己独特的见解，但更多的时候，他选择在自己的小天地里寻找精神的慰藉。战争的残酷现实让这个文人感到深深的无力，但他依然坚持着自己的文人气节。

        小顺子是个活泼的孩子，他每天都在胡同里跑来跑去，和其他的孩子们一起玩耍。他的笑声总是能感染到周围的人，让这个古老的胡同充满了生机。即使在战争的阴霾下，孩子们依然保持着他们的天真和快乐，这或许就是生活的希望所在。小顺子不懂得大人们的烦恼，他只是简单地享受着童年的快乐。

        李四大爷是个老实人，他在胡同里开了个小杂货铺。虽然生意不大，但是童叟无欺，街坊邻居们都愿意到他这里买东西。他的妻子是个能干的女人，把小铺子管理得很好。在那个物资匮乏的年代，能够维持一个小铺子的经营已经很不容易了。李四大爷经常帮助邻居们，即使自己的生活也不宽裕。

        胡同里的生活是平静的，每天清晨，人们就开始忙碌起来。有的人挑着水桶去井边打水，有的人牵着羊去街上卖奶，有的人挑着菜担子去菜市场。这种平静的生活在战争来临之前是那么的珍贵，人们都珍惜着这样的日子。邻里之间相互照顾，孩子们在院子里玩耍，老人们在门口晒太阳聊天。

        冠晓荷是个复杂的人物，他有文化，也有野心。在日本人占领北平的时候，他选择了与敌人合作，这让胡同里的人们都看不起他。但是他的妻子还是个好人，只是被丈夫连累了。冠晓荷的选择代表了那个时代一部分知识分子的软弱和妥协，他们在民族大义和个人利益之间选择了后者。

        春天来了，胡同里的槐树发芽了，小鸟们在枝头歌唱。孩子们在院子里玩耍，老人们在门口晒太阳。这样的日子让人感到温暖和希望。即使在最黑暗的时期，生活依然要继续，人们依然要保持对美好未来的希望。春天的到来总是能够给人们带来新的希望和力量。

        战争的阴云笼罩着整个城市，胡同里的人们也感受到了压力。有的人选择了抗争，有的人选择了妥协，有的人选择了逃避。每个人都在用自己的方式应对这个艰难的时代。祁瑞宣面临着痛苦的选择，他既不愿意与日本人合作，也不敢公开反抗，这种内心的煎熬让他备受折磨。

        老舍先生用他细腻的笔触描绘了胡同里的众生相，每个人物都有自己的特点和命运。他们的喜怒哀乐构成了这部伟大作品的丰富内涵。从祁老爷子的达观，到祁瑞宣的痛苦，从韵梅的坚强，到冠晓荷的堕落，每个人物都是那个时代的缩影。

        在那个动荡的年代，普通人的生活是不容易的。他们要面对战争的威胁，要面对生活的困难，要面对道德的选择。但是他们依然坚强地活着，为了家人，为了希望。即使在最困难的时候，人们依然保持着对美好生活的向往。

        胡同里的邻里关系是复杂的，有友好的，也有矛盾的。但是在大的困难面前，大家还是会相互帮助。这种邻里之间的温情是中华民族传统文化的重要组成部分。在那个特殊的年代，这种人与人之间的温情显得更加珍贵。

        祁瑞宣是个有理想的青年，他受过良好的教育，有自己的抱负。但是在日本人占领期间，他的理想和现实之间产生了尖锐的矛盾。他不愿意做汉奸，但是也不能完全抵抗。这种内心的矛盾和痛苦是那个时代很多知识分子的真实写照。

        小妞子是个可爱的孩子，她的天真无邪给这个沉重的故事增添了一丝亮色。她不懂得大人们的复杂心理，只是简单地生活着，快乐着。孩子们的天真和快乐在那个黑暗的年代显得格外珍贵，它们代表着生活的希望和未来。

        程长顺是个朴实的人，他没有什么文化，但是有自己的原则和底线。他不愿意向日本人低头，宁愿过艰苦的生活也要保持自己的尊严。他的坚持代表了中国人民不屈不挠的精神，即使在最困难的时候也不愿意妥协。

        胡同里的生活节奏是缓慢的，人们有时间去观察周围的变化，去思考生活的意义。这种慢节奏的生活在今天看来是珍贵的，它让人们有机会去体验生活的细节。在那个年代，即使生活艰难，人们依然能够从平凡的日常中找到乐趣。

        老二是个有个性的人，他不愿意受约束，喜欢自由自在的生活。但是在战争年代，这种个性给他带来了麻烦，也给家人带来了担忧。他的反叛精神在某种程度上代表着年轻一代对传统束缚的反抗，但在那个特殊的时代，这种反抗往往会带来意想不到的后果。

        胡同里的四合院是北京传统建筑的代表，它们见证了一代又一代人的生活。每个院子里都有自己的故事，每个房间里都有自己的记忆。这些古老的建筑承载着深厚的历史文化底蕴，即使在战争的破坏下，依然坚强地屹立着。

        在《四世同堂》这部作品中，老舍先生不仅描绘了个人的命运，也反映了整个民族的命运。小胡同里的故事其实就是大中国的缩影。每个人物的遭遇都代表着那个时代某一类人的命运，他们的选择和结局反映了整个民族在那个特殊历史时期的精神状态。

        战争结束了，但是人们心中的创伤需要时间来愈合。胡同里的人们重新开始了正常的生活，但是那段艰难的经历永远不会被忘记。历史的教训提醒着人们珍惜和平，珍惜现在的美好生活。四世同堂的故事将永远流传下去，成为后人了解那个时代的重要窗口。
        """
        return content.strip()

    def split_text_randomly(self, text, min_length=100, max_length=200):
        """
        将文本按100-200字的随机长度进行分割
        """
        # 清理文本，移除多余的空白字符
        text = ''.join(text.split())

        segments = []
        start = 0

        while start < len(text):
            # 随机选择段落长度
            segment_length = random.randint(min_length, max_length)
            end = min(start + segment_length, len(text))

            segment = text[start:end]
            if segment.strip():  # 确保段落不为空
                segments.append(segment.strip())

            start = end

        return segments

    def generate_random_content(self, base_content):
        """
        基于基础内容生成随机变化的内容
        """
        # 随机选择一个基础段落
        base_paragraph = random.choice(base_content)

        # 随机添加一些变化
        variations = [
            "在那个年代，",
            "据说，",
            "人们常常说，",
            "老一辈人总是提到，",
            "历史记录显示，",
            "根据回忆，",
            "有人说，",
            "大家都知道，",
            "传说中，",
            "众所周知，"
        ]

        endings = [
            "这就是当时的情况。",
            "这样的事情在那个年代很常见。",
            "这个故事至今还在流传。",
            "这是一个值得回忆的故事。",
            "这样的经历让人难以忘怀。",
            "这就是老北京的生活。",
            "这种精神值得我们学习。",
            "这个时代已经过去了。",
            "这样的生活现在已经很难看到了。",
            "这是历史的见证。"
        ]

        # 随机组合内容
        if random.random() < 0.3:
            content = random.choice(variations) + base_paragraph
        else:
            content = base_paragraph

        if random.random() < 0.3:
            content += random.choice(endings)

        return content

    def generate_document(self, text_segments, doc_id):
        """基于文本段落生成一个文档"""
        # 随机选择一个文本段落
        content = random.choice(text_segments)

        # 生成随机的额外字段以增加文档大小
        random_field = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=random.randint(100, 500)))

        doc = {
            "chapter": f"第{random.randint(1, 100)}章",
            "content": content,
            "timestamp": datetime.now(),
            "word_count": len(content),
            "paragraph_id": f"para_{doc_id}",
            "random_field": random_field
        }

        return doc

    def get_index_size_gb(self):
        """获取索引大小（GB）"""
        try:
            response = self.session.get(f"{self.es_url}/_cat/indices/{self.index_name}?bytes=b&h=store.size&format=json")
            if response.status_code == 200:
                data = response.json()
                if data and len(data) > 0:
                    size_bytes = int(data[0]['store.size'])
                    size_gb = size_bytes / (1024 * 1024 * 1024)
                    return size_gb
            return 0
        except Exception as e:
            logger.error(f"获取索引大小失败: {e}")
            return 0

    def bulk_insert(self, documents):
        """批量插入文档使用HTTP bulk API"""
        # 构建bulk请求体
        bulk_data = []
        for doc in documents:
            # 添加action行
            action = {"index": {"_index": self.index_name}}
            bulk_data.append(json.dumps(action))
            # 添加文档行
            bulk_data.append(json.dumps(doc, ensure_ascii=False, default=str))

        # 每行以换行符结束，最后也要有换行符
        bulk_body = '\n'.join(bulk_data) + '\n'

        try:
            response = self.session.post(
                f"{self.es_url}/_bulk",
                headers={'Content-Type': 'application/x-ndjson'},
                data=bulk_body.encode('utf-8'),
                timeout=30  # 添加超时设置
            )

            if response.status_code == 200:
                result = response.json()
                # 检查是否有错误
                if result.get('errors'):
                    error_count = 0
                    error_details = []
                    for item in result['items']:
                        if 'error' in item.get('index', {}):
                            error_count += 1
                            error_info = item['index']['error']
                            error_details.append(f"类型: {error_info.get('type')}, 原因: {error_info.get('reason')}")

                    if error_count > 0:
                        logger.warning(f"批量插入有 {error_count} 个错误")
                        # 打印前5个错误的详细信息
                        for i, error in enumerate(error_details[:5]):
                            logger.error(f"错误 {i+1}: {error}")
                        if len(error_details) > 5:
                            logger.error(f"... 还有 {len(error_details)-5} 个类似错误")
                return True
            else:
                logger.error(f"批量插入失败: HTTP {response.status_code} - {response.text}")
                return False
        except requests.exceptions.SSLError as e:
            logger.error(f"SSL连接错误: {e}")
            logger.error("建议检查ES集群的SSL配置或使用 --no-verify-ssl 参数")
            return False
        except requests.exceptions.ConnectionError as e:
            logger.error(f"连接错误: {e}")
            logger.error("请检查ES集群地址和端口是否正确")
            return False
        except requests.exceptions.Timeout as e:
            logger.error(f"请求超时: {e}")
            logger.error("ES集群响应超时，可能负载过高")
            return False
        except Exception as e:
            logger.error(f"批量插入失败: {e}")
            logger.error(f"错误类型: {type(e).__name__}")
            return False

    def run(self):
        """运行数据生成器"""
        start_time = time.time()
        start_datetime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        logger.info(f"开始生成数据，开始时间: {start_datetime}，目标bulk次数: {self.target_bulk_count}")

        # 加载文本内容
        text_content = self.load_text_content()

        # 将文本分割成100-200字的段落
        text_segments = self.split_text_randomly(text_content, min_length=100, max_length=200)
        logger.info(f"分割出 {len(text_segments)} 个文本段落")

        doc_count = 0
        bulk_count = 0
        bulk_times = []  # 记录每次bulk的耗时

        while bulk_count < self.target_bulk_count:
            # 生成批量文档
            documents = []
            for i in range(self.batch_size):
                doc = self.generate_document(text_segments, doc_count + i)
                documents.append(doc)

            # 记录单次bulk开始时间
            bulk_start = time.time()

            # 批量插入
            if self.bulk_insert(documents):
                bulk_end = time.time()
                bulk_duration = bulk_end - bulk_start
                bulk_times.append(bulk_duration)

                doc_count += self.batch_size
                bulk_count += 1

                # 定期检查和报告进度
                if bulk_count % self.check_interval == 0:
                    current_size = self.get_index_size_gb()
                    avg_bulk_time = sum(bulk_times[-self.check_interval:]) / len(bulk_times[-self.check_interval:])
                    logger.info(f"已完成 {bulk_count} 次bulk操作，插入 {doc_count} 条文档，当前索引大小: {current_size:.2f}GB，最近{self.check_interval}次bulk平均耗时: {avg_bulk_time:.3f}秒")

            # 避免过于频繁的插入
            #time.sleep(0.01)  # 减少延迟，提高测试速度

        end_time = time.time()
        end_datetime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        total_duration = end_time - start_time

        # 计算统计信息
        final_size = self.get_index_size_gb()
        avg_bulk_time = sum(bulk_times) / len(bulk_times) if bulk_times else 0
        total_docs_per_sec = doc_count / total_duration if total_duration > 0 else 0
        bulk_per_sec = bulk_count / total_duration if total_duration > 0 else 0

        logger.info(f"数据生成完成！")
        logger.info(f"开始时间: {start_datetime}")
        logger.info(f"结束时间: {end_datetime}")
        logger.info(f"总耗时: {total_duration:.2f}秒 ({total_duration/60:.2f}分钟)")
        logger.info(f"总计完成: {bulk_count} 次bulk操作")
        logger.info(f"总计插入: {doc_count} 条文档")
        logger.info(f"最终索引大小: {final_size:.2f}GB")
        logger.info(f"平均每次bulk耗时: {avg_bulk_time:.3f}秒")
        logger.info(f"平均bulk速率: {bulk_per_sec:.2f}次/秒")
        logger.info(f"平均文档写入速率: {total_docs_per_sec:.0f}条/秒")

def main():
"""主函数"""
parser = argparse.ArgumentParser(description='四世同堂中文内容写入 Elasticsearch 脚本')
parser.add_argument('--host', default='localhost', help='ES 主机地址 (默认: localhost)')
parser.add_argument('--port', type=int, default=9200, help='ES 端口 (默认: 9200)')
parser.add_argument('--index', required=True, help='索引名称 (必填)')
parser.add_argument('--bulk-count', type=int, default=1000, help='目标 bulk 次数 (默认: 10000)')
parser.add_argument('--batch-size', type=int, default=1000, help='每次 bulk 的文档数量 (默认: 1000)')
parser.add_argument('--https', action='store_true', help='使用 HTTPS 协议')
parser.add_argument('--username', help='ES 用户名')
parser.add_argument('--password', help='ES 密码')
parser.add_argument('--no-verify-ssl', action='store_true', help='禁用 SSL 证书验证（默认已禁用）')

    args = parser.parse_args()

    protocol = "HTTPS" if args.https else "HTTP"
    auth_info = f"认证: {args.username}" if args.username else "无认证"
    ssl_info = "禁用SSL验证" if args.https else ""
    logger.info(f"开始运行脚本，参数: {protocol}://{args.host}:{args.port}, 索引={args.index}, bulk次数={args.bulk_count}, {auth_info} {ssl_info}")

    try:
        generator = ESDataGenerator(
            args.host,
            args.port,
            args.index,
            args.bulk_count,
            args.batch_size,
            args.https,
            args.username,
            args.password,
            not args.no_verify_ssl  # 传入verify_ssl参数，但实际上总是False
        )
        generator.run()
    except KeyboardInterrupt:
        logger.info("用户中断了程序")
    except Exception as e:
        logger.error(f"程序运行出错: {e}")
        logger.error(f"错误类型: {type(e).__name__}")

if **name** == "**main**":
main()

根据脚本中的测试文本添加的词库如下：

POST .analysis_ik/\_doc
{
"dict_type": "main_dicts",
"dict_key": "test_dic",
"dict_content": """祁老人
祁天佑
韵梅
祁瑞宣
老二
钱默吟
小顺子
李四大爷
冠晓荷
小妞子
程长顺
老舍
李四大爷
小羊圈胡同
北平城
胡同
小茶馆
小门脸
小房屋
四合院
院子
祁家
小院子
杂货铺
小铺子
井边
街上
菜市场
门口
枝头
城市
房间
北京
清朝
民国
战乱
战争
日本人
抗战
大槐树
槐树
小鸟
羊
门脸
房屋
水桶
菜担子
铺子
老头儿
儿子
教书先生
儿媳妇
女人
大家庭
孩子
孩子们
街坊邻居
妻子
老人
文人
知识分子
青年
汉奸
岁月
一生
变迁
衰落
建立
态度
威望
尊严
学问
诗
性格
时局
见解
小天地
精神
慰藉
现实
无力
气节
笑声
生机
阴霾
天真
快乐
希望
烦恼
童年
生意
生活
物资
年代
经营
日子
邻里
文化
野心
敌人
选择
软弱
妥协
民族大义
个人利益
温暖
时期
未来
力量
压力
抗争
逃避
方式
时代
煎熬
折磨
笔触
众生相
人物
特点
命运
喜怒哀乐
内涵
达观
痛苦
坚强
堕落
缩影
威胁
困难
道德
家人
向往
关系
矛盾
温情
传统文化
组成部分
理想
教育
抱负
占领
写照
亮色
心理
原则
底线
节奏
意义
细节
乐趣
个性
约束
麻烦
担忧
反叛精神
束缚
反抗
后果
建筑
代表
故事
记忆
历史文化底蕴
破坏
作品
创伤
经历
教训
和平
窗口
清晨
春天
内心
玩耍
聊天
晒太阳
歌唱
合作
打水
卖奶
帮助
"""
}

进行 2 次集中写入的记录如下：

# ik_max_test
2025-07-13 20:15:33,294 - INFO - 开始时间: 2025-07-13 19:45:07
2025-07-13 20:15:33,294 - INFO - 结束时间: 2025-07-13 20:15:33
2025-07-13 20:15:33,294 - INFO - 总耗时: 1825.31秒 (30.42分钟)
2025-07-13 20:15:33,294 - INFO - 总计完成: 1000 次bulk操作
2025-07-13 20:15:33,294 - INFO - 总计插入: 1000000 条文档
2025-07-13 20:15:33,294 - INFO - 最终索引大小: 0.92GB
2025-07-13 20:15:33,294 - INFO - 平均每次bulk耗时: 1.790秒
2025-07-13 20:15:33,294 - INFO - 平均bulk速率: 0.55次/秒
2025-07-13 20:15:33,294 - INFO - 平均文档写入速率: 548条/秒

# ik_custom_test
2025-07-13 21:17:47,309 - INFO - 开始时间: 2025-07-13 20:44:03
2025-07-13 21:17:47,309 - INFO - 结束时间: 2025-07-13 21:17:47
2025-07-13 21:17:47,309 - INFO - 总耗时: 2023.53秒 (33.73分钟)
2025-07-13 21:17:47,309 - INFO - 总计完成: 1000 次bulk操作
2025-07-13 21:17:47,309 - INFO - 总计插入: 1000000 条文档
2025-07-13 21:17:47,309 - INFO - 最终索引大小: 0.92GB
2025-07-13 21:17:47,309 - INFO - 平均每次bulk耗时: 1.986秒
2025-07-13 21:17:47,309 - INFO - 平均bulk速率: 0.49次/秒
2025-07-13 21:17:47,309 - INFO - 平均文档写入速率: 494条/秒

可以看到，有一定损耗，自定义词库词典的效率是之前的 90%。

相关阅读

• IK 字段级别词典升级：IK reload API
• Easysearch 新功能： IK 字段级别词典

关于 IK Analysis

IK Analysis 插件集成了 Lucene IK 分析器，并支持自定义词典。它支持 Easysearch\Elasticsearch\OpenSearch 的主要版本。由 INFINI Labs 维护并提供支持。

该插件包含分析器：ik_smart 和 ik_max_word，以及分词器：ik_smart 和 ik_max_word

开源地址：https://github.com/infinilabs/analysis-ik

作者：金多安，极限科技（INFINI Labs）搜索运维专家，Elastic 认证专家，搜索客社区日报责任编辑。一直从事与搜索运维相关的工作，日常会去挖掘 ES / Lucene 方向的搜索技术原理，保持搜索相关技术发展的关注。
原文：https://infinilabs.cn/blog/2025/ik-field-level-dictionarys-3/

INFINI Labs 产品更新发布！此次更新主要包括 Coco AI v0.7.0 新增 macOS Spotlight 和 Windows 文件搜索支持、语音输入功能，以及全屏集成模式；Easysearch v1.14.0 引入完整文本嵌入模型、语义检索 API 和搜索管道功能等，全面提升产品性能和稳定性。

Coco AI v0.7.0

Coco AI 是一款完全开源、跨平台的企业级智能搜索与助手系统，专为现代企业打造。它通过统一搜索入口，连接企业内外部的异构数据源，融合大模型能力，帮助团队高效访问知识，智能决策协作。

Coco AI 本次详细更新记录如下：

Coco AI 客户端 v0.7.0

功能特性 (Features)

• 文件搜索支持 Spotlight（macOS） (#705)
• 语音输入支持（搜索模式 & 聊天模式） (#732)
• 文本转语音现已由 LLM 驱动 (#750)
• Windows 文件搜索支持 (#762)

问题修复 (Bug Fixes)

• 文件搜索：优先应用过滤器后再处理 from/size 参数 (#741)
• 文件搜索：按名称与内容搜索时未匹配文件名问题 (#743)
• 修复 Windows 平台窗口被移动时自动隐藏的问题 (#748)
• 修复删除快捷键时未注销扩展热键的问题 (#770)
• 修复应用索引未遵循搜索范围配置的问题 (#773)
• 修复子页面缺失分类标题的问题 (#772)
• 修复快捷 AI 入口显示错误的问题 (#779)
• 语音播放相关的小问题修复 (#780)
• 修复 Linux 平台任务栏图标显示异常 (#783)
• 修复子页面数据不一致问题 (#784)
• 修复扩展安装状态显示错误 (#789)
• 增加 HTTP 流请求的超时容忍度，提升稳定性 (#798)
• 修复回车键行为异常问题 (#794)
• 修复重命名后选中状态失效的问题 (#800)
• 修复 Windows 右键菜单中快捷键异常问题 (#804)
• 修复因 "state() 在 manage() 之前调用" 引起的 panic (#806)
• 修复多行输入问题 (#808)
• 修复 Ctrl+K 快捷键无效问题 (#815)
• 修复窗口配置同步失败问题 (#818)
• 修复子页面回车键无法使用问题 (#819)
• 修复 Ubuntu (GNOME) 下打开应用时崩溃问题 (#821)

改进优化 (Improvements)

• 文件状态检测优先使用 stat(2) (#737)
• 文件搜索扩展类型重命名为 extension (#738)
• 创建聊天记录及发送聊天 API (#739)
• 更多文件类型图标支持 (#740)
• 替换 meval-rs 依赖，清除编译警告 (#745)
• Assistant、数据源、MCP Server 接口参数重构 (#746)
• 扩展代码结构调整 (#747)
• 升级 applications-rs 依赖版本 (#751)
• QuickLink/quick_link 重命名为 Quicklink/quicklink (#752)
• Assistant 样式与参数微调 (#753)
• 可选字段默认不强制要求填写 (#758)
• 搜索聊天组件新增 formatUrl、think 数据及图标地址支持 (#765)
• Coco App HTTP 请求统一添加请求头 (#744)
• 响应体反序列化前增加状态码判断 (#767)
• 启动页适配手机屏幕宽度 (#768)
• 搜索聊天新增语言参数与格式化 URL 参数 (#775)
• 未登录状态不请求用户接口 (#795)
• Windows 文件搜索清理查询字符串中的非法字符 (#802)
• 崩溃日志中展示 backtrace 信息 (#805)

相关截图

Coco AI 服务端 v0.7.0

功能特性 (Features)

• 重构了映射（mappings）的实现
• 新增了基于 HTTP 流式传输的聊天 API
• 新增了文件上传的配置选项
• 聊天消息中现已支持附件
• 为调试目的，增加记录大语言模型（LLM）请求的日志
• 新增 RSS 连接器
• 支持在初始化时配置模型的默认推理参数
• 新增本地文件系统（Local FS）连接器
• 新增 S3 连接器

问题修复(Bug Fixes)

• 修复了查询参数 "filter" 不生效的问题
• 修复了列表中分页功能不工作的问题
• 修复了在没有网络的情况下本地图标无法显示的问题
• 修复了大语言模型（LLM）提供商列表中状态显示不正确的问题
• 修复了带附件的聊天 API
• 防止了在 LLM 意图解析出错时可能出现的空指针异常
• 修复了删除多个 URL 输入框时功能不正常的问题
• 修复了启用本地模型提供商后状态未及时更新的问题
• 确保在 RAG（检索增强生成）处理过程中正确使用数据源
• 修复了提示词模板选择不正确的问题
• 防止了当用户取消正在进行的回复时可能导致回复消息丢失的问题
• 使第一条聊天消息可以被取消

改进优化 (Improvements)

• 重构了用户 ID 的处理方式
• 跳过空的流式响应数据块
• 重构了查询的实现
• 对更多敏感的搜索结果进行屏蔽处理
• 重构了附件相关的 API
• 为智能助理增加了上传设置
• 重构了 ORM 和安全接口
• 在附件上传 API 中移除了对 session_id 的检查
• 为搜索框增加了 formatUrl 功能
• 为集成页面增加了全屏模式
• 程序现在会忽略无效的连接器
• 程序现在会跳过无效的 MCP 服务器
• 对于内置的智能助理和提供商，隐藏了删除按钮
• 处理了提示词模板的默认值
• 如果某个集成功能被禁用，其按钮预览将显示为禁用状态
• 手动刷新流式输出的第一行数据，以改善响应体验

Easysearch v1.14.0

重大变更(Breaking Changes)

• AI 模块 从 modules 迁移至 plugins 目录下，方便调用 knn 插件
• 旧的文本向量化接口 _ai/embed 已不再支持，将在后续版本删除

功能特性 (Features)

• 插件模块新增完整的文本嵌入模型集成功能，涵盖从数据导入到向量检索的全流程
• 新增语义检索 API，简化向量搜索使用流程
• 新增语义检索处理器配置大模型信息
• 新增搜索管道（Search pipelines），轻松地在 Easysearch 内部处理查询请求和查询结果
• 多模型集成支持
  • OpenAI 向量模型：直接调用 OpenAI 的嵌入接口（如 text-embedding-3-small）
  • Ollama 本地模型：支持离线环境或私有化部署的向量生成
• IK 分词器提供 reload API，能够对存量自定义词典进行完整更新
• IK 分词器能够通过词库索引对默认词库进行自定义添加

改进优化 (Improvements)

• 增强数据摄取管道（ingest pipeline）
  • 在数据索引阶段支持文本向量化，文档可自动生成向量表示
  • 导入数据时通过 ingest 管道进行向量化时支持单条和批量模式，适配大模型的请求限制场景
• 更新 Easysearch Docker 初始化文档
• IK 分词器优化自定义词库加载逻辑，减少内存占用

Console v1.29.8

INFINI Console 是一款开源的非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管，企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 本次详细更新记录如下：

问题修复(Bug Fixes)

• 在获取分片级别的分片状态指标时，shard_id 参数未生效的问题
• 优化了监控图表中坐标轴标签的显示效果
• 在更改指标级别后，统计数据未刷新的问题
• 根据响应中的 key 来进行 rollup 检查
• 因 omitempty JSON 标签导致更新不生效时，改为使用 save 方法

改进优化 (Improvements)

• 为指标请求添加了自定义的超时错误处理
• 优化了动态分区逻辑
• 此版本包含了底层 Framework v1.2.0 的更新，解决了一些常见问题，并增强了整体稳定性和性能。虽然 Console 本身没有直接的变更，但从 Framework 中继承的改进间接地使 Console 受益。

Gateway v1.29.8

INFINI Gateway 是一个开源的面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

改进优化 (Improvements)

• 此版本包含了底层 Framework v1.2.0 的更新，解决了一些常见问题，并增强了整体稳定性和性能。虽然 Gateway 本身没有直接的变更，但从 Framework 中继承的改进间接地使 Gateway 受益。

Agent v1.29.8

INFINI Agent 负责采集和上传 Elasticsearch, Easysearch, Opensearch 集群的日志和指标信息，通过 INFINI Console 管理，支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

功能特性 (Features)

• 在 Kubernetes 环境下通过环境变量 http.port 探测 Easysearch 的 HTTP 端口

改进优化 (Improvements)

• 此版本包含了底层 Framework v1.2.0 的更新，解决了一些常见问题，并增强了整体稳定性和性能。虽然 Agent 本身没有直接的变更，但从 Framework 中继承的改进间接地使 Agent 受益。

Loadgen v1.29.8

INFINI Loadgen 是一款开源的专为 Easysearch、Elasticsearch、OpenSearch 设计的轻量级性能测试工具。

Loadgen 本次更新如下：

改进优化 (Improvements)

• 此版本包含了底层 Framework v1.2.0 的更新，解决了一些常见问题，并增强了整体稳定性和性能。虽然 Loadgen 本身没有直接的变更，但从 Framework 中继承的改进间接地使 Loadgen 受益。

Framework 1.2.0

INFINI Framework 是 INFINI Labs 基于 Golang 的产品的核心基础，已开源。该框架以开发者为中心设计，简化了构建高性能、可扩展且可靠的应用程序的过程。

Framework 本次更新如下：

功能特性 (Features)

• ORM 操作钩子 (Hooks)：为 ORM（数据访问层）的数据操作新增了钩子（Hooks），允许进行更灵活的二次开发。
• 新增 Create API：新增了用于创建文档的 _create API 接口，确保文档 ID 的唯一性。
• URL terms 查询：现在 URL 的查询参数也支持 terms 类型的查询了，可以一次匹配多个值。

问题修复 (Bug Fixes)

• 修复了通过 HTTP 插件设置的自定义 HTTP 头部信息未被正确应用的问题。
• 修复了 JSON 解析器的一个问题，现在可以正确处理带引号的、且包含下划线 _ 的 JSON 键（key）。

改进 (Improvements)

• 查询过滤器优化: 系统现在会自动将多个针对同一字段的 term 过滤器合并为一个更高效的 terms 过滤器，以提升查询性能。
• 查询接口重构: 对核心的查询接口进行了重构，使其结构更清晰，为未来的功能扩展打下基础。

更多详情请查看以下各产品的 Release Notes 或联系我们的技术支持团队！

• Coco AI App
• Coco AI Server
• INFINI Easysearch
• INFINI Console
• INFINI Gateway
• INFINI Agent
• INFINI Loadgen
• INFINI Framework

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

下载地址： https://infinilabs.cn/download

邮件：hello@infini.ltd

电话：(+86) 400-139-9200

Discord：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

关于极限科技（INFINI Labs）

极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：https://infinilabs.cn

之前介绍 IK 字段级别字典 使用的时候，对于字典的更新只是支持词典库的新增，并不支持对存量词典库的修改或者删除。经过这段时间的开发，已经可以兼容词典库的更新，主要通过 IK reload API 来实现。

IK reload API

IK reload API 通过对词典库的全量重新加载来实现词典库的更新或者删除。用户可以通过下面的命令实现：

# 测试索引准备

PUT my-index-000001
{
  "settings": {
    "number_of_shards": 3,
    "analysis": {
      "analyzer": {
        "my_custom_analyzer": {
          "type": "custom",
          "tokenizer": "my_tokenizer"
        }
      },
      "tokenizer": {
        "my_tokenizer": {

          "type": "ik_smart",
          "custom_dict_enable": true,
          "load_default_dicts":false, # 这里不包含默认词库
          "lowcase_enable": true,
          "dict_key": "test_dic"
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "test_ik": {
        "type": "text",
        "analyzer": "my_custom_analyzer"
      }
    }
  }
}

# 原来词库分词效果，只预置了分词“自强不息”
GET my-index-000001/_analyze
{
  "analyzer": "my_custom_analyzer",
  "text":"自强不息，杨树林"
}

{
  "tokens": [
    {
      "token": "自强不息",
      "start_offset": 0,
      "end_offset": 4,
      "type": "CN_WORD",
      "position": 0
    },
    {
      "token": "杨",
      "start_offset": 5,
      "end_offset": 6,
      "type": "CN_CHAR",
      "position": 1
    },
    {
      "token": "树",
      "start_offset": 6,
      "end_offset": 7,
      "type": "CN_CHAR",
      "position": 2
    },
    {
      "token": "林",
      "start_offset": 7,
      "end_offset": 8,
      "type": "CN_CHAR",
      "position": 3
    }
  ]
}

# 更新词库
POST .analysis_ik/_doc
{
  "dict_key": "test_dic",
  "dict_type": "main_dicts",
  "dict_content":"杨树林"
}
# 删除词库，词库文档的id为coayoJcBFHNnLYAKfTML
DELETE .analysis_ik/_doc/coayoJcBFHNnLYAKfTML?refresh=true

# 重载词库
POST _ik/_reload
{}

# 更新后的词库效果
GET my-index-000001/_analyze
{
  "analyzer": "my_custom_analyzer",
  "text":"自强不息，杨树林"
}

{
  "tokens": [
    {
      "token": "自",
      "start_offset": 0,
      "end_offset": 1,
      "type": "CN_CHAR",
      "position": 0
    },
    {
      "token": "强",
      "start_offset": 1,
      "end_offset": 2,
      "type": "CN_CHAR",
      "position": 1
    },
    {
      "token": "不",
      "start_offset": 2,
      "end_offset": 3,
      "type": "CN_CHAR",
      "position": 2
    },
    {
      "token": "息",
      "start_offset": 3,
      "end_offset": 4,
      "type": "CN_CHAR",
      "position": 3
    },
    {
      "token": "杨树林",
      "start_offset": 5,
      "end_offset": 8,
      "type": "CN_WORD",
      "position": 4
    }
  ]
}

这里是实现索引里全部的词库更新。

也可以实现单独的词典库更新

POST _ik/_reload
{"dict_key":"test_dic”}

# debug 日志
[2025-07-09T15:30:29,439][INFO ][o.e.a.i.ReloadIK         ] [ik-1] 收到重载IK词典的请求，将在所有节点上执行。dict_key: test_dic, dict_index: .analysis_ik
[2025-07-09T15:30:29,439][INFO ][o.e.a.i.a.TransportReloadIKDictionaryAction] [ik-1] 在节点 [R6ESV5h1Q8OZMNoosSDEmg] 上执行词典重载操作，dict_key: test_dic, dict_index: .analysis_ik

这里传入的 dict_key 对应的词库 id。

对于自定义的词库存储索引，也可以指定词库索引的名称，如果不指定则默认使用 .analysis_ik

POST _ik/_reload
{"dict_index":"ik_index"}

# debug 日志
[2025-07-09T15:32:59,196][INFO ][o.e.a.i.a.TransportReloadIKDictionaryAction] [ik-1] 在节点 [R6ESV5h1Q8OZMNoosSDEmg] 上执行词典重载操作，dict_key: null, dict_index: test_ik
[2025-07-09T15:32:59,196][INFO ][o.w.a.d.ReloadDict       ] [ik-1] Reloading all dictionaries

注：

1. 更新或者删除词库重载后只是对后续写入的文档生效，对已索引的文档无效；
2. 因为用户无法直接更改 IK 内置的词库（即默认配置路径下的词库文件），因此 reload API 不会影响内置词库的信息。

相关阅读

• IK 字段级别词典的升级之路
• Easysearch 新功能： IK 字段级别词典

关于 IK Analysis

IK Analysis 插件集成了 Lucene IK 分析器，并支持自定义词典。它支持 Easysearch\Elasticsearch\OpenSearch 的主要版本。由 INFINI Labs 维护并提供支持。

该插件包含分析器：ik_smart 和 ik_max_word，以及分词器：ik_smart 和 ik_max_word

开源地址：https://github.com/infinilabs/analysis-ik

作者：金多安，极限科技（INFINI Labs）搜索运维专家，Elastic 认证专家，搜索客社区日报责任编辑。一直从事与搜索运维相关的工作，日常会去挖掘 ES / Lucene 方向的搜索技术原理，保持搜索相关技术发展的关注。
原文：https://infinilabs.cn/blog/2025/ik-field-level-dictionarys-2/

背景

在当前 AI 与搜索深度融合的时代，语义搜索已成为企业级应用的核心能力之一。作为 Elasticsearch 的国产化替代方案，Easysearch 不仅具备高性能、高可用、弹性伸缩等企业级特性，更通过灵活的插件化架构，支持多种主流 Embedding 模型服务，包括 阿里云通义千问（DashScope） 和 本地化 Ollama 服务，实现对 OpenAI 接口规范的完美兼容。

本文将详细介绍如何在 Easysearch 中集成阿里云和 Ollama 的 Embedding API，构建端到端的语义搜索系统，并提供完整的配置示例与流程图解析。

一、为什么选择 Easysearch？

Easysearch 是由极限科技（INFINI Labs）自主研发的分布式近实时搜索型数据库，具备以下核心优势：

• ✅ 完全兼容 Elasticsearch 7.x API 及 8.x 常用操作
• ✅ 原生支持向量检索（kNN）、语义搜索、混合检索
• ✅ 内置数据摄入管道与搜索管道，支持 AI 模型集成
• ✅ 支持国产化部署、数据安全可控
• ✅ 高性能、低延迟、可扩展性强

尤其在 AI 增强搜索场景中，Easysearch 提供了强大的 text_embedding 和 semantic_query_enricher 处理器，允许无缝接入外部 Embedding 模型服务。

二、支持的 Embedding 服务

Easysearch 通过标准 OpenAI 兼容接口无缝集成各类第三方 Embedding 模型服务，理论上支持所有符合 OpenAI Embedding API 规范的模型。以下是已验证的典型服务示例：

核心优势：

1. 广泛兼容性
   支持任意实现 OpenAI Embedding API 格式（/v1/embeddings）的服务，包括：

   • 请求格式：{ "input": "text", "model": "model_name" }
   • 响应格式：{ "data": [{ "embedding": [...] }] }

2. 即插即用
   仅需配置服务端点的 base_url 和 api_key 即可快速接入新模型。

3. 混合部署
   可同时配置多个云端或本地模型，根据业务需求灵活切换。

三、结合 AI 服务流程图

说明：

• 索引阶段：通过 Ingest Pipeline 调用 Embedding API，将文本转为向量并存储。
• 搜索阶段：通过 Search Pipeline 动态生成查询向量，执行语义相似度匹配。
• 所有 API 调用均兼容 OpenAI 接口格式，降低集成成本。

四、集成阿里云 DashScope（通义千问）

阿里云 DashScope 提供高性能文本嵌入模型 text-embedding-v4，支持 256 维向量输出，适用于中文语义理解任务。

1. 创建 Ingest Pipeline（索引时生成向量）

PUT _ingest/pipeline/text-embedding-aliyun
{
  "description": "阿里云用于生成文本嵌入向量的管道",
  "processors": [
    {
      "text_embedding": {
        "url": "https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings",
        "vendor": "openai",
        "api_key": "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
        "text_field": "input_text",
        "vector_field": "text_vector",
        "model_id": "text-embedding-v4",
        "dims": 256,
        "batch_size": 5
      }
    }
  ]
}

2. 创建索引并定义向量字段

PUT /my-index
{
  "mappings": {
    "properties": {
      "input_text": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "text_vector": {
        "type": "knn_dense_float_vector",
        "knn": {
          "dims": 256,
          "model": "lsh",
          "similarity": "cosine",
          "L": 99,
          "k": 1
        }
      }
    }
  }
}

3. 使用 Pipeline 批量写入数据

POST /_bulk?pipeline=text-embedding-aliyun&refresh=wait_for
{ "index": { "_index": "my-index", "_id": "1" } }
{ "input_text": "风急天高猿啸哀，渚清沙白鸟飞回..." }
{ "index": { "_index": "my-index", "_id": "2" } }
{ "input_text": "月落乌啼霜满天，江枫渔火对愁眠..." }
...

4. 配置 Search Pipeline（搜索时动态生成向量）

PUT /_search/pipeline/search_model_aliyun
{
  "request_processors": [
    {
      "semantic_query_enricher": {
        "tag": "tag1",
        "description": "阿里云 search embedding model",
        "url": "https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings",
        "vendor": "openai",
        "api_key": "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
        "default_model_id": "text-embedding-v4",
        "vector_field_model_id": {
          "text_vector": "text-embedding-v4"
        }
      }
    }
  ]
}

5. 设置索引默认搜索管道

PUT /my-index/_settings
{
  "index.search.default_pipeline": "search_model_aliyun"
}

6. 执行语义搜索

GET /my-index/_search
{
  "_source": "input_text",
  "query": {
    "semantic": {
      "text_vector": {
        "query_text": "风急天高猿啸哀，渚清沙白鸟飞回...",
        "candidates": 10,
        "query_strategy": "LSH_COSINE"
      }
    }
  }
}

搜索结果示例：

"hits": [
  {
    "_id": "1",
    "_score": 2.0,
    "_source": { "input_text": "风急天高猿啸哀..." }
  },
  {
    "_id": "4",
    "_score": 1.75,
    "_source": { "input_text": "白日依山尽..." }
  },
  ...
]

结果显示：相同诗句匹配得分最高，其他古诗按语义相似度排序，效果理想。

五、集成本地 Ollama 服务

Ollama 支持在本地运行开源 Embedding 模型（如 nomic-embed-text），适合对数据隐私要求高的场景。

1. 启动 Ollama 服务

ollama serve
ollama pull nomic-embed-text:latest

2. 创建 Ingest Pipeline（使用 Ollama）

PUT _ingest/pipeline/ollama-embedding-pipeline
{
  "description": "Ollama embedding 示例",
  "processors": [
    {
      "text_embedding": {
        "url": "http://localhost:11434/api/embed",
        "vendor": "ollama",
        "text_field": "input_text",
        "vector_field": "text_vector",
        "model_id": "nomic-embed-text:latest"
      }
    }
  ]
}

3. 创建 Search Pipeline（搜索时使用 Ollama）

PUT /_search/pipeline/ollama_model_pipeline
{
  "request_processors": [
    {
      "semantic_query_enricher": {
        "tag": "tag1",
        "description": "Sets the ollama model",
        "url": "http://localhost:11434/api/embed",
        "vendor": "ollama",
        "default_model_id": "nomic-embed-text:latest",
        "vector_field_model_id": {
          "text_vector": "nomic-embed-text:latest"
        }
      }
    }
  ]
}

后续步骤与阿里云一致：创建索引 → 写入数据 → 搜索查询。

六、安全性说明

Easysearch 在处理 API Key 时采取以下安全措施：

• 🔐 所有 api_key 在返回时自动加密脱敏（如 TfUmLjPg...infinilabs）
• 🔒 支持密钥管理插件（如 Hashicorp Vault 集成）
• 🛡️ 支持 HTTPS、RBAC、审计日志等企业级安全功能

确保敏感信息不被泄露，满足合规要求。

七、总结

通过 Easysearch 的 Ingest Pipeline 与 Search Pipeline，我们可以轻松集成：

• ✅ 阿里云 DashScope（云端高性能）
• ✅ Ollama（本地私有化部署）
• ✅ 其他支持 OpenAI 接口的 Embedding 服务

无论是追求性能还是数据安全，Easysearch 都能提供灵活、高效的语义搜索解决方案。

八、下一步建议

• 尝试混合检索：结合关键词匹配与语义搜索
• 使用 Rerank 模型提升排序精度
• 部署多节点集群提升吞吐量
• 接入 INFINI Gateway 实现统一 API 网关管理

参考链接

• Easysearch 官网文档
• 阿里云 DashScope 文档
• Ollama 官方网站
• INFINI Labs GitHub

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://docs.infinilabs.com/easysearch

作者：张磊，极限科技（INFINI Labs）搜索引擎研发负责人，对 Elasticsearch 和 Lucene 源码比较熟悉，目前主要负责公司的 Easysearch 产品的研发以及客户服务工作。
原文：https://infinilabs.cn/blog/2025/Easysearch-Integration-with-Alibaba-CloudOllama-Embedding-API/

2025 年 7 月 17 日 在北京召开的 TDBC 2025 可信数据库发展大会·人工智能与数据库融合发展分论坛 上，国内领先的搜索数据库及解决方案提供商 极限科技（INFINI Labs）正式发布其创新产品 —— Coco AI，一款面向企业的 AI 智能搜索与高效协作平台。极限数据（北京）科技有限公司创始人曾勇在《下一代企业搜索与 AI 的融合探索》主题演讲中，深入探讨了企业搜索的未来趋势及 Coco AI 的核心价值。

破解企业数据困境：多平台孤岛与隐私安全挑战

随着企业数据呈爆发式增长，信息分散在本地文件系统、云存储（如 S3）、协作工具（如内网 Wiki、工单系统、 CRM、ERP 等）及代码仓库（如 Gitcode、Gitee 等）等多个平台，员工往往需要频繁切换系统进行检索，效率低下。同时，企业对数据隐私和安全的高标准要求，使得直接使用公有云 AI 工具或上云存储存在潜在风险。此外，企业内部积累的海量知识未能有效转化为生产力，传统知识库管理依赖人工维护，效果有限。

Coco AI 的推出，正是为了帮助企业解决这些痛点，通过统一搜索、AI 增强、隐私优先的设计理念，打造一站式智能搜索中心，提升知识利用效率与协作体验。

Coco AI 核心功能：重新定义企业搜索与协作

1. 跨平台统一搜索

• 支持本地文件、云端应用或自研各业务系统及第三方平台的无缝搜索，提供“本地搜”“云端搜”“混合搜”及“多模态搜”（文本、语音、图像、视频）能力，彻底告别数据碎片化。

2. 无缝 AI 集成

• 基于语义搜索与生成式 AI 技术，支持快速提取企业内部知识，并接入 OpenAI、DeepSeek 等主流大模型。
• 相比集中式知识库，采用联邦数据连接方式，降低维护成本。
• 提供高度可定制的 RAG（检索增强生成）管道，让 AI 回答更精准、更贴合业务需求。

3. 隐私优先设计

• 支持自托管部署，数据全程加密，提供企业级权限控制与动态脱敏功能，确保敏感信息不外泄，满足金融、医疗等高合规性行业需求。

4. 灵活扩展性

• 通过 MCP 协议、Connector 及 Remote Search Adapter（RSA）对接企业自有数据源，支持 API 和插件扩展，适配多样化业务场景。

轻量级体验，高效团队协作

Coco AI 以“轻量级、无存在感”为设计理念，内置实用工具（如计算器、便签）及团队协作功能，支持快速定位共享资源。其智能助手可在搜索中实时总结结果，提升决策效率。此外，APP 端支持多服务器连接，实现跨域协作而无需集中同步数据，兼顾便利性与安全性。

技术架构与生态兼容

Coco AI Server 基于极限科技的搜索型数据库的 Easysearch 即可运行，支持从单机到 PB 级分布式扩展，并针对中文语义进行了深度优化。其开放生态兼容 MCP 协议、主流大模型及第三方工具，企业可灵活集成现有系统，降低迁移成本。

开源共建，赋能开发者生态

极限科技始终秉持开放合作理念，Coco AI 已正式开源，并邀请全球开发者共同参与生态建设。用户可通过 GitHub/Gitee/GitCode 来获取代码、提交贡献，或加入官方社群获取技术支持。

曾勇表示：“Coco AI 的诞生，不仅是极限科技在搜索与 AI 融合领域的一次重大突破，更是我们对‘让搜索更简单’这一使命的践行。未来，我们将持续优化产品，助力企业释放数据价值，迈向智能化搜索新未来。”

2025 年 7 月 17 日 在北京召开的 TDBC 2025 可信数据库发展大会·数据库生态及国际化分论坛 上，全球数据库领域专家、学者与企业代表齐聚。极限数据（北京）科技有限公司联合创始人曾嘉毅发表《搜索型数据库生态建设及展望》主题演讲，剖析技术创新与实践，为行业提供高效数据检索与智能应用方案。

破解数据检索挑战，AI 赋能搜索升级

首先，我们需要面对结构化数据。典型处理方式是使用传统关系型数据库。但是，关系型数据库的设计初衷就决定了它面对的挑战：关系型数据库优先保证事务性，其数据分层结构导致查询需要层层下钻，同时传统关系型数据库能够处理的数据规模也是受限的。搜索型数据库针对以上挑战可以实现读写分离、多表聚合查询、数据库加速等。

与此同时，企业数据中大约 85% 为非结构化或半结构化数据，如图片、视频等，传统数据库处理困难。极限科技运用语义解析与 AI 向量化技术，语义解析深入理解数据语义并转化为结构化信息，AI 向量化将其映射到高维空间实现向量化表示，二者结合完成非结构化数据的标签提取与索引构建，提升检索准确性与效率。

针对中文文本，极限科技进行字段化处理研究。中文语法复杂、语义丰富，传统方法难以满足检索需求。公司通过自研算法精准分词与字段提取，结合向量化技术提升中文数据检索效果。同时，融合向量化全量搜索与模糊搜索，前者快速定位相似数据，后者处理用户输入的不准确信息，提高搜索容错性。

平台化建设与工具开源：打造全链路能力

极限科技构建的管控平台功能强大。支持多集群元原生编排和管理，企业可依业务场景和用户需求灵活调整集群资源，同时实现一键升级、备份管理等；提供统一监控、统一身份管理服务，实时监控系统组件与运行状态，及时预警问题。该平台兼容多厂商环境，企业可无缝集成现有系统，降低迁移成本与风险。公司开发的搜索服务网关针对检索服务提供流量分发与链路加速能力，进而实现查询分析、干预等高阶功能。

此外，极限科技积极推动搜索周边工具开源贡献。数据迁移工具 ESM 助力企业快速安全迁移数据至自家搜索型数据库，缩短迁移周期、降低风险；性能压测工具 Loadgen 模拟复杂场景测试系统性能，评估性能瓶颈与承载能力；中文分词工具 IK/Pinyin 支持多种分词模式与自定义词典，满足不同用户需求。开源工具促进技术交流创新，支持行业生态发展。

“Coco” AI 搜索与智能体结合模式：重构搜索体验

Coco AI 采用获得国家专利设计的人机交互体验，将搜索与 AI 进行无缝结合。传统 RAG 存在大模型直接回答搜索问题存在训练成本高、回答不精准问题。 Coco AI 后台灵活，支持为不同类型问题分配专属“小助手”。“小助手”针对特定问题优化配置，精准理解用户意图、提供准确回答，降低训练成本、提升回答精准度与效率。可以快速量身打造企业专属的 AI 智能体工具箱。

Coco AI 结合本地与云端协同搜索技术，连接本地文件、数据库及外部应用系统数据源。用户搜索时，可以同时对本地和外部 Coco Server 引擎同时处理查询请求，然后对结果进行打分与整合去重排序，结合大模型总结分析最终结果，实现意图理解与统一信息获取，打破信息孤岛，提供全面准确高效的搜索服务。

展望未来：AI 搜索与开放生态

极限科技对搜索型数据库未来有清晰规划。下一代 AI 搜索架构将深度融合向量检索与智能体技术。向量检索已发挥重要作用，智能体技术能自主感知、决策与行动。二者融合使 AI 搜索系统更智能理解用户需求，主动提供个性化服务，如依历史记录推荐信息，面对复杂任务自主分解协调资源处理。

在企业数据应用场景上，下一代架构将进一步优化拓展。除传统文档检索、数据查询，还将深入生产、运营、管理等环节，提供全面深入的数据分析与决策支持。如在生产制造中实时分析设备数据、提前发现故障隐患；在市场营销中深度挖掘客户数据、制定精准营销策略。

为推动行业发展，极限科技将持续推进开源战略，通过 GitHub/Gitee/GitCode 等平台共享核心技术代码与文档，与全球开发者紧密合作。吸引更多开发者参与研发创新，共同解决技术难题。同时积极参与行业标准制定推广，促进市场规范化标准化发展，构建开放共享共赢的搜索型数据库生态。

此次分享展示了极限科技的技术实力与创新成果，为行业发展提供新思路方向。相信未来，极限科技将秉持创新、开放、合作理念，推动技术发展应用，为企业数字化转型与行业发展注入新动力。