[OpenSearch Service] ML Commons(AI Connectors) を Workflow API で構築して、Slackメッセージのハイブリッド検索を作成してみました

create_connector
       ↓ connector_id
register_model
       ↓ model_id
deploy_model
       ↓ model_id
create_ingest_pipeline
       ↓ pipeline_id
create_index
       ↓ index_name
create_search_pipeline

$ git clone https://github.com/furuya02/slack-hybrid-search-workflow.git
$ cd slack-hybrid-search-workflow
$ cd cdk
$ pnpm install
$ pnpm cdk deploy

def lambda_handler(event, context):
    body = json.loads(event.get('body', '{}'))

    # URL 検証（Slack App 設定時）
    if body.get('type') == 'url_verification':
        return {'statusCode': 200, 'body': json.dumps({'challenge': body.get('challenge')})}

    # メッセージイベント処理
    if body.get('type') == 'event_callback':
        e = body.get('event', {})
        if e.get('type') != 'message' or e.get('bot_id') or e.get('subtype') or not e.get('text'):
            return {'statusCode': 200, 'body': '{}'}

        doc = {
            'message_id': e.get('client_msg_id'),
            'channel_id': e.get('channel'),
            'user_id': e.get('user'),
            'text': e.get('text'),
            'timestamp': e.get('ts')
        }
        # Ingest Pipeline を指定してインデックス（ベクトル化は OpenSearch 内部で自動実行）
        get_client().index(index=INDEX, body=doc, pipeline=PIPELINE, refresh=True)

    return {'statusCode': 200, 'body': '{}'}

def lambda_handler(event, context):
    body = json.loads(event.get('body', '{}'))
    query, mode, size = body.get('query', ''), body.get('mode', 'hybrid'), body.get('size', 10)

    neural = {'neural': {'text_embedding': {'query_text': query, 'model_id': MODEL_ID, 'k': size}}}

    if mode == 'keyword':
        q = {'match': {'text': query}}
        res = get_client().search(index=INDEX, body={'size': size, 'query': q})
    elif mode == 'vector':
        res = get_client().search(index=INDEX, body={'size': size, 'query': neural})
    else:  # hybrid
        q = {'hybrid': {'queries': [{'match': {'text': query}}, neural]}}
        res = get_client().search(index=INDEX, body={'size': size, 'query': q},
                                  params={'search_pipeline': PIPELINE})
    # ... 結果を整形して返却

$ cd slack-hybrid-search-workflow
$ ./scripts/setup-workflow-api.sh

# スクリプトの要点
WORKFLOW_JSON=$(cat "$SCRIPT_DIR/workflow-template.json" | \
    sed "s|\${BEDROCK_ROLE_ARN}|$BEDROCK_ROLE_ARN|g" | \
    sed "s|\${AWS_REGION}|$AWS_REGION|g")

curl -s -X POST \
    "https://${DOMAIN_ENDPOINT}/_plugins/_flow_framework/workflow?provision=true" \
    --aws-sigv4 "aws:amz:$AWS_REGION:es" \
    --user "$AWS_ACCESS_KEY_ID:$AWS_SECRET_ACCESS_KEY" \
    -H "Content-Type: application/json" \
    -H "x-amz-security-token: $AWS_SESSION_TOKEN" \
    -d "$WORKFLOW_JSON"

{
  "workflow_id": "dv_zYZ0BS3ey_-URlTo8"
}

{
  "id": "create_connector",
  "type": "create_connector",
  "user_inputs": {
    "name": "Bedrock Titan Connector",
    "protocol": "aws_sigv4",
    "credential": { "roleArn": "${BEDROCK_ROLE_ARN}" },
    "parameters": {
      "region": "${AWS_REGION}",
      "service_name": "bedrock",
      "model": "amazon.titan-embed-text-v2:0"
    },
    "actions": [{
      "action_type": "predict",
      "method": "POST",
      "url": "https://bedrock-runtime.${AWS_REGION}.amazonaws.com/model/amazon.titan-embed-text-v2:0/invoke",
      "request_body": "{ \"inputText\": \"${parameters.inputText}\", \"dimensions\": 1024, \"normalize\": true }",
      "pre_process_function": "...",
      "post_process_function": "..."
    }]
  }
}

Neural Search の内部形式           Bedrock が期待する形式
{ "text_docs": ["会議の議事録"] }  →  { "parameters": { "inputText": "会議の議事録" } }

StringBuilder builder = new StringBuilder();
builder.append("\"");
String first = params.text_docs[0];    // 配列の最初の要素を取得
builder.append(first);
builder.append("\"");
def parameters = "{" +"\"inputText\":" + builder + "}";
return  "{" +"\"parameters\":" + parameters + "}";

def name = "sentence_embedding";
def dataType = "FLOAT32";
if (params.embedding == null || params.embedding.length == 0) {
    return params.message;
}
def shape = [params.embedding.length];
def json = "{" +
           "\"name\":\"" + name + "\"," +
           "\"data_type\":\"" + dataType + "\"," +
           "\"shape\":" + shape + "," +
           "\"data\":" + params.embedding +
           "}";
return json;

{
  "id": "register_model",
  "type": "register_remote_model",
  "user_inputs": { "name": "Titan Embeddings V2", "function_name": "remote" },
  "previous_node_inputs": { "create_connector": "connector_id" }
},
{
  "id": "deploy_model",
  "type": "deploy_model",
  "previous_node_inputs": { "register_model": "model_id" }
}

