基于Lindorm多模能力快速搭建智能搜索服务

环节

说明

涉及引擎

创建父表（知识库）

创建一个父表，用于存储知识库的原始文本。

"""
创建父表语句，可按需修改
"""
def create_parent_table(self):
    sql = """
        CREATE TABLE IF NOT EXISTS {} (
            document_id  VARCHAR, 
            title VARCHAR, 
            context VARCHAR,
            status   INT, 
            metadata JSON, 
            PRIMARY KEY (document_id)
        )
    """.format(self.parent.row_parent_table)
    print("Create parent table sql: ", sql)
    self.common_create_table(sql)

宽表引擎

创建子表（切分知识库）

创建一个子表，用于存储原文本切分后的段落，这些段落可以根据长度或者结合长度与标点符号来切分。

"""
创建子表语句，可按需修改
"""
def create_child_table(self):
    sql = """
        CREATE TABLE IF NOT EXISTS {} (
            document_id VARCHAR,
            chunking_position INT,
            title  VARCHAR,
            {}   VARCHAR,
            {} VARCHAR,
            metadata JSON,
            chunking_number INT,
            PRIMARY KEY (document_id, chunking_position)
        )
    """.format(self.parent.row_child_table, 
               self.parent.text_field, 
               self.parent.vector_field)
    print("Create child table sql: ", sql)
    self.common_create_table(sql)

宽表引擎

创建搜索Pipeline

对子表创建写入和查询Pipeline，写入时子表内容自动同步到搜索引擎，在搜索引擎内完成Embedding后，将结果写入向量引擎。

查询时自动对文本进行Embedding，通过向量引擎进行向量检索。

"""
创建pipeline,搜索内部自动对text字段调用ai引擎进行embedding,写入vector_field 字段
"""
def create_pipeline(self):
    inner_ai_host = Config.AI_HOST
    if "-pub" in inner_ai_host:
        inner_ai_host = inner_ai_host.replace("-pub", "-vpc")
                
    pipeline = {
        "description": "demo_chunking pipeline",
        "processors": [
            {
                "text-embedding": {
                    "inputFields": [self.parent.text_field],   
                    "outputFields": [self.parent.vector_field],
                    "userName": Config.LD_USER,
                    "password": Config.LD_PASSWORD,
                    "url": "http://{}:{}/dashscope/compatible-mode/v1/embeddings".format(inner_ai_host, int(Config.AI_PORT)),
                    "modeName": self.parent.embedding_model_name
                }
            }
        ]
    }    
    try:
        resp = self.client.ingest.put_pipeline(id=self.parent.pipeline_name, body=pipeline)
        if resp.get("acknowledged"):
            print("Create ingest pipeline success", resp)
        else:
            print("Create ingest pipeline error", resp)
    except Exception as e:
        print(f"Create ingest pipeline error: {e}")

    # 创建查询用的 search pipeline
    search_pipeline_url = f"http://{Config.SEARCH_HOST}:{Config.SEARCH_PORT}/_search/pipeline/{self.parent.search_pipeline_name}"
    search_pipeline_data = {
        "description": "demo multimodal embedding pipeline",
        "request_processors": [
            {
                "text-embedding": {
                    "model_config": {
                        "inputFields": [self.parent.text_field],
                        "outputFields": [self.parent.vector_field],
                        "userName": Config.LD_USER,
                        "password": Config.LD_PASSWORD,
                        "url": f"http://{inner_ai_host}:{Config.AI_PORT}/dashscope/compatible-mode/v1/embeddings",
                        "modeName": self.parent.embedding_model_name
                    }
                }
            }
        ]
    }
    try:
        resp = requests.put(search_pipeline_url, auth=(Config.LD_USER, Config.LD_PASSWORD),
                            json=search_pipeline_data, headers={"Content-Type": "application/json"})
        if resp.status_code in [200, 201]:
            print("Create search pipeline success", resp)
        else:
            print("Create search pipeline error ", resp)
    except Exception as e:
        print(f"Create search pipeline error: {e}")

宽表引擎

搜索引擎

向量引擎

创建搜索索引和向量索引

创建搜索索引和向量索引，提高检索效率。

示例中使用IVFBQ索引，如果数据量较小，可以使用HNSW索引。

