基于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)

宽表引擎

部署及调用模型

部署Embedding模型：

部署自带开源BGE-M3模型（内置模型，无需单独下载），用于后续的文本向量化处理，将文本向量化后，可以通过语义实现近似检索。

"""
创建Embedding模型，目前推荐使用BGE-M3模型即可
"""
def create_embedding_model(self):
    if self.check_model_exists(self.parent.embedding_model_name):
        print("Model {} exists, skip create".format(self.parent.embedding_model_name))
        return

    self.common_create_model(self.parent.embedding_model_name, 
                             "huggingface://BAAI/bge-m3",
                             "FEATURE_EXTRACTION",
                             "BGE_M3")
                             
"""
对输入的文本进行Embedding成向量
"""
def text_embedding(self, input_text:str):
    url = "http://{}:{}/v1/ai/models/{}/infer".format(Config.AI_HOST, 
                                                      Config.AI_PORT, 
                                                      self.parent.embedding_model_name)
    input_text_utf8 = input_text.encode('utf-8').decode('utf-8')
    data = {
        "input": [input_text_utf8]
    }
    response = requests.post(url, data=json.dumps(data), headers=self.headers)
    json_response = response.json()
    if response.status_code != 200 or json_response["success"] is False:
        raise Exception("http request failed, status code: {}".format(json_response["msg"]))
    return json_response["data"][0]

AI引擎

部署ReRank模型：

部署自带开源BGE ReRank模型（内置模型，无需单独下载），用于对RAG检索结果重排。

"""
创建ReRanker模型，目前推荐使用BGE-Reranker-V2-M3
"""
def create_reranker_model(self):
    if self.check_model_exists(self.parent.reranker_model_name):
        print("Model {} exists, skip create".format(self.parent.reranker_model_name))
        return
    
    self.common_create_model(self.parent.reranker_model_name,
                             "huggingface://BAAI/bge-reranker-v2-m3",
                             "SEMANTIC_SIMILARITY",
                             "BGE_RERANKER_V2_M3")


"""
根据问题以及目前答案的候选集，对答案进行重新排序
* input_text: 输入的问题
* chunks: 答案列表
"""
def reranker(self, input_text: str, chunks: List[str]):
    url = "http://{}:{}/v1/ai/models/{}/infer".format(Config.AI_HOST,
                                                      Config.AI_PORT, 
                                                      self.parent.reranker_model_name)        
    data = {
        "input": {"query": input_text, "chunks": chunks}
    }
    
    response = requests.post(url, data=json.dumps(data), headers=self.headers)
    json_response = response.json()
    if response.status_code != 200 or json_response["success"] is False:
        raise Exception("http request failed, status code: {}".format(json_response["msg"]))
    return json_response["data"]

AI引擎

调用问答模型：

通过API Key的方式访问大模型，将提示词（Prompt）提交给大模型。API Key可通过百炼平台申请，详情请参见获取API Key。

"""
使用api_key访问通义千问的方式
"""
# reference: https://help.aliyun.com/zh/dashscope/developer-reference/qwen-api
class AliQwen():
    def __init__(self):
        self.api_key = Config.DASHSCOPE_API_KEY
        self.model_name = "qwen-turbo"
        self.PROMPT_TEMPLATE = """已知信息：
{context} 
根据上述已知信息，专业的来回答用户的问题。如果无法从中得到答案，请说 “根据已知信息无法回答该问题” 或 “没有提供足够的相关信息”，不允许在答案中添加编造成分，答案请使用中文。 问题是：{question}"""
    
    """
    非流式对话大模型
    """
    def chat(self, prompt: str):
        response = Generation.call(model=self.model_name, prompt=prompt, stream=False, api_key=self.api_key)
        if response.status_code == HTTPStatus.OK:
            return response.output.text
        else:
            raise Exception(response.message)
    
    """
    流式对话大模型
    """
    def chat_stream(self, prompt: str):
        responses = Generation.call(model=self.model_name, prompt=prompt, stream=True, api_key=self.api_key)
        for response in responses:
            if response.status_code == HTTPStatus.OK:
                yield response.output.text
            else:
                raise Exception(response.message)
    
    """
    问题与相关提示一起组装
    """
    def gen_prompt(self, query: str, context: str):
        return self.PROMPT_TEMPLATE.replace("{question}", query).replace("{context}", context)

-

创建搜索Pipeline

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

"""
创建Pipeline,搜索内部自动对text字段调用AI引擎进行Embedding,写入vector_field 字段
"""
def create_pipeline(self):
    if self.check_pipeline_exists():
        print("Pipeline {} exists".format(self.parent.pipeline_name))
        # 如果pipeline已经存在，目前策略是跳过，如果是需要调整参数重新创建，则注释掉下方的return
        return
    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://{}:{}".format(inner_ai_host, int(Config.AI_PORT)),
                    "modeName": self.parent.embedding_model_name
                }
            }
        ]
    }   
    try:
        response = self.client.ingest.put_pipeline(id=self.parent.pipeline_name, body=pipeline)
        print("Create pipeline success", response)
    except Exception as e:
        print("Create pipeline errr ", e) 

宽表引擎

搜索引擎

向量引擎

创建搜索索引和向量索引

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

示例中使用HNSW，如果数据量大，推荐使用IVF_PQ。小数据集则可以使用FLAT。

"""
子表创建搜索索引，本文范例中使用的是HNSW索引，如果数据量较大，建议使用IVF_PQ索引
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": "hnsw",
                    "space_type": "cosinesimil",
                    "parameters": {
                        "ef_construction": 500,
                        "m": 24
                        }
                    }
            }')) WITH (INDEX_SETTINGS='{
            "default_pipeline": "%s",
            "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,
                    self.parent.pipeline_name)
    print("Create search index sql: \n ", sql)
    self.common_create_table(sql)

宽表引擎

搜索引擎

向量引擎

检索方式

具体描述

流程图

全文与向量混合检索

全文与向量混合检索。

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(item['chunk'] for item in reranker_result[0:topk])
    ali_qwen = AliQwen()
    prompt = ali_qwen.gen_prompt(query, prompt_context)
    output_text = ""
    for part in ali_qwen.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，并通过API Key访问大模型，获得回答。

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)        
    ali_qwen = AliQwen()
    prompt = ali_qwen.gen_prompt(query, prompt_context)    
    # stream
    for part in ali_qwen.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()