ServiceNow × Databricksで作る類似インシデント検索― 分析はData Lakeに任せる①

スバルヒーロー

2026年4月26日 09:22

はじめに

「ServiceNowにAI Agentを実装したい」という声をよく聞きます。しかし、いざ設計を始めると壁に当たります。

・真の価値のあるユースケースが、ServiceNow単体では成立しにくい

これまで当ブログでは、ServiceNowの分析業務において、様々な観点で切り込んだ記事を掲載してきました。

今回はさらにAI Agentの領域に踏み込みます。

本記事では、その構造的理由を正確に解説した上で、データ分析基盤のDatabricksと組み合わせることで初めて実現できる「類似インシデント検索 Agentic Workflow」を実装します。

対象読者：

以下のような方々に、読んで頂けると幸いです。

ServiceNowのAI Agentに可能性を感じているが、何を作るべきか迷っている技術者
Databricksの実用的なユースケースを模索しているデータエンジニア
ITSM領域でのAI活用を検討している意思決定者

Part 1: Why ― なぜServiceNow単体のAIは分析処理の限界にぶち当たるのか

1-1. 「AIが必要な判断」と「ルールで十分な判断」を区別してみよう

まず根本的な問いから始めます。

”インシデント対応業務” にて、AIが必要な場面とはどんな時ですか？

答えは、意外とシンプルです。

・「過去のデータを横断的に分析しないと、正しい判断ができない場面」です。

逆に言えば：

承認フローの自動化 → 既存のルールベースで十分
SLAタイマーの監視 → GlideAggregateで十分
キーワードで近いインシデントを探す → AI Searchで十分 ※後述

真にAIが必要なのは、「このインシデント、過去に似たものはあったか。意味的に近いものを見つけ、解決手順を再利用できるか」というような、意味理解を伴う横断検索なのです。

平易に述べると ”履歴レコードデータの仕分け分析” がそこに相当するかと思います。

1-2. ServiceNow単体で試みると何が起きるか

① AI Search（Now Platform）

ServiceNowのAI Searchはセマンティック検索への対応を本格化しており、Moveworksの買収によってコンテキスト認識力も強化される方向にあります。進化の方向性は正しいかなと。

ただし今回のユースケース――「過去インシデント全量を横断してcosine similarityで類似度を数値化し、閾値判定でフローを分岐させる」――には現時点では対応していません。

検索結果を構造化スコアとして取り出し、プログラマブルに扱える形にはなっていないからです。

「検索して表示する」と「スコアを取得してフローを制御する」は、要件として別物です。

② Predictive Intelligence（PI）のSimilarity Framework

PIにはSimilarity Frameworkという機能が存在します。

これはMLモデルを使って類似ケースを推薦してくれる機能です。

「あれ？　これ、うまく使えるんじゃないか！？」　と期待値が上がりますが、、しかし実運用に乗せるには、以下のコンディションが求められます：

https://www.servicenow.com/community/platform-analytics-forum/predictive-intelligence-data-limit/m-p/1225663

結果、「類似事象だけ抽出するのに、この手間はかけたくないよなぁ」という理由にょり、現場での利用を避けられてしまう傾向に陥ります。

③ なぜServiceNowは分析基盤を持てないのか

続いて、データ分析が苦手となる本質論に入ります。

ServiceNowのデータアクセス層の中心にあるのは GlideRecord と ACL（Access Control List）が存在します。

この2つは密結合しており、データアクセスはガバナンス観点からの強力なACLが前面にそびえ立ち、SQLアクセスを禁じGlideRecord APIからのアクセスのみを許可するといった構造です。

この構造の特性こそが、高度なクエリやベクトル計算を動かすための分析エンジンを内部に持つことを難しくさせていると考えられます。
※これについてはこちらでも記載してます。

もちろん、ServiceNowはこの課題についても既に認識しており、何度か当ブログでも記載しましたが、分析プラットフォームベンダーの"Pyramid Analytics"を買収するなどでこの課題に向けて取り組み始めています。

