Quick start: Build an intelligent search service

Step

Description

Engines involved

Create a parent table (knowledge base)

Create a parent table to store the original text of the knowledge base.

"""
SQL statement to create the parent table. Modify it as needed.
"""
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)

wide-column engine

Create a child table (split knowledge base)

Create a child table to store the text chunks after splitting the original text. You can split the text by length or by a combination of length and punctuation.

"""
SQL statement to create the child table. Modify it as needed.
"""
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)

wide-column engine

Create a search pipeline

Create an ingest pipeline and a search pipeline for the child table. During ingestion, the child table content is automatically synced to the search engine. The search engine generates embeddings and then writes the results to the vector engine.

During a query, the text is automatically embedded, and vector retrieval is performed through the vector engine.

"""
Create a pipeline. The search engine automatically calls the AI engine to generate embeddings for the text field and writes them to the 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}")

    # Create a search pipeline for queries.
    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}")

wide-column engine

search engine

vector engine

Create search and vector indexes

Create search and vector indexes to improve retrieval efficiency.

This example uses an IVFBQ index. If your dataset is small, you can use an HNSW index.

"""
Create a search index on the child table. This example uses an IVFBQ index.
https://www.alibabacloud.com/help/document_detail/2773371.html
"""
# Reference: https://www.alibabacloud.com/help/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)

wide-column engine

search engine

vector engine

Configure pipelines for the search index

Configure the default ingest and search pipelines for the search index.

def set_index_pipelines(self):
    """
    Use the Elasticsearch API to set the default ingest and search pipelines for the index.
    """
    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}")

search engine

vector engine

Build a vector index

For IVFPQ and IVFBQ indexes, you must manually trigger an index build after writing a sufficient amount of data. Once the build is complete, you can write and query data without needing to rebuild. For more information, see Build an index.

def build_vector_index(self):
    """Trigger vector index building and wait for completion."""
        
    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"Starting to build vector index: {index_name}.{field_name}")
        
    # 1. Trigger the build.
    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"Trigger build response: {resp.status_code}, {resp.text}")
            
        # 2. Poll to check the status.
        task_url = f"{url_base}/_plugins/_vector/index/tasks"
        task_data = {
            "indexName": index_name,
            "fieldName": field_name,
            "taskIds": "[]"
        }
            
        max_wait = 300  # Wait for a maximum of 5 minutes.
        start_time = time.time()
            
        while time.time() - start_time < max_wait:
            time.sleep(5)  # Check every 5 seconds.
                
            resp = requests.get(task_url, auth=auth, json=task_data, headers={"Content-Type": "application/json"})
            result = resp.json()
            print(f"Build status: {result}")
                
            if "payload" in result:
                for task in result["payload"]:
                    if "FINISH" in task:
                        print("Vector index build complete!")
                        return True
                    elif "FAILED" in task:
                        print(f"Vector index build failed: {task}")
                        return False
                            
        print("Build timed out.")
        return False
            
    except Exception as e:
        print(f"Error building vector index: {e}")
        return False

full-text index

vector index

Retrieval method

Description

Workflow

Hybrid full-text and vector retrieval

Combines results from both full-text and vector retrieval.

def demo_rrf_search():
    lindorm = Lindorm()
    query="Main subjects of the International Junior Science Olympiad"
    results = lindorm.lindormSearch.rrf_search(query)
    display(JSON(results, expanded=True, root="rrf_search_result"))
    lindorm.close()
demo_rrf_search()

Hybrid full-text and vector retrieval + reranking

Performs a hybrid retrieval and then reranks the results.

def demo_rerank():
    lindorm = Lindorm()
    query="Main subjects of the International Junior Science Olympiad"
    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()

Hybrid full-text and vector retrieval + reranking + prompt

1. First, perform hybrid full-text and vector retrieval, and then rerank the results.

2. Combines the question and retrieval results into a prompt and submits it to the large model.

def demo_chat_with_child_chunking():
    lindorm = Lindorm()
    topk=int(Config.SEARCH_TOP_K)
    query="Main subjects of the International Junior Science Olympiad"
    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>The prompt template is:\n{wrapped_text}</pre>"))
    lindorm.close()
    
demo_chat_with_child_chunking()

Hybrid full-text and vector retrieval + reranking + Context Prompt

1. First, perform hybrid full-text and vector retrieval, and then rerank the results.

2. Queries the parent table using the PRIMARY KEY from the retrieval results to fetch the original context.

3. Combines the question and context to form a prompt, then sends it to the large model through the AI engine to get an answer.

def demo_chat_with_parent():
    lindorm = Lindorm()
    topk=int(Config.SEARCH_TOP_K)
    query="Main subjects of the International Junior Science Olympiad"
    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>The prompt template is:\n{wrapped_text}</pre>")) 
    lindorm.close()

demo_chat_with_parent()