ServiceNowのワークフローレコード、Table API × Streamlit × 生成AI で簡単に予測分析できた話

2026年4月15日 21:44

ServiceNowの腹持ちレコードデータ、とりわけワークフローの履歴レコードデータを元に将来予測分析を実施する要件、実はなかなか対応が厳しかったりします。

その技術チャレンジは以下の特性からです。

・現行Platformでは、ACL制御前提のGlideRecordを基本とするデータアクセスのみ。SQLが使えないので集計クエリが苦手。
※物理的に不可ではないが、実装するには、まぁまぁ面倒

・分析用 Performance Analytics 機能は、”これから蓄積していくKPI監視” 用途であって、”既存の履歴データを遡って分析する” BIツールではない。

もちろん、ServiceNowでもこの辺りの課題認識はしており、2026年2月にはPyramid Analyticsといった分析SaaSプラットフォーム企業を買収し、機能を搭載してゆくなどの様相を対外的には示しています。

https://www.servicenow.com/workflow/news/pyramid-analytics-expanding-data-driven-ai.html?state=seamless

とはいえ、現時点（2026年4月）では、この ”将来予測分析” 要件を実装するのであるならば、やはりServiceNowの外側に分析機能を持たせることが、コスト面含めてリーズナブルなアプローチであると思います。

そこで、今回はこの要件を
・ServiceNow Table API × Streamlit × 生成AI (Cohere)
という組み合わせで実現できるアプリケーションを開発して実現します。

Streamlitを使ったアプリは以前もこのブログで扱っています。その際の記事も参考ください。

１．環境＆ワークフロー履歴データ準備

以下のような環境を準備しました。
・ServiceNow：Australia [PDI]
・Python：3.13
・Cohere API

ワークフローの履歴データを1,000件作りこます。以下Pythonプログラムを実行するとCSVファイルが生成されます。

import csv
import random
from datetime import date, timedelta

# --- 設定 ---
OUTPUT_FILE = "fin_wf_data.csv"
TOTAL_RECORDS = 1000
START_DATE = date(2025, 1, 1)
END_DATE = date(2026, 4, 14)

# Choice値
CATEGORIES = ["capital_investment", "human_cost", "advertise_cost"]
STATUSES = ["requested", "approved", "rejected"]

# 部門リスト
DIVISIONS = ["営業部", "マーケティング部", "人事部", "経理部", "開発部", "総務部", "購買部"]

# 承認者リスト
APPROVERS = ["田中部長", "鈴木部長", "佐藤部長", "山田部長", "高橋部長"]

# 申請者リスト
REQUESTERS = [
    "山本太郎", "佐藤花子", "田中一郎", "鈴木明子", "高橋健二",
    "渡辺幸子", "伊藤誠", "中村美咲", "小林雄介", "加藤由美"
]

# カテゴリ別金額レンジ（リアリティのため）
COST_RANGE = {
    "capital_investment": (500000, 5000000),
    "human_cost":         (200000, 1500000),
    "advertise_cost":     (100000, 3000000),
}

# 承認率・却下率（申請中は直近データに多め）
def get_status_weights(req_date: date) -> list:
    """直近3ヶ月は申請中が多い、それ以前は承認済みが多い"""
    cutoff = date(2026, 1, 1)
    if req_date >= cutoff:
        return [0.5, 0.3, 0.2]   # requested, approved, rejected
    else:
        return [0.05, 0.75, 0.2]


def random_date(start: date, end: date) -> date:
    delta = (end - start).days
    return start + timedelta(days=random.randint(0, delta))


def get_fiscal_info(d: date) -> tuple:
    """会計年度・会計月・予算期を返す（4月始まり想定）"""
    if d.month >= 4:
        fiscal_year = d.year
        fiscal_month = d.month - 3
    else:
        fiscal_year = d.year - 1
        fiscal_month = d.month + 9

    if fiscal_month <= 3:
        budget_period = "Q1"
    elif fiscal_month <= 6:
        budget_period = "Q2"
    elif fiscal_month <= 9:
        budget_period = "Q3"
    else:
        budget_period = "Q4"

    return fiscal_year, fiscal_month, budget_period