しかし、現時点（2026年4月時）ではまだこの課題解決に向けたアプローチが取れないのが現実です。

要約すると：

ServiceNowは「プロセスを動かす基盤」として強力に設計されており、「データを横断分析する基盤」からは遠い構造。

よって、ServiceNowを最大に使いこなすイメージは以下です。

・分析が不要な判断はルールベースで十分。
・分析が必要な判断は外部基盤に移譲。ServiceNowはフロー制御のオーケストレーターとして全体を統制。

こうした役割の明確化と、この認識こそが、正しいアーキテクチャ設計の出発点になるのではと思ってます。

1-3. では、ServiceNow AI Agentの正しい役割とは

ServiceNow AI Agentが真に力を発揮するのは、分析結果を受け取ってプロセスを判断・実行するフェーズです。

[外部分析基盤]          [ServiceNow AI Agent]
データを見る        →    判断してプロセスを動かす
Databricks              Integration Hub + Flow

分析はDatabricksに任せ、ServiceNowはその結果を受け取り、チケットのクローズ、エスカレーション、担当者への通知を自動実行します。

これこそが「AIを正しく使う構造」です。

Part 2: How ― 実装

2-1. アーキテクチャ全体図

今回作成を予定している機能は、以下アーキテクチャー図の通りです。

2-2. Databricks側の実装

Databricksの個人用無償環境の払い出し手順はこちらを参考にしてください。

Step 1: ライブラリのインストール

Databricksのメニューの[新規]→[ノートブック]より、ノートブックを作成します。

以下コマンドを最初のセルに貼り付けます。

%pip install sentence-transformers
dbutils.library.restartPython()

１）sentence-transformers：文や画像をベクトル表現に変換することができる深層学習モデルを使えるライブラリ。
２）restartPython()：Python環境の再起動

Step 2: サンプルインシデントデータをDelta Tableに格納

続いて、Databricks側に類似事象を検証できるサンプルデータをバルクで登録しておきます。

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

spark = SparkSession.builder.getOrCreate()

# サンプルインシデントデータ（日本語）
incidents = [
    (1, "INC001", "メールサーバーが応答しない",       "SMTPサーバー再起動で解消。ポート25の疎通確認が先決。"),
    (2, "INC002", "社内ポータルにアクセスできない",     "DNS設定の誤りが原因。hostsファイルを修正して解消。"),
    (3, "INC003", "VPN接続が切断される",              "認証証明書の期限切れ。証明書更新で解消。"),
    (4, "INC004", "プリンターが認識されない",          "ドライバーの再インストールで解消。"),
    (5, "INC005", "Teamsの音声通話が途切れる",         "ネットワーク帯域の問題。QoS設定を調整して解消。"),
    (6, "INC006", "バックアップジョブが失敗する",       "ディスク容量不足。不要ファイル削除後にジョブ再実行。"),
    (7, "INC007", "Active Directoryへのログインが失敗", "アカウントロックアウト。管理者がロック解除で解消。"),
    (8, "INC008", "社内Wikiが表示されない",            "Webサーバーのメモリ不足。サービス再起動で解消。"),
    (9, "INC009", "OutlookからExchangeに繋がらない",   "Exchange接続設定の誤り。プロファイル再作成で解消。"),
   (10, "INC010", "ファイルサーバーへのアクセスが遅い", "ネットワークスイッチの障害。スイッチ交換で解消。"),
]

schema = StructType([
    StructField("id",          IntegerType(), False),
    StructField("number",      StringType(),  False),
    StructField("short_desc",  StringType(),  False),
    StructField("resolution",  StringType(),  False),
])

df = spark.createDataFrame(incidents, schema)
df.write.format("delta").mode("overwrite").saveAsTable("incidents")

print(f"✅ Delta Table作成完了: {df.count()}件")
df.show(truncate=False)

Delta Tableへ、インシデントのサンプルデータの登録が完了しました。

Step 3: ベクトル埋め込みの事前計算とDelta Tableへの保存

さて、ここから今回のブログ記事の一番コアとなる ”分析処理エンジン” の構築に入ってゆきます。