"""
子表创建搜索索引，本文范例中使用的是ivfbq索引
https://help.aliyun.com/document_detail/2773371.html
"""
# Reference: https://help.aliyun.com/document_detail/260841.html
def create_child_table_index(self):
    sql = """
        CREATE INDEX IF NOT EXISTS %s USING SEARCH ON %s (
                document_id(indexed=false,columnStored=false),
                chunking_position(indexed=false,columnStored=false),
                title(type=text,analyzer=ik),
                metadata(indexed=false,columnStored=false),
                %s(type=text,analyzer=ik),
                %s(mapping='{
                    "type": "knn_vector",
                    "dimension": 1024,
                    "data_type": "float",
                    "method": {
                        "engine": "lvector",
                        "name": "ivfbq",
                        "space_type": "cosinesimil",
                        "parameters": {
                            "exbits": 2,
                            "nlist": 50
                            }
                        }
                }')) WITH (INDEX_SETTINGS='{
                "index": {
                    "knn" : "true",
                    "knn.vector_empty_value_to_keep" : true
                }}',
                    SOURCE_SETTINGS=
                    '{
                        "includes": ["document_id", "chunking_position","title", "text"]
                    }',
                    numShards=2
                )
        """.strip() % (self.parent.chunking_index_name,
                       self.parent.row_child_table,
                       self.parent.text_field,
                       self.parent.vector_field)
    print("Create search index sql: \n ", sql)
    self.common_create_table(sql)

宽表引擎

搜索引擎

向量引擎

搜索索引配置Pipeline

为搜索索引配置写入和查询Pipeline。

def set_index_pipelines(self):
    """
    通过ES API为索引设置默认的写入pipeline和查询pipeline
    """
    url = f"http://{Config.SEARCH_HOST}:{Config.SEARCH_PORT}/{self.index_name}/_settings"
    settings = {
        "index.default_pipeline": self.parent.pipeline_name,
        "index.search.default_pipeline": self.parent.search_pipeline_name
    }
    try:
        resp = requests.put(url, auth=(Config.LD_USER, Config.LD_PASSWORD),
                            json=settings, headers={"Content-Type": "application/json"})
        if resp.status_code in [200, 201]:
            print(f"Index pipelines set successfully: write={self.parent.pipeline_name}, search={self.parent.search_pipeline_name}")
        else:
            print(f"Set index pipelines error: {resp.status_code}, {resp.text}")
    except Exception as e:
        print(f"Set index pipelines error: {e}")

搜索引擎

向量引擎

构建向量索引

ivfpq、ivfbq类型的索引需要在写入一定数据量之后进行索引构建手动触发索引构建完成后，后续可正常写入和查询，无需再次构建索引，具体可参考索引构建。

def build_vector_index(self):
    """触发向量索引构建并等待完成"""
        
    url_base = f"http://{Config.SEARCH_HOST}:{Config.SEARCH_PORT}"
    auth = (Config.LD_USER, Config.LD_PASSWORD)
    index_name = f"default.{self.row_child_table}.{self.chunking_index_name}"
    field_name = self.vector_field
        
    print(f"开始构建向量索引: {index_name}.{field_name}")
        
    # 1. 触发构建
    build_url = f"{url_base}/_plugins/_vector/index/build"
    build_data = {
        "indexName": index_name,
        "fieldName": field_name,
        "removeOldIndex": "true"
    }
        
    try:
        resp = requests.post(build_url, auth=auth, json=build_data, headers={"Content-Type": "application/json"})
        print(f"触发构建响应: {resp.status_code}, {resp.text}")
            
        # 2. 轮询检查状态
        task_url = f"{url_base}/_plugins/_vector/index/tasks"
        task_data = {
            "indexName": index_name,
            "fieldName": field_name,
            "taskIds": "[]"
        }
            
        max_wait = 300  # 最多等5分钟
        start_time = time.time()
            
        while time.time() - start_time < max_wait:
            time.sleep(5)  # 每5秒检查一次
                
            resp = requests.get(task_url, auth=auth, json=task_data, headers={"Content-Type": "application/json"})
            result = resp.json()
            print(f"构建状态: {result}")
                
            if "payload" in result:
                for task in result["payload"]:
                    if "FINISH" in task:
                        print("向量索引构建完成!")
                        return True
                    elif "FAILED" in task:
                        print(f"向量索引构建失败: {task}")
                        return False
                            
        print("等待构建超时")
        return False
            
    except Exception as e:
        print(f"构建向量索引出错: {e}")
        return False