def generate_approval_date(req_date: date, status: str) -> str:
    """承認済み・却下の場合のみ承認日を生成"""
    if status == "requested":
        return ""
    delta = random.randint(3, 30)
    approval_date = req_date + timedelta(days=delta)
    if approval_date > END_DATE:
        approval_date = END_DATE
    return approval_date.strftime("%Y-%m-%d")


# --- メイン生成 ---
records = []

for i in range(1, TOTAL_RECORDS + 1):
    req_date = random_date(START_DATE, END_DATE)
    category = random.choice(CATEGORIES)
    status = random.choices(STATUSES, weights=get_status_weights(req_date))[0]
    fiscal_year, fiscal_month, budget_period = get_fiscal_info(req_date)
    cost_min, cost_max = COST_RANGE[category]
    cost = random.randrange(cost_min, cost_max, 10000)
    approval_date = generate_approval_date(req_date, status)
    approver = random.choice(APPROVERS) if status != "requested" else ""

    records.append({
        "u_requested_date": req_date.strftime("%Y-%m-%d"),
        "u_requested_div":  random.choice(DIVISIONS),
        "u_request_cost":   cost,
        "u_request_category": category,
        "u_request_stat":   status,
        "u_fiscal_year":    fiscal_year,
        "u_fiscal_month":   fiscal_month,
        "u_budget_period":  budget_period,
        "u_approval_date":  approval_date,
        "u_approver":       approver,
    })

# CSV出力
fieldnames = [
    "u_requested_date", "u_requested_div", "u_request_cost",
    "u_request_category", "u_request_stat", "u_fiscal_year",
    "u_fiscal_month", "u_budget_period", "u_approval_date", "u_approver"
]