まず、"paraphrase-multilingual-mpnet-base-v2" はHugging Faceが公開している事前学習済みモデルとなってます。

数億規模の多言語テキストで既にトレーニング済みです。

ここではそのモデルをそのまま使い、インシデントの short_desc テキストを768次元のベクトル（数値の配列）に変換するだけです。

この「テキスト → ベクトル変換」を　"エンコード（埋め込み生成）" と呼びます。

意味的に近い文章は、ベクトル空間上でも近い位置に配置される性質を持ちます。

これを事前計算してDelta Tableに保存しておくことで、新規インシデント到着時の検索高速化を狙うわけですね。

一方で、Predective IntelligenceのSimilarity Frameworkが「自社データで1万件以上のトレーニングが必要」なのに対し、このアーキテクチャは訓練データゼロ・即日稼働が可能な点が根本的に異なります。

比べてみて明確ですね。PI使わずにこちらを最初から使いたくなりますね・・

以下ステップを、[コード]のセルに貼り付けて実行します。

import pandas as pd
from sentence_transformers import SentenceTransformer
import json

# モデルのロード（初回はダウンロードが発生）
model = SentenceTransformer("paraphrase-multilingual-mpnet-base-v2")
print("✅ モデルロード完了")

# インシデントデータの取得
df_pd = spark.table("incidents").toPandas()

# ベクトル化（short_descを対象）
descriptions = df_pd["short_desc"].tolist()
embeddings = model.encode(descriptions, convert_to_tensor=False)

# ベクトルをJSON文字列としてDelta Tableに保存
df_pd["embedding"] = [json.dumps(e.tolist()) for e in embeddings]

# Sparkに戻してDelta Tableに保存
df_with_emb = spark.createDataFrame(df_pd)
df_with_emb.write.format("delta").mode("overwrite").saveAsTable("incidents_with_embeddings")

print(f"✅ ベクトル化完了・保存済み: {len(df_pd)}件")

１）sentence-transformersの事前学習済みモデル（paraphrase-multilingual-mpnet-base-v2）をそのまま使って、インシデントのテキストをベクトル（数値の配列）に変換
２）そのベクトルをDelta Tableに保存

Step 4: 類似検索関数の実装

このステップの狙いは、以下です。

・新規インシデントのテキストと、過去インシデント全量を比較して、意味的に最も近いものをスコア付きで返す関数を作ること

仕組みは3段階に分解：

新規インシデントをベクトル化 — Step 3と同じエンコード処理を、今度はリアルタイムで新規テキストに対して実行
cosine similarityで距離を計測 — 2つのベクトルがどれだけ「同じ方向を向いているか」を-1〜1のスコアで表現します。1.0が完全一致、0.0が無関係、マイナスは逆の意味。
閾値でフィルタ・上位k件を返す — ServiceNow側のフロー分岐に使えるよう、スコアを構造化JSONとして返す。

※cosine similarityを選ぶ理由： ベクトルの「長さ（文章の長短）」ではなく「方向（意味の向き）」だけで比較するため、「メールが届かない」（短文）と「メールサーバーへの接続が確立できずメッセージの送受信が不可能な状態」（長文）も正しく高スコアになる。

import numpy as np
import json
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("paraphrase-multilingual-mpnet-base-v2")

def cosine_similarity(a, b):
    """2つのベクトル間のcosine similarityを計算"""
    a = np.array(a)
    b = np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

def find_similar_incidents(query_text: str, top_k: int = 3, threshold: float = 0.0):
    """
    クエリテキストに対して類似インシデントを返す
    
    Args:
        query_text: 新規インシデントの概要
        top_k: 返す件数
        threshold: 最低スコア閾値（0.0=全件返す）
    
    Returns:
        list of dict: [{number, short_desc, resolution, score}, ...]
    """
    # クエリのベクトル化
    query_embedding = model.encode(query_text, convert_to_tensor=False).tolist()
    
    # Delta Tableからデータ取得
    df_pd = spark.table("incidents_with_embeddings").toPandas()
    
    # 全インシデントとのcosine similarity計算
    results = []
    for _, row in df_pd.iterrows():
        stored_embedding = json.loads(row["embedding"])
        score = cosine_similarity(query_embedding, stored_embedding)
        
        if score >= threshold:
            results.append({
                "number":     row["number"],
                "short_desc": row["short_desc"],
                "resolution": row["resolution"],
                "score":      round(score, 4)
            })
    
    # スコア降順でソートして上位k件を返す
    results.sort(key=lambda x: x["score"], reverse=True)
    return results[:top_k]