{
  "id": "create_ingest_pipeline",
  "type": "create_ingest_pipeline",
  "user_inputs": {
    "pipeline_id": "slack-ingest-pipeline",
    "configurations": {
      "processors": [{
        "text_embedding": {
          "model_id": "${{deploy_model.model_id}}",
          "field_map": { "text": "text_embedding" }
        }
      }]
    }
  },
  "previous_node_inputs": { "deploy_model": "model_id" }
}

{
  "id": "create_index",
  "type": "create_index",
  "user_inputs": {
    "index_name": "slack-messages",
    "configurations": {
      "settings": { "index": { "knn": true, "default_pipeline": "slack-ingest-pipeline" } },
      "mappings": { "properties": {
        "message_id": { "type": "keyword" },
        "channel_id": { "type": "keyword" },
        "user_id": { "type": "keyword" },
        "text": { "type": "text" },
        "text_embedding": {
          "type": "knn_vector", "dimension": 1024,
          "method": { "name": "hnsw", "engine": "faiss", "space_type": "l2" }
        },
        "timestamp": { "type": "keyword" },
        "thread_ts": { "type": "keyword" },
        "team_id": { "type": "keyword" },
        "event_time": { "type": "long" }
      }}
    }
  }
}

{
  "id": "create_search_pipeline",
  "type": "create_search_pipeline",
  "user_inputs": {
    "pipeline_id": "hybrid-search-pipeline",
    "configurations": {
      "phase_results_processors": [{
        "normalization-processor": {
          "normalization": { "technique": "min_max" },
          "combination": { "technique": "arithmetic_mean", "parameters": { "weights": [0.3, 0.7] } }
        }
      }]
    }
  }
}

$ curl -s -X GET \
    "https://${DOMAIN_ENDPOINT}/_plugins/_flow_framework/workflow/${WORKFLOW_ID}/_status" \
    --aws-sigv4 "aws:amz:ap-northeast-1:es" \
    --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" \
    -H "x-amz-security-token: ${AWS_SESSION_TOKEN}" | jq .

{
  "state": "COMPLETED",
  "resources_created": [
    { "workflow_step_name": "create_connector", "resource_id": "<connector_id>" },
    { "workflow_step_name": "register_model", "resource_id": "<model_id>" },
    { "workflow_step_name": "deploy_model", "resource_id": "<model_id>" },
    { "workflow_step_name": "create_ingest_pipeline", "resource_id": "slack-ingest-pipeline" },
    { "workflow_step_name": "create_index", "resource_id": "slack-messages" },
    { "workflow_step_name": "create_search_pipeline", "resource_id": "hybrid-search-pipeline" }
  ]
}

$ aws lambda update-function-configuration \
    --function-name SlackHybridSearch-Search \
    --environment "Variables={OPENSEARCH_ENDPOINT=<DomainEndpoint>,INDEX_NAME=slack-messages,SEARCH_PIPELINE=hybrid-search-pipeline,MODEL_ID=<model_id>}"

$ ./scripts/load-sample-data.sh

# API エンドポイントを取得
$ API_ENDPOINT=$(aws cloudformation describe-stacks \
    --stack-name SlackHybridSearchStack \
    --query "Stacks[0].Outputs[?OutputKey=='ApiEndpoint'].OutputValue" \
    --output text)

$ curl -s -X POST "${API_ENDPOINT}search" \
    -H "Content-Type: application/json" \
    -d '{"query": "会議", "mode": "keyword"}' \
    | jq -r '.results[] | "\(.score | tostring | .[0:6]) | \(.text)"'

$ curl -s -X POST "${API_ENDPOINT}search" \
    -H "Content-Type: application/json" \
    -d '{"query": "パフォーマンスを改善したい", "mode": "vector"}' \
    | jq -r '.results[] | "\(.score | tostring | .[0:6]) | \(.text)"'

$ curl -s -X POST "${API_ENDPOINT}search" \
    -H "Content-Type: application/json" \
    -d '{"query": "Lambda が遅い", "mode": "hybrid"}' \
    | jq -r '.results[] | "\(.score | tostring | .[0:6]) | \(.text)"'

# 全ドキュメントを削除
$ curl -s -X POST "https://${DOMAIN_ENDPOINT}/slack-messages/_delete_by_query" \
    --aws-sigv4 "aws:amz:ap-northeast-1:es" \
    --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" \
    -H "x-amz-security-token: ${AWS_SESSION_TOKEN}" \
    -H "Content-Type: application/json" \
    -d '{"query": {"match_all": {}}}' | jq .

# 削除確認（0件になっていることを確認）
$ curl -s -X GET "https://${DOMAIN_ENDPOINT}/slack-messages/_count" \
    --aws-sigv4 "aws:amz:ap-northeast-1:es" \
    --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" \
    -H "x-amz-security-token: ${AWS_SESSION_TOKEN}" | jq .

今日の定例会議は15時からです。議事録は後で共有します。

$ curl -s -X GET "https://${DOMAIN_ENDPOINT}/slack-messages/_count" \
    --aws-sigv4 "aws:amz:ap-northeast-1:es" \
    --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" \
    -H "x-amz-security-token: ${AWS_SESSION_TOKEN}" | jq .

# キーワード検索
$ curl -s -X POST "${API_ENDPOINT}search" \
    -H "Content-Type: application/json" \
    -d '{"query": "定例会議", "mode": "keyword"}' \
    | jq -r '.results[] | "\(.score | tostring | .[0:6]) | \(.text)"'

# ハイブリッド検索（意味的にも検索）
$ curl -s -X POST "${API_ENDPOINT}search" \
    -H "Content-Type: application/json" \
    -d '{"query": "ミーティングの予定", "mode": "hybrid"}' \
    | jq -r '.results[] | "\(.score | tostring | .[0:6]) | \(.text)"'