Lindorm:Démarrage rapide : Création d'un service de recherche intelligente

Étape

Description

Moteurs impliqués

Créer une table parente (base de connaissances)

Créez une table parente pour stocker le texte original de la base de connaissances.

"""
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)

moteur wide-column

Créer une table enfant (base de connaissances segmentée)

Créez une table enfant pour stocker les fragments de texte après le découpage du texte original. Vous pouvez segmenter le texte par longueur ou par une combinaison de longueur et de ponctuation.

"""
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)

moteur wide-column

Créer un pipeline de recherche

Créez un pipeline d'ingestion et un pipeline de recherche pour la table enfant. Lors de l'ingestion, le contenu de la table enfant est automatiquement synchronisé avec le moteur de recherche, qui génère des embeddings et les écrit dans le moteur vectoriel.

Lors d'une requête, le texte est automatiquement intégré (embedded) et la récupération vectorielle est effectuée via le moteur vectoriel.

"""
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}")

moteur wide-column

moteur de recherche

moteur vectoriel

Créer des index de recherche et vectoriels

Créez des index de recherche et vectoriels pour améliorer l'efficacité de la récupération.

Cet exemple utilise un index IVFBQ. Si votre jeu de données est petit, vous pouvez utiliser un index HNSW.

"""
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)

moteur wide-column

moteur de recherche

moteur vectoriel

Configurer les pipelines pour l'index de recherche

Configurez les pipelines d'ingestion et de recherche par défaut pour l'index de recherche.

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}")

moteur de recherche

moteur vectoriel

Créer un index vectoriel

Pour les index IVFPQ et IVFBQ, vous devez déclencher manuellement la construction de l'index après avoir écrit une quantité suffisante de données. Une fois la construction terminée, vous pouvez écrire et interroger les données sans reconstruction. Pour plus d'informations, consultez Créer un 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

index full-text

index vectoriel

Méthode de récupération

Description

Flux de travail

Récupération hybride full-text et vectorielle

Combine les résultats de la récupération full-text et de la récupération vectorielle.

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()

Récupération hybride full-text et vectorielle + réorganisation

Effectue une récupération hybride, puis réorganise les résultats.

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()

Récupération hybride full-text et vectorielle + réorganisation + invite

1. Effectuez d'abord une récupération hybride full-text et vectorielle, puis réorganisez les résultats.

2. Combinez la question et les résultats de récupération dans une invite, puis soumettez-la au grand modèle.

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()

Récupération hybride full-text et vectorielle + réorganisation + Invite contextuelle

1. Effectuez d'abord une récupération hybride full-text et vectorielle, puis réorganisez les résultats.

2. Interrogez la table parente en utilisant la PRIMARY KEY issue des résultats de récupération pour récupérer le contexte original.

3. Combinez la question et le contexte pour former une invite, puis envoyez-la au grand modèle via le moteur d'IA pour obtenir une réponse.

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()