# 動作確認
test_queries = [
    "メールが送受信できない",
    "社内システムにログインできない",
    "ネットワークが繋がらない",
]

for query in test_queries:
    print(f"\n🔍 クエリ: 「{query}」")
    results = find_similar_incidents(query, top_k=3, threshold=0.5)
    for r in results:
        marker = "✅" if r["score"] >= 0.8 else "🔶"
        print(f"  {marker} [{r['score']}] {r['number']}: {r['short_desc']}")

１）cosine_similarity()： 2つのベクトルがどれだけ「同じ方向を向いているか」を-1〜1のスコア化を実施
２）model.encode()：クエリのベクトル処理
３）test_queries：今回は３つのクエリを投げて類似性を検証

以下のような結果が返ります。

実行結果（期待値）：

🔍 クエリ: 「メールが送受信できない」
  ✅ [0.8828] INC001: メールサーバーが応答しない
  🔶 [0.7276] INC002: 社内ポータルにアクセスできない
  🔶 [0.6026] INC004: プリンターが認識されない

🔍 クエリ: 「社内システムにログインできない」
  ✅ [0.862] INC002: 社内ポータルにアクセスできない
  🔶 [0.6342] INC001: メールサーバーが応答しない
  🔶 [0.5904] INC008: 社内Wikiが表示されない

🔍 クエリ: 「ネットワークが繋がらない」
  🔶 [0.6924] INC001: メールサーバーが応答しない
  🔶 [0.6676] INC002: 社内ポータルにアクセスできない
  🔶 [0.5478] INC004: プリンターが認識されない

スコア0.8以上は「意味的に同一の問題」と判断できる実用レベルです。

「メール」「サーバー」「接続できない」といった概念の意味関係は、そのモデル（paraphrase-multilingual-mpnet-base-v2）の重みの中に既に埋め込まれています。

それをちょっと借りて使うだけで、自社データでの追加学習は一切不要な点、むちゃくちゃ便利ですよね！

Step 5a: Databricks AppsでREST APIとして公開

ここから、Databricks側にREST APIでリクエストを受け付ける受け口を作成します。

※残念ながら、以下の実装は無償の個人利用環境では、Appsが利用できないため、実装はできません。

以下コードはApps利用できる方の環境にてご利用ください。

# app.py（Databricks Apps用 Flask API）
from flask import Flask, request, jsonify
import numpy as np
import json
from sentence_transformers import SentenceTransformer
from databricks.sdk.runtime import spark  # Databricks環境のSpark

app = Flask(__name__)
model = SentenceTransformer("paraphrase-multilingual-mpnet-base-v2")

def cosine_similarity(a, b):
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "ok"})

@app.route("/similar-incidents", methods=["POST"])
def similar_incidents():
    """
    Request Body:
    {
        "query": "メールが送受信できない",
        "top_k": 3,
        "threshold": 0.0
    }
    
    Response:
    {
        "query": "...",
        "results": [
            {"number": "INC001", "short_desc": "...", "resolution": "...", "score": 0.883}
        ],
        "matched": true,
        "best_score": 0.883
    }
    """
    body      = request.get_json()
    query     = body.get("query", "")
    top_k     = body.get("top_k", 3)
    threshold = body.get("threshold", 0.0)

    if not query:
        return jsonify({"error": "query is required"}), 400

    # ベクトル化
    query_emb = model.encode(query, convert_to_tensor=False).tolist()
    
    # Delta Tableから検索
    df_pd = spark.table("incidents_with_embeddings").toPandas()
    
    results = []
    for _, row in df_pd.iterrows():
        stored_emb = json.loads(row["embedding"])
        score = cosine_similarity(query_emb, stored_emb)
        if score >= threshold:
            results.append({
                "number":     row["number"],
                "short_desc": row["short_desc"],
                "resolution": row["resolution"],
                "score":      round(score, 4)
            })
    
    results.sort(key=lambda x: x["score"], reverse=True)
    top_results = results[:top_k]
    best_score  = top_results[0]["score"] if top_results else 0.0

    return jsonify({
        "query":      query,
        "results":    top_results,
        "matched":    best_score >= 0.8,   # 閾値0.8で一致判定
        "best_score": best_score
    })

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