搜索索引

向量索引

检索方式

具体描述

流程图

全文与向量混合检索

全文与向量混合检索。

def demo_rrf_search():
    lindorm = Lindorm()
    query="国际初中科学奥林匹克主要比赛科目"
    results = lindorm.lindormSearch.rrf_search(query)
    display(JSON(results, expanded=True, root="rrf_search_result"))
    lindorm.close()
demo_rrf_search()

全文与向量混合检索+使用检索重排

先进行全文与向量混合检索，然后对混合检索的结果重排序。

def demo_rerank():
    lindorm = Lindorm()
    query="国际初中科学奥林匹克主要比赛科目"
    topk=int(Config.SEARCH_TOP_K)
    origin_result = lindorm.lindormSearch.rrf_search(query,  topk * 2)
    display(JSON(origin_result[0:topk], expanded=True, root="Before rerank result"))
    texts = [item["_source"]["text"] for item in origin_result]
    reranker_result = lindorm.lindormAI.reranker(query, texts)
    reranked_origin_result = handler_reranker(origin_result, reranker_result, topk)
    display(JSON(reranked_origin_result, expanded=True, root="After rerank result"))
    lindorm.close()
demo_rerank()

全文与向量混合检索+使用检索重排+Prompt

1. 先执行全文与向量混合检索，对混合检索的结果重排序。

2. 将问题以及检索结果一起形成Prompt，提交给大模型。

def demo_chat_with_child_chunking():
    lindorm = Lindorm()
    topk=int(Config.SEARCH_TOP_K)
    query="国际初中科学奥林匹克主要比赛科目"
    search_result = lindorm.lindormSearch.rrf_search(query, topk * 2)
    texts = [item["_source"]["text"] for item in search_result]
    reranker_result = lindorm.lindormAI.reranker(query, texts)
    prompt_context = "\n".join(texts[item['index']] for item in reranker_result[0:topk])
    prompt = lindorm.lindormAI.gen_prompt(query, prompt_context)
    output_text = ""
    for part in lindorm.lindormAI.chat_stream(prompt):
        output_text = part 
        wrapped_text = wrap_text(output_text, 80)
        clear_output(wait=True)
        display(HTML(f"<pre style='color: red;'>{wrapped_text}</pre>"))
    
    wrapped_text = wrap_text(prompt, 120)
    display(HTML(f"<pre>提示模版为:\n{wrapped_text}</pre>"))
    lindorm.close()
    
demo_chat_with_child_chunking()

全文与向量混合检索+使用检索重排+Context Prompt

1. 先执行全文与向量混合检索，对混合检索的结果重排序。

2. 根据父表（create_parent_table表）的PRIMARY KEY字段，从父表中查询初步检索结果的Context。

3. 将问题与Context一起形成Prompt，并通过AI引擎访问大模型，获得回答。

def demo_chat_with_parent():
    lindorm = Lindorm()
    topk=int(Config.SEARCH_TOP_K)
    query="国际初中科学奥林匹克主要比赛科目"
    search_result = lindorm.lindormSearch.rrf_search(query, topk * 2)
    texts = [item["_source"]["text"] for item in search_result]
    reranker_result = lindorm.lindormAI.reranker(query, texts)
    reranked_origin_result = handler_reranker(search_result, reranker_result, topk)
    unique_document_ids = list(OrderedDict.fromkeys(item['_source']['document_id'] for item in reranked_origin_result))
    contexts = []
    for document_id in unique_document_ids:
        contexts.append(lindorm.lindormRow.get_parent_context(pk=document_id))
    prompt_context = "\n".join(contexts)        
    prompt = lindorm.lindormAI.gen_prompt(query, prompt_context)    
    # stream
    for part in lindorm.lindormAI.chat_stream(prompt):
        output_text = part 
        wrapped_text = wrap_text(output_text, 80)
        clear_output(wait=True)
        display(HTML(f"<pre style='color: red;'>{wrapped_text}</pre>"))
    
    wrapped_text = wrap_text(prompt, 120)
    display(HTML(f"<pre>提示模版为:\n{wrapped_text}</pre>")) 
    lindorm.close()

demo_chat_with_parent()