本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

搭配機器學習離線批次推論使用 OpenSearch 擷取管道

Amazon OpenSearch Ingestion (OSI) 管道支援機器學習 (ML) 離線批次推論處理，以低成本有效率地充實大量資料。每當您有可非同步處理的大型資料集時，請使用離線批次推論。離線批次推論適用於 Amazon Bedrock 和 SageMaker 模型。此功能適用於所有支援 OpenSearch Service 2 AWS 區域 .17+ 網域的 OpenSearch Ingestion。

離線批次推論處理會利用稱為 ML Commons 的 OpenSearch 功能。ML Commons 透過傳輸和 REST API 呼叫提供 ML 演算法。這些呼叫會為每個 ML 請求選擇適當的節點和資源，並監控 ML 任務以確保正常執行時間。如此一來，ML Commons 可讓您利用現有的開放原始碼 ML 演算法，並減少開發新的 ML 功能所需的工作量。如需 ML Commons 的詳細資訊，請參閱 OpenSearch.org 文件中的機器學習。

運作方式

您可以在 OpenSearch Ingestion 上建立離線批次推論管道，方法是將機器學習推論處理器新增至管道。此處理器可讓您的管道連線至 SageMaker 等 AI 服務，以執行批次推論任務。您可以將處理器設定為透過目標網域上執行的 AI 連接器 （使用 batch_predict 支援） 連接到所需的 AI 服務。

OpenSearch Ingestion 使用具有 ML Commons 的ml_inference處理器來建立離線批次推論任務。ML Commons 接著會使用 batch_predict API，該 API 會使用部署在 Amazon Bedrock、Amazon SageMaker、Cohere 和 OpenAI 中外部模型伺服器上的模型，在離線非同步模式下對大型資料集執行推論。下圖顯示 OpenSearch Ingestion 管道，可協調多個元件以端對端執行此程序：

管道元件的運作方式如下：

管道 1 （資料準備和轉換）*：

管道 2 （觸發 ML batch_inference)：

管道 3 （大量擷取）：

關於 ml_inference 處理器

OpenSearch Ingestion 使用 S3 掃描來源和 ML 推論處理器之間的特殊整合來進行批次處理。S3 掃描以僅限中繼資料模式運作，可有效率地收集 S3 檔案資訊，而無需讀取實際的檔案內容。ml_inference 處理器使用 S3 檔案 URLs與 ML Commons 協調以進行批次處理。此設計可將掃描階段期間不必要的資料傳輸降至最低，以最佳化批次推論工作流程。您可以使用參數定義ml_inference處理器。請見此處範例：

processor:
    - ml_inference:
        # The endpoint URL of your OpenSearch domain
        host: "https://AWS test-offlinebatch-123456789abcdefg.us-west-2.es.amazonaws.com"
        
        # Type of inference operation:
        # - batch_predict: for batch processing
        # - predict: for real-time inference
        action_type: "batch_predict"
        
        # Remote ML model service provider (Amazon Bedrock or SageMaker)
        service_name: "bedrock"
        
        # Unique identifier for the ML model
        model_id: "AWS TestModelID123456789abcde"
        
        # S3 path where batch inference results will be stored
        output_path: "s3://amzn-s3-demo-bucket/"
      
        # Supports ISO_8601 notation strings like PT20.345S or PT15M
        # These settings control how long to keep your inputs in the processor for retry on throttling errors
        retry_time_window: "PT9M"
        
        # AWS configuration settings
        aws:
            # AWS 區域 where the Lambda function is deployed
            region: "us-west-2"
            # IAM role ARN for Lambda function execution
            sts_role_arn: "arn:aws::iam::account_id:role/Admin"
        
        # Dead-letter queue settings for storing errors
        dlq:
          s3:
            region: us-west-2
            bucket: batch-inference-dlq
            key_path_prefix: bedrock-dlq
            sts_role_arn: arn:aws:iam::account_id:role/OSI-invoke-ml
            
        # Conditional expression that determines when to trigger the processor
        # In this case, only process when bucket matches "amzn-s3-demo-bucket"
        ml_when: /bucket == "amzn-s3-demo-bucket"

使用 ml_inference 處理器改善擷取效能

OpenSearch Ingestion ml_inference 處理器可大幅增強啟用 ML 搜尋的資料擷取效能。處理器非常適合需要機器學習模型產生資料的使用案例，包括語意搜尋、多模式搜尋、文件擴充和查詢理解。在語意搜尋中，處理器可以按數量級加速大量、高維度向量的建立和擷取。