Databricks Appsにデプロイすると、固定URLが払い出される。例：

https://your-workspace.azuredatabricksapps.net/similar-incidents

このエンドポイントをServiceNowのIntegration Hubから呼び出します。

Step 5b: 【Free Edition対応】Jobs APIでNotebookをREST API代わりに使う

Databricks AppsはPremium版が必要ですが、Jobs APIはFree Editionでも完全に使えます。　この仕組みを使って外部接続できるようにします。

仕組みは以下のとおりです。

ServiceNowからHTTPでJobをトリガー → Serverlessクラスターが起動してNotebookが実行される → dbutils.notebook.exit() でJSON結果を返す → ServiceNowがポーリングで結果を取得。

ServiceNow Integration Hub
    │
    │ ① POST /api/2.1/jobs/runs/submit
    │   (query_text をパラメータとして渡す)
    ↓
Databricks Jobs API
    │
    │ ② run_id を即時返却
    ↓
Serverless Notebook 起動・実行
    │  ・Delta Tableから過去インシデント取得
    │  ・sentence-transformers でcosine similarity計算
    │  ・dbutils.notebook.exit(JSON文字列) で結果を返す
    ↓
    │ ③ GET /api/2.1/jobs/runs/get?run_id=xxx
    │   (TERMINATED になるまでポーリング)
    ↓
    │ ④ GET /api/2.1/jobs/runs/get-output?run_id=xxx
    │   notebook_output.result に Step 5 と完全同一のJSONが入っている
    ↓
ServiceNow Integration Hub が結果受信

注意点：レイテンシ
Serverlessのコールドスタートで初回30〜60秒かかる場合があります。

インシデント発生から自動クローズまでのリアルタイム性よりも「繋がること」を優先する検証フェーズではこれで十分です。

本番運用では Step 5（Databricks Apps）に切り替えることでミリ秒レベルに改善できます。

Jobs APIを利用する場合には、Step4に以下コードをセルより追加します。

import json
import numpy as np
from sentence_transformers import SentenceTransformer

# 依存関係にインストール済みのライブラリからロード
model = SentenceTransformer("paraphrase-multilingual-mpnet-base-v2")

# ---- Jobs APIからのパラメータ受け取り ----
try:
    query = dbutils.widgets.get("query")
except Exception:
    query = "メールが送受信できない"

try:
    top_k = int(dbutils.widgets.get("top_k"))
except Exception:
    top_k = 3

try:
    threshold = float(dbutils.widgets.get("threshold"))
except Exception:
    threshold = 0.0

# ---- cosine similarity計算 ----
def cosine_similarity(a, b):
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

def find_similar_incidents(query_text, top_k=3, threshold=0.0):
    query_emb = model.encode(query_text, convert_to_tensor=False).tolist()
    df_pd     = spark.table("incidents_with_embeddings").toPandas()
    results   = []
    for _, row in df_pd.iterrows():
        stored_emb = json.loads(row["embedding"])
        score      = cosine_similarity(query_emb, stored_emb)
        if score >= threshold:
            results.append({
                "number":     row["number"],
                "short_desc": row["short_desc"],
                "resolution": row["resolution"],
                "score":      round(score, 4)
            })
    results.sort(key=lambda x: x["score"], reverse=True)
    return results[:top_k]