with open(OUTPUT_FILE, "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(records)

print(f"{TOTAL_RECORDS}件のダミーデータを {OUTPUT_FILE} に出力しました。")

これで、環境面とデータ面の下準備は整いました。

２．財務ワークフローアプリケーション

ここでは、分析対象のカスタムアプリケーションとして、
・財務ワークフローアプリケーション
を作成しておきます。

レコードデータを抽出するアプリ用途となるので、レコードデータが格納できる程度の簡単なアプリ構成で仕立てます。

①データモデル（テーブル定義）
アプリケーションスコープ名とカスタムテーブル構成。

フィールドの定義。

②データのバルクインポート
Data Integration機能を使って、一括で入れます。

トランスフォームマッピングの定義。取り込み元のカラムと取り込み先テーブルのフィールドをマッピングさせます。

データインポート完了

これで分析用のアプリケーションおよびレコードデータは準備できました。

３．カスタムアプリ分析ツールの作成

このアプリのシーケンスを示します。

Pythonのpandasで、集計・統計→plotlyでグラフ化→Cohere LLMで分析コメントの応答　という流れで生成してます。

このブログではおなじみになってきました、”一回に纏めてLLMへ分析を依頼” というアプローチにて、AI利用コストを抑えた形で実装をしています。

この ”纏める件" のもう１つの背景は、大量レコードの分析をバルクでLLMに投げてしまうと、どうしてもコンテキストウィンドウの上限値に、引っかかってしまうことを避ける目的も含まれています。

では、最初にカスタムアプリケーションのTable APIへ疎通できるかどうかを確認してみます。

疎通させるシステム管理者へカスタムアプリケーションのロールを付与しておきます。

PowerShellコマンドでは以下のような定義を行います。

curl.exe -u "admin:password" "https://instance.service-now.com/api/now/table/your-table?sysparm_limit=1" -H "Accept: application/json"

１）password：adminユーザのパスワード文字列を指定
２）instance：PDIのインスタンス名を指定
３）your-table：カスタムアプリの物理テーブル名
→例）x_98XXXXX_fin_wf_wpp

以下のようなステータスが応答されれば、Table APIへの疎通成功です！
カスタムテーブルのレコードが応答されています。

{"result":[{"u_fiscal_month":"3","u_request_cost":"2880000","u_request_stat":"requested","sys_mod_count":"0","sys_updated_on":"2026-04-15 07:40:35","sys_tags":"","u_fiscal_year":"2025","number":"FINWFAP0001002","sys_id":"018d850dc3d083506afd9f3ed4013109","u_requester":"","sys_updated_by":"admin","u_approver":"","sys_created_on":"2026-04-15 07:40:35","u_requested_date":"2025-06-08","u_budget_period":"Q1","u_request_category":"capital_investment","sys_created_by":"admin","u_approval_date":"","u_requested_div":"人事部"}]}

続いて、Python実行環境の確認とインストールを行います。

python --version

当方環境では"3.13.12"がインストールされていました。インストールされてない場合にはインストールの実行を行ってください。

続いて、データ操作用のPandasなどのPython用ライブラリ一式をインストールします。

pip install streamlit requests pandas plotly cohere scikit-learn numpy　python-dotenv

ライブラリのインストールが成功したら、環境準備は完了です。

では、今回のメイン処理にあたるPythonスクリプトを作成します。
以下のようなパス構成で２つのファイルを作成します。
※envファイルより、接続するパラメータ値を読み取る構成にしてます。

C:\work\sn_wf_analyzer\
├── fin_wf_app.py
└── .env

・.envに以下を上記ファイルにコピー＆ペーストして下さい。
・実際の値を設定してください。

SN_INSTANCE=https://XXXXXX.service-now.com
SN_USER=admin
SN_PASS=XXXXXXX
COHERE_API_KEY=XXXXXXX

・fin_wf_app.pyファイルに以下ステップをコピー＆ペーストしてください。

import os
import streamlit as st
import requests
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import numpy as np
import cohere
from sklearn.linear_model import LinearRegression
from requests.auth import HTTPBasicAuth
from dotenv import load_dotenv

load_dotenv()

# ============================================================
# 設定
# ============================================================
st.set_page_config(page_title="財務WF 分析ダッシュボード", layout="wide")

SN_INSTANCE = os.getenv("SN_INSTANCE")
SN_USER     = os.getenv("SN_USER")
SN_PASS     = os.getenv("SN_PASS")
COHERE_KEY  = os.getenv("COHERE_API_KEY")
TABLE_NAME  = "your-table"　                #カスタムテーブルを指定　

CATEGORY_LABELS = {
    "capital_investment": "設備投資",
    "human_cost":         "人件費",
    "advertise_cost":     "広告費",
}
STATUS_LABELS = {
    "requested": "申請中",
    "approved":  "承認済み",
    "rejected":  "却下",
}

# ============================================================
# Table API からレコード取得
# ============================================================
@st.cache_data(ttl=300)
def fetch_records() -> pd.DataFrame:
    url = f"{SN_INSTANCE}/api/now/table/{TABLE_NAME}"
    params = {
        "sysparm_limit": 2000,
        "sysparm_fields": (
            "u_requested_date,u_requested_div,u_request_cost,"
            "u_request_category,u_request_stat,"
            "u_fiscal_year,u_fiscal_month,u_budget_period,"
            "u_approval_date,u_approver"
        ),
    }
    resp = requests.get(
        url,
        params=params,
        auth=HTTPBasicAuth(SN_USER, SN_PASS),
        headers={"Accept": "application/json"},
        timeout=30,
    )
    resp.raise_for_status()
    records = resp.json().get("result", [])
    df = pd.DataFrame(records)

    # 型変換
    df["u_requested_date"] = pd.to_datetime(df["u_requested_date"], errors="coerce")
    df["u_request_cost"]   = pd.to_numeric(df["u_request_cost"], errors="coerce").fillna(0)
    df["u_fiscal_year"]    = pd.to_numeric(df["u_fiscal_year"], errors="coerce")
    df["u_fiscal_month"]   = pd.to_numeric(df["u_fiscal_month"], errors="coerce")

    # ラベル変換
    df["カテゴリ"] = df["u_request_category"].map(CATEGORY_LABELS).fillna(df["u_request_category"])
    df["状況"]     = df["u_request_stat"].map(STATUS_LABELS).fillna(df["u_request_stat"])

    return df


# ============================================================
# Cohere 分析コメント生成
# ============================================================
def get_cohere_analysis(summary_text: str, user_question: str) -> str:
    co = cohere.ClientV2(api_key=COHERE_KEY)
    prompt = f"""
あなたは財務分析の専門家です。以下の集計データをもとに、ユーザーの質問に日本語で簡潔に答えてください。

【集計データ】
{summary_text}

【ユーザーの質問】
{user_question}
"""
    response = co.chat(
        model="command-r-plus-08-2024",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.message.content[0].text


# ============================================================
# メイン
# ============================================================
st.title("💹 財務WF 分析ダッシュボード")
st.caption("ServiceNow Fin WF App × Cohere による予算申請分析")

# データ取得
with st.spinner("ServiceNowからデータを取得中..."):
    try:
        df = fetch_records()
    except Exception as e:
        st.error(f"データ取得エラー: {e}")
        st.stop()

st.success(f"✅ {len(df):,} 件のレコードを取得しました")

# ============================================================
# サイドバー：フィルター
# ============================================================
st.sidebar.header("🔍 絞り込み条件")

# 部門
all_divs = sorted(df["u_requested_div"].dropna().unique().tolist())
selected_div = st.sidebar.selectbox("申請部門", ["全部門"] + all_divs)

# 会計年度
all_years = sorted(df["u_fiscal_year"].dropna().unique().astype(int).tolist())
selected_year = st.sidebar.selectbox("会計年度", ["全年度"] + all_years)

# 申請カテゴリ
all_cats = ["全カテゴリ"] + list(CATEGORY_LABELS.values())
selected_cat = st.sidebar.selectbox("申請カテゴリ", all_cats)

# 申請状況
all_stats = ["全状況"] + list(STATUS_LABELS.values())
selected_stat = st.sidebar.selectbox("申請状況", all_stats)

# ============================================================
# フィルター適用
# ============================================================
filtered = df.copy()
if selected_div != "全部門":
    filtered = filtered[filtered["u_requested_div"] == selected_div]
if selected_year != "全年度":
    filtered = filtered[filtered["u_fiscal_year"] == selected_year]
if selected_cat != "全カテゴリ":
    filtered = filtered[filtered["カテゴリ"] == selected_cat]
if selected_stat != "全状況":
    filtered = filtered[filtered["状況"] == selected_stat]

if filtered.empty:
    st.warning("条件に一致するレコードがありません。")
    st.stop()

# ============================================================
# KPIカード
# ============================================================
col1, col2, col3, col4 = st.columns(4)
col1.metric("総申請件数",   f"{len(filtered):,} 件")
col2.metric("総申請金額",   f"¥{filtered['u_request_cost'].sum():,.0f}")
col3.metric("承認済み件数", f"{(filtered['状況']=='承認済み').sum():,} 件")
col4.metric("承認率",
            f"{(filtered['状況']=='承認済み').sum() / len(filtered) * 100:.1f}%")

st.divider()

# ============================================================
# グラフ
# ============================================================
col_l, col_r = st.columns(2)

# 月次申請金額トレンド＋予測
with col_l:
    st.subheader("📈 月次申請金額トレンド＋3ヶ月予測")
    monthly = (
        filtered.groupby(["u_fiscal_year", "u_fiscal_month"])["u_request_cost"]
        .sum()
        .reset_index()
    )
    monthly["年月"] = (
        monthly["u_fiscal_year"].astype(int).astype(str)
        + "-"
        + monthly["u_fiscal_month"].astype(int).astype(str).str.zfill(2)
    )
    monthly = monthly.sort_values(["u_fiscal_year", "u_fiscal_month"]).reset_index(drop=True)

    # 線形回帰で3ヶ月予測
    X = np.arange(len(monthly)).reshape(-1, 1)
    y = monthly["u_request_cost"].values
    model = LinearRegression().fit(X, y)

    future_X = np.arange(len(monthly), len(monthly) + 3).reshape(-1, 1)
    predictions = model.predict(future_X)

    # 予測用の年月ラベルを生成
    last_year  = int(monthly["u_fiscal_year"].iloc[-1])
    last_month = int(monthly["u_fiscal_month"].iloc[-1])
    future_labels = []
    for i in range(1, 4):
        m = last_month + i
        y_label = last_year + (m - 1) // 12
        m_label = (m - 1) % 12 + 1
        future_labels.append(f"{y_label}-{str(m_label).zfill(2)}")

    # グラフ描画
    fig1 = go.Figure()
    fig1.add_trace(go.Scatter(
        x=monthly["年月"], y=monthly["u_request_cost"],
        mode="lines+markers", name="実績", line=dict(color="#1f77b4")
    ))
    fig1.add_trace(go.Scatter(
        x=future_labels, y=predictions,
        mode="lines+markers", name="予測（線形回帰）",
        line=dict(color="#ff7f0e", dash="dash")
    ))
    fig1.update_layout(xaxis_tickangle=45, legend=dict(orientation="h"))
    st.plotly_chart(fig1, use_container_width=True)

    # 予測サマリーをセッションに保持（Cohere用）
    prediction_summary = "\n".join([
        f"  {future_labels[i]}: {predictions[i]:,.0f}円（予測）"
        for i in range(3)
    ])

# カテゴリ別申請金額
with col_r:
    st.subheader("🥧 カテゴリ別申請金額")
    cat_sum = filtered.groupby("カテゴリ")["u_request_cost"].sum().reset_index()
    fig2 = px.pie(cat_sum, names="カテゴリ", values="u_request_cost",
                  hole=0.4)
    st.plotly_chart(fig2, use_container_width=True)

# 部門別申請金額
st.subheader("🏢 部門別申請金額")
div_sum = (
    filtered.groupby("u_requested_div")["u_request_cost"]
    .sum()
    .reset_index()
    .sort_values("u_request_cost", ascending=False)
)
fig3 = px.bar(div_sum, x="u_requested_div", y="u_request_cost",
              labels={"u_requested_div": "部門", "u_request_cost": "申請金額（円）"},
              color="u_request_cost", color_continuous_scale="Blues")
st.plotly_chart(fig3, use_container_width=True)

st.divider()

# ============================================================
# Cohere 分析
# ============================================================
st.subheader("🤖 Cohere による分析コメント")

user_question = st.text_input(
    "分析したい内容を入力してください",
    placeholder="例：申請が多い部門と時期の傾向を教えてください",
)

if st.button("分析実行", type="primary") and user_question:
    # 集計サマリーを作成してCohereに渡す
    summary_lines = [
        f"総申請件数: {len(filtered)}件",
        f"総申請金額: {filtered['u_request_cost'].sum():,.0f}円",
        f"承認率: {(filtered['状況']=='承認済み').sum() / len(filtered) * 100:.1f}%",
        "",
        "【部門別申請金額Top3】",
    ]
    for _, row in div_sum.head(3).iterrows():
        summary_lines.append(f"  {row['u_requested_div']}: {row['u_request_cost']:,.0f}円")

    summary_lines += ["", "【カテゴリ別申請金額】"]
    for _, row in cat_sum.iterrows():
        summary_lines.append(f"  {row['カテゴリ']}: {row['u_request_cost']:,.0f}円")

    summary_lines += ["", "【月次トレンド（直近6ヶ月）】"]
    for _, row in monthly.tail(6).iterrows():
        summary_lines.append(f"  {row['年月']}: {row['u_request_cost']:,.0f}円")

    summary_lines += ["", "【線形回帰による今後3ヶ月の予測】"]
    summary_lines.append(prediction_summary)

    summary_text = "\n".join(summary_lines)

    with st.spinner("Cohereが分析中..."):
        try:
            comment = get_cohere_analysis(summary_text, user_question)
            st.info(comment)
        except Exception as e:
            st.error(f"Cohere APIエラー: {e}")

# ============================================================
# 生データ表示
# ============================================================
with st.expander("📋 生データを見る"):
    st.dataframe(
        filtered[[
            "u_requested_date", "u_requested_div", "u_request_cost",
            "カテゴリ", "状況", "u_fiscal_year", "u_fiscal_month",
            "u_budget_period", "u_approval_date", "u_approver",
        ]].rename(columns={
            "u_requested_date": "申請日",
            "u_requested_div":  "部門",
            "u_request_cost":   "申請金額",
            "u_fiscal_year":    "会計年度",
            "u_fiscal_month":   "会計月",
            "u_budget_period":  "予算期",
            "u_approval_date":  "承認日",
            "u_approver":       "承認者",
        }),
        use_container_width=True,
    )

１）型変換：Cohere側へ渡す際に適応フォーマットへ変換

２）Cohereの利用モデル：command-r-plus-08-2024
※command-r-plusを指定したところ、2025/9で廃止されており、↑のモデルへ変更しました。実装時に該当モデルが利用可能かどうかを確認ください。

３）集計→グラフ化→分析コメント追加までの流れ：

４．動作確認

では、動作確認をしてみましょう。

python -m streamlit run fin_wf_app.py

以下のようなステータスが応答されます。

ブラウザ上でタブが自動で開かれます。画面がロードされるまで待ちます。

すでに、1000件のレコードを取得しています！
集計からグラフ化までの生成が行われ、画面に表示されています。

左サイドバーにあるレコードのフィルターを使って、条件による切り替え表示が行えます。

・マーケティング部門 - 2025年度 - 広告費　で出力

さて、今回のメイン機能である、”傾向分析” を実施したいと思います。
LLMへ傾向分析コメントを求めてみましょう。

集計結果をLLMに送信し、分析結果が受領できました！
良い感じです！

続いて、"予測分析 " コメントも求めてみましょう。

集計から、トレンド予測までを返してくれてます！

５．まとめ

如何でしたでしょうか？　

ワークフローレコードデータからの傾向や予測分析は、こうしたアプローチで簡易に実現することも可能です。

せっかく蓄積されたワークフローデータは宝の山です！

「分析する機能を組むのが大変だからなあ」と諦めないで、ぜひ活用の場をご検討頂ければと。

今後、ServiceNowの中に組み込まれる分析機能が出てきた際に、今回のような ”集計から予測” といった分析要件への適応性を見極めて頂くと良いと思います。

もう一点、この記事の冒頭で ”SQLが書けない” とコメントしました。

しかし、AustraliaリリースにRaptorDB Proという有償版DBへ "SQL API"というコネクターが搭載されてきます。

これ、、実際にSQLで腹持ちデータを高速読み取りできる機能を提供してくれるようになるのですが、、今回実装した要件は、このSQL APIを使うと、Python側で処理させていた機能やより高度な集計もできるのではないかと期待したりしています。　

まだ実際に触る環境がないので、、いつか触れることができましたら、当ブログへ書いてみようと思います。

次回記事にも、乞うご期待ください！