相較於即時模型調用，處理器的離線批次推論功能具有獨特的優勢。雖然即時處理需要容量限制的即時模型伺服器，但批次推論會隨需動態擴展運算資源，並平行處理資料。例如，當 OpenSearch Ingestion 管道收到十億個來源資料請求時，它會為 ML 批次推論輸入建立 100 個 S3 檔案。然後，ml_inference處理器使用 100 個 ml.m4.xlarge Amazon Elastic Compute Cloud (Amazon EC2) 執行個體啟動 SageMaker 批次任務，在 14 小時內完成十億個請求的向量化，這項任務幾乎不可能在即時模式下完成。

設定 ml_inference 處理器以擷取語意搜尋的資料請求

下列程序會逐步引導您設定和設定 OpenSearch Ingestion ml_inference 處理器，以使用文字內嵌模型擷取 10 億個資料請求以進行語意搜尋。

步驟 1：在 OpenSearch 中建立連接器並註冊模型

針對下列程序，請使用 ML Commons batch_inference_sagemaker_connector_blueprint 在 Amazon SageMaker 中建立連接器和模型。如果您偏好使用 OpenSearch CloudFormation 整合範本，請參閱本節（替代程序） 步驟 1：使用 CloudFormation 整合範本建立連接器和模型稍後的 。

（替代程序） 步驟 1：使用 CloudFormation 整合範本建立連接器和模型

如果您願意，您可以使用 AWS CloudFormation 自動建立 ML 推論所需的所有必要 Amazon SageMaker 連接器和模型。此方法使用 Amazon OpenSearch Service 主控台中可用的預先設定範本來簡化設定。如需詳細資訊，請參閱使用 CloudFormation 設定語意搜尋的遠端推論。

步驟 2：為 ML 離線批次推論建立 OpenSearch 擷取管道

使用下列範例建立適用於 ML 離線批次推論的 OpenSearch 擷取管道。如需建立 OpenSearch Ingestion 管道的詳細資訊，請參閱 建立 Amazon OpenSearch Ingestion 管道。

開始之前

在下列範例中，您會為 sts_role_arn 參數指定 IAM 角色 ARN。使用下列程序來驗證此角色是否對應至可存取 OpenSearch 中 ml-commons 的後端角色。

為 ML 離線批次推論建立 OpenSearch 擷取管道的範例

version: '2'
extension:
  osis_configuration_metadata:
    builder_type: visual
sagemaker-batch-job-pipeline:
  source:
    s3:
      acknowledgments: true
      delete_s3_objects_on_read: false
      scan:
        buckets:
          - bucket:
              name: name
              data_selection: metadata_only
              filter:
                include_prefix:
                  - sagemaker/sagemaker_djl_batch_input
                exclude_suffix:
                  - .manifest
          - bucket:
              name: name
              data_selection: data_only
              filter:
                include_prefix:
                  - sagemaker/output/
        scheduling:
          interval: PT6M
      aws:
        region: name
      default_bucket_owner: account_ID
      codec:
        ndjson:
          include_empty_objects: false
      compression: none
      workers: '1'
  processor:
    - ml_inference:
        host: "https://search-AWStest-offlinebatch-123456789abcdef.us-west-2.es.amazonaws.com"
        aws_sigv4: true
        action_type: "batch_predict"
        service_name: "sagemaker"
        model_id: "model_ID"
        output_path: "s3://AWStest-offlinebatch/sagemaker/output"
        aws:
          region: "us-west-2"
          sts_role_arn: "arn:aws:iam::account_ID:role/Admin"
        ml_when: /bucket == "AWStest-offlinebatch"
        dlq:
          s3:
            region: us-west-2
            bucket: batch-inference-dlq
            key_path_prefix: bedrock-dlq
            sts_role_arn: arn:aws:iam::account_ID:role/OSI-invoke-ml
    - copy_values:
        entries:
          - from_key: /text
            to_key: chapter
          - from_key: /SageMakerOutput
            to_key: chapter_embedding
          - delete_entries:
            with_keys:
          - text
          - SageMakerOutput
  sink:
    - opensearch:
        hosts: ["https://search-AWStest-offlinebatch-123456789abcdef.us-west-2.es.amazonaws.com"]
        aws:
          serverless: false
          region: us-west-2
        routes:
          - ml-ingest-route
        index_type: custom
        index: test-nlp-index
  routes:
    - ml-ingest-route: /chapter != null and /title != null

步驟 3：準備資料以供擷取

若要準備資料以進行 ML 離線批次推論處理，請使用您自己的工具或程序自行準備資料，或使用 OpenSearch Data Prepper。使用管道來取用資料來源中的資料，或建立機器學習資料集，以確認資料組織成正確的格式。