# ---- 実行 ----
top_results = find_similar_incidents(query, top_k=top_k, threshold=threshold)
best_score  = top_results[0]["score"] if top_results else 0.0

output = {
    "query":      query,
    "results":    top_results,
    "matched":    best_score >= 0.8,
    "best_score": best_score
}

dbutils.notebook.exit(json.dumps(output, ensure_ascii=False))

これで、Jobs API経由でDatabricks側のNoteで実装したステップを呼ぶことができるようになります。

2-3. ServiceNow側の接続設定

Step 1: カスタムREST Action（Integration Hub）の作成

・アプリケーションスコープ：Dbricks Integration（任意文字列）

Workflow Studioより、カスタムアクションを作成します。
・アクション名：Databricks - Find Similar Incidents
・スコープ：上記で作成したアプリケーションスコープ

続いて、カスタムアクションへ入力変数の定義を行います。
・query_text String：String型、入力必須
・threshold　　　：Floating Point Number
・top_k ：Integer

Script Step の設定：

Steps タブで Script ステップを追加します。
Databricks側は以下のような要求を期待します。
①Notebookジョブの実行でシーケンスrun_idを取得
②実行ジョブが "TERMINATED" になるまでポーリング
③最後に、結果セットを受領

今回の連携処理は、この②が曲者です。これが無ければノーコードでステップ作れたのですが、、このポーリングだけは状態が完了するまでループで処理を回さないといけません。

よって、スクリプトステップに3つの処理全てを搭載する形にしました。

※これは別にDatabricks側が悪いというわけではないんですね・・
PythonをサポートしないServiceNow上では、databricksの接続ライブラリが利用できないからこのような構成を組まざるを得ないというオチです。

(function execute(inputs, outputs) {

    // 接続情報（自分の環境に合わせて書き換える）
    var baseUrl      = "https://dbc-xxxxxxxx.cloud.databricks.com";
    var pat          = "dapixxxxxxxxxx";
    var notebookPath = "/Users/your-email@example.com/similarity_search_api";

    // Script Stepの入力変数から参照する（inputs.変数名で参照）
    var queryText  = inputs.query_text;
    var topK       = String(inputs.top_k     || 3);
    var thresh     = String(inputs.threshold || 0.0);
    var authHeader = "Bearer " + pat;

    // ----------------------------------------------------------------
    // ① runs/submit: Notebookジョブをトリガーしてrun_idを取得
    // ----------------------------------------------------------------
    var submitReq = new sn_ws.RESTMessageV2();
    submitReq.setEndpoint(baseUrl + "/api/2.1/jobs/runs/submit");
    submitReq.setHttpMethod("POST");
    submitReq.setRequestHeader("Authorization", authHeader);
    submitReq.setRequestHeader("Content-Type", "application/json");
    submitReq.setRequestBody(JSON.stringify({
        "run_name": "similar_incident_search",
        "tasks": [{
            "task_key": "search",
            "notebook_task": {
                "notebook_path": notebookPath,
                "base_parameters": {
                    "query":     queryText,
                    "top_k":     topK,
                    "threshold": thresh
                }
            }
        }],
        "queue": { "enabled": true }
    }));

    var submitRes = submitReq.execute();
    if (submitRes.getStatusCode() != 200) {
        gs.error("Databricks submit failed: " + submitRes.getBody());
        return;
    }
    var runId = JSON.parse(submitRes.getBody()).run_id;

    // ----------------------------------------------------------------
    // ② runs/get: TERMINATEDになるまでポーリング（15秒間隔・最大20回 = 5分）
    // ----------------------------------------------------------------
    var lifecycleState = "";
    var maxRetry       = 20;
    var count          = 0;

   while (lifecycleState != "TERMINATED" && count < maxRetry) {
        // 15秒待機（ビジーウェイト）
        var waitUntil = new GlideDateTime();
        waitUntil.addSeconds(15);
        while (new GlideDateTime().before(waitUntil)) { /* wait */ }


        var pollReq = new sn_ws.RESTMessageV2();
        pollReq.setEndpoint(baseUrl + "/api/2.1/jobs/runs/get?run_id=" + runId);
        pollReq.setHttpMethod("GET");
        pollReq.setRequestHeader("Authorization", authHeader);

        var pollBody   = JSON.parse(pollReq.execute().getBody());
        lifecycleState = pollBody.state.life_cycle_state;
        count++;
    }

    if (lifecycleState != "TERMINATED") {
        gs.error("Databricks job did not complete within timeout. run_id: " + runId);
        return;
    }

    // ----------------------------------------------------------------
    // ③ runs/get-output: Notebookの実行結果を取得
    // notebook_output.result = dbutils.notebook.exit() で返したJSON文字列
    // ----------------------------------------------------------------
    var outputReq = new sn_ws.RESTMessageV2();
    outputReq.setEndpoint(baseUrl + "/api/2.1/jobs/runs/get-output?run_id=" + runId);
    outputReq.setHttpMethod("GET");
    outputReq.setRequestHeader("Authorization", authHeader);

    var outputBody = JSON.parse(outputReq.execute().getBody());
    var result     = JSON.parse(outputBody.notebook_output.result);

    outputs.matched    = result.matched;
    outputs.best_score = result.best_score;
    outputs.resolution = result.results && result.results.length > 0
                         ? result.results[0].resolution : "";
    outputs.top_number = result.results && result.results.length > 0
                         ? result.results[0].number : "";

})(inputs, outputs);

１）以下の定数値は適宜にsys_propertiesなどから取り出せるように変更対応ください。※ここでは検証を目的にハードコードさせてます
var baseUrl = "https://dbc-xxxxxxxx.cloud.databricks.com";
var pat = "dapixxxxxxxxxx";
var notebookPath = "/Users/your-email@example.com/similarity_search_api";
※DatabricksのPATトークンキーの取得はこちらを参考ください。

２）GlideDateTime()：アプリスコープを切ると、gs.sleep()使えません。ポーリング処理でループさせます。

続いてスクリプトから得られる出力値向けの変数を定義します。
・matched : True/False
・best_score：Floating Point Number
・resolution：String
・top_number：String

この状態でカスタムアクションをテスト走行させてみます。
・query_text：メールが送受信できない
・threshold：0.0
・top_k：3

ServiceNowからDatabricks側のJobsAPI経由でノートブック（Pythonスクリプト）を実行開始させました。

続いて、Databricks側の実行ジョブを確認します。
[ジョブの実行]より、最新ジョブを選択。

２分近く待つと、以下のようなジョブ実行応答が完了します。

Databricks側では、処理が終わっていますが、、当初の考慮点の課題がやはり残りました。

・無償個人環境では、常にModelライブラリインストールが求められ、1分以上のインストール処理が常時入る
・ServiceNowのトランザクションタイムアウトに掛かってしまい、ServiceNowは結果受領まで処理を待てずに、コネクションを切ってしまう。

この無償環境では、この制約で接続検証を続けるのは厳しいですね・・
本記事内容が膨らんできたので、一旦本記事の検証作業はここまでとします。

まとめ

今回の検証作業でできたことと、残った課題を纏めます。

✅ 動作確認済み

❌ 今回越えられなかった壁

Serverless Cold Start（毎回コンテナリセット）
  → sentence-transformers インストール: 約44秒
    vs
ServiceNow スコープアプリ Script Step タイムアウト: 約22秒

→ 構造的に接続不可

追記：Unity Catalog Volumeへのモデルキャッシュも試みましたが、Serverlessはコンテナごとリセットされるため効果なし。

依存関係（Dependencies）にsentence-transformersを追加しても同様。Free EditionではServerlessの常時起動が使えないため、根本解決にならなかった。
※これらは、個人無償環境の制約であり、Premium版では発生しない。

しかし、本来の目的である

・ServiceNow側からDatabricks側へインシデントの”類似性の検出”という分析要求を投げ、回答を作らせる

というところまでは辿り着けました。

次回は、この受領回答をもらった形より、AI Agentの動きをどのように作るかを公開する予定とします。お楽しみに！