下列範例使用 MS MARCO 資料集，其中包含自然語言處理任務的真實使用者查詢集合。資料集以 JSONL 格式建構，其中每一行代表傳送至 ML 內嵌模型的請求：

{"_id": "1185869", "text": ")what was the immediate impact of the Paris Peace Treaties of 1947?", "metadata": {"world war 2"}}
{"_id": "1185868", "text": "_________ justice is designed to repair the harm to victim, the community and the offender caused by the offender criminal act. question 19 options:", "metadata": {"law"}}
{"_id": "597651", "text": "what is amber", "metadata": {"nothing"}}
{"_id": "403613", "text": "is autoimmune hepatitis a bile acid synthesis disorder", "metadata": {"self immune"}}
...

若要使用 MS MARCO 資料集進行測試，假設您建構分佈於 100 個檔案的 10 億個輸入請求，每個請求都包含 1，000 萬個請求。這些檔案會存放在字首為 s3：//offlinebatch/sagemaker/sagemaker_djl_batch_input/ 的 Amazon S3 中。OpenSearch Ingestion 管道會同時掃描這 100 個檔案，並啟動具有 100 個工作者的 SageMaker 批次任務以進行平行處理，從而實現高效的向量化，並將十億份文件擷取到 OpenSearch。

在生產環境中，您可以使用 OpenSearch 擷取管道來產生用於批次推論輸入的 S3 檔案。管道支援各種資料來源，並依排程運作，以持續將來源資料轉換為 S3 檔案。然後，AI 伺服器會透過排定的離線批次任務自動處理這些檔案，以確保持續資料處理和擷取。

步驟 4：監控批次推論任務

您可以使用 SageMaker 主控台或 監控批次推論任務 AWS CLI。您也可以使用 Get Task API 來監控批次任務：

GET /_plugins/_ml/tasks/_search
{
  "query": {
    "bool": {
      "filter": [
        {
          "term": {
            "state": "RUNNING"
          }
        }
      ]
    }
  },
  "_source": ["model_id", "state", "task_type", "create_time", "last_update_time"]
}

API 會傳回作用中批次任務的清單：

{
  "took": 2,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 3,
      "relation": "eq"
    },
    "max_score": 0.0,
    "hits": [
      {
        "_index": ".plugins-ml-task",
        "_id": "nyWbv5EB_tT1A82ZCu-e",
        "_score": 0.0,
        "_source": {
          "model_id": "nyWbv5EB_tT1A82ZCu-e",
          "state": "RUNNING",
          "task_type": "BATCH_PREDICTION",
          "create_time": 1725496527958,
          "last_update_time": 1725496527958
        }
      },
      {
        "_index": ".plugins-ml-task",
        "_id": "miKbv5EB_tT1A82ZCu-f",
        "_score": 0.0,
        "_source": {
          "model_id": "miKbv5EB_tT1A82ZCu-f",
          "state": "RUNNING",
          "task_type": "BATCH_PREDICTION",
          "create_time": 1725496528123,
          "last_update_time": 1725496528123
        }
      },
      {
        "_index": ".plugins-ml-task",
        "_id": "kiLbv5EB_tT1A82ZCu-g",
        "_score": 0.0,
        "_source": {
          "model_id": "kiLbv5EB_tT1A82ZCu-g",
          "state": "RUNNING",
          "task_type": "BATCH_PREDICTION",
          "create_time": 1725496529456,
          "last_update_time": 1725496529456
        }
      }
    ]
  }
}

步驟 5：執行搜尋

監控批次推論任務並確認已完成之後，您可以執行各種類型的 AI 搜尋，包括語意、混合、對話 （使用 RAG)、神經稀疏和多模態。如需 OpenSearch Service 支援的 AI 搜尋詳細資訊，請參閱 AI 搜尋。

若要搜尋原始向量，請使用knn查詢類型，提供vector陣列做為輸入，並指定傳回的結果k數目：

GET /my-raw-vector-index/_search
{
  "query": {
    "knn": {
      "my_vector": {
        "vector": [0.1, 0.2, 0.3],
        "k": 2
      }
    }
  }
}

若要執行 AI 支援的搜尋，請使用 neural 查詢類型。指定query_text輸入、您在 OpenSearch Ingestion 管道中設定的內嵌模型model_id的 ，以及傳回的結果k數目。若要從搜尋結果中排除內嵌，請在 _source.excludes 參數中指定內嵌欄位的名稱：

GET /my-ai-search-index/_search
{
  "_source": {
    "excludes": [
      "output_embedding"
    ]
  },
  "query": {
    "neural": {
      "output_embedding": {
        "query_text": "What is AI search?",
        "model_id": "mBGzipQB2gmRjlv_dOoB",
        "k": 2
      }
    }
  }
}