Databricksで生成AIガバナンスを実装する：AIガードレール×プロンプト管理ハンズオン

Posted at 2026-06-10

はじめに

生成AIエージェントの企業導入が進む中、ガバナンスの欠如が深刻な課題になっています。

各チームが独自にLLMを利用し、モデルの乱立が起きている
誰が何を使い、コストがいくらかかっているか可視性がない
機密データが監査証跡なしでモデルに送信されるコンプライアンスリスク
品質を測定・改善する仕組みがない

この記事では、DatabricksのAIゲートウェイとMLflowを使って、生成AIアプリケーションにガバナンスを組み込む方法をハンズオン形式で紹介します。約70分で以下を構築します。

座学資料はこちらで公開しています。

ハンズオンノートブックは以下で公開しています。

前提条件

Databricksワークスペースへのアクセス権限
Unity AIゲートウェイ（Beta）がAccount Console Previewsで有効化済み
MLflow Prompt Registry（Beta）がWorkspace Previewsで有効化済み

0. 環境セットアップ

まずライブラリのインストールと設定変数を定義します。

%pip install -U openai "mlflow[databricks]>=3.1.0" databricks-sdk
dbutils.library.restartPython()

CATALOG = "workshop"
SCHEMA = "llmops"
MODEL_NAME = "databricks-meta-llama-3-3-70b-instruct"

参加者ごとのユニーク識別子とAIゲートウェイエンドポイント名を自動生成します。

USERNAME = (
    spark.sql("SELECT current_user()").first()[0]
    .split("@")[0]
    .replace(".", "_")
    .replace("-", "_")
)
ENDPOINT_NAME = f"ws_{USERNAME}"

OpenAIクライアントとMLflowの初期化

AIゲートウェイのエンドポイントは、従来の/serving-endpointsではなく/ai-gateway/mlflow/v1をベースURLとして使用します。OpenAI互換のSDKでそのままアクセスできます。

from openai import OpenAI
import mlflow

_host = spark.conf.get("spark.databricks.workspaceUrl")
_token = (
    dbutils.notebook.entry_point.getDbutils()
    .notebook().getContext().apiToken().get()
)

# AIゲートウェイ向けのOpenAI互換クライアント
client = OpenAI(
    api_key=_token,
    base_url=f"https://{_host}/ai-gateway/mlflow/v1",
)

# MLflowトレーシングの有効化
_current_user = spark.sql("SELECT current_user()").first()[0]
mlflow.set_experiment(f"/Users/{_current_user}/llmops_workshop")
mlflow.set_registry_uri("databricks-uc")
mlflow.tracing.enable()

# OpenAIクライアントの自動計装（トークン使用量が自動記録される）
mlflow.openai.autolog()

ここで設定したmlflow.openai.autolog()が重要です。これにより、OpenAIクライアント経由のLLM呼び出しごとにトークン使用量がMLflow Traceのスパンに自動記録されます。

1. AIゲートウェイとガードレール

AIゲートウェイは、生成AIのコントロールプレーンです。統一アクセス、コスト管理、集中ガバナンス、本番環境の信頼性を提供します。

1.1 AIゲートウェイエンドポイントの作成

DatabricksのUI上で以下の手順でエンドポイントを作成します。

左サイドバーの AIゲートウェイ をクリック
+ AI Gateway Endpoint をクリック
名前にノートブックが出力した自分専用のエンドポイント名を入力
プロバイダー「Databricksによるホスト」→ Meta Llama 3.3 70B Instruct を選択
作成をクリック

1.2 ビルトインガードレールの設定

ビルトインガードレールはDatabricksがチューニング済みの検出モデルで動作します。エンドポイントの ガードレール タブから以下を追加します。

入力ガードレール（LLMの前）

ガードレールタイプ	フェーズ	説明
PIIのマスキング	サニタイズ	個人情報をプレースホルダーに置換してからモデルに送信
安全でないコンテンツ	ブロック	有害コンテンツを含むリクエストをブロック

出力ガードレール（LLMの後）

ガードレールタイプ	フェーズ	説明
PIIのマスキング	出力ガードレール	モデルの応答に含まれる個人情報を置換

注意: 入力ブロックガードレールは最大3つまでです。後でカスタムガードレールを2つ追加するため、ビルトインのブロックは 安全でないコンテンツ のみにします。

1.3 ビルトインガードレールの動作検証

ヘルパー関数query_llmでテストします。

# 正常リクエスト → 通過する
query_llm("東京の有名な観光スポットを3つ、簡潔に教えてください。")

# 入力PIIマスキング → 電話番号・住所・メールがプレースホルダーに置換される
query_llm(
    "以下のお客様情報を確認して、要約してください。\n"
    "氏名: 田中太郎\n"
    "電話番号: 090-1234-5678\n"
    "住所: 東京都渋谷区神宮前1-2-3\n"
    "メール: tanaka.taro@example.com"
)

観察ポイント: 電話番号や住所はマスキングされますが、日本語の人名はマスキングされない場合があります。これはビルトインPII検出の限界です。

検証結果をまとめると:

PII種別	検出	備考
電話番号	✅	`記載なし`に置換
住所	✅	`記載なし`に置換
メールアドレス	✅	`記載なし`に置換
日本語の人名	⚠️	マスキングされない場合がある
マイナンバー	❌	ビルトインでは対応外

日本固有のPIIにはカスタムガードレールで対応します。

1.4 カスタムガードレールの作成

カスタムガードレールはLLMがエバリュエーター（評価者）としてポリシー判定を行います。エバリュエーターが{"flagged": true}を返すとリクエストがブロックされます。

ガードレール1: マイナンバー検出

あなたはマイナンバー（個人番号）の検出を行うセキュリティ評価者です。
必ずJSON形式のみで回答してください。それ以外のテキストは出力しないでください。

以下のいずれかに該当する場合: {"flagged": true}
該当しない場合: {"flagged": false}

判定基準:
- 「マイナンバー」または「個人番号」という言葉と数字が一緒に含まれている
- 12桁の数字がハイフン区切り・スペース区切り・連続のいずれかの形式で含まれている

【違反の例】
- 「マイナンバーは1234-5678-9012です」→ {"flagged": true}
- 「個人番号: 123456789012 を登録してください」→ {"flagged": true}

【違反ではない例】
- 「マイナンバーカードの申請方法を教えて」→ {"flagged": false}
- 「電話番号は090-1234-5678です」→ {"flagged": false}

ガードレール2: 競合他社ブロック（プロンプトはノートブック参照）

プロンプト設計のポイント

JSON出力指示を明示 — {"flagged": true} / {"flagged": false} の形式をプロンプト内に記載
few-shot例を両方向で記載 — 違反/非違反の具体例をJSON形式で
1ガードレール = 1関心事 — マイナンバー検出と競合他社ブロックは別々に作成

1.5 カスタムガードレールの適用範囲

カスタムガードレールはLLMが評価を行うため、意味的な判定（競合他社の言及、ポリシー違反の検出など）に適しています。一方、マイナンバーのようなパターンベースの検出はLLMの確率的な性質により精度が安定しないため、本番では多層防御アーキテクチャを推奨します。

2. 簡易エージェントの構築

ガードレール付きエンドポイントをLLMバックエンドとして、function callingを使った簡易カスタマーサポートエージェントを構築します。

2.1 ツールとシステムプロンプトの定義

# ダミーの商品検索ツール
@mlflow.trace(name="search_products")
def search_products(keyword: str) -> str:
    """キーワードに一致する商品・サービスを検索して返す"""
    results = []
    for name, info in PRODUCT_DATA.items():
        if keyword in name or keyword in info["特徴"] or keyword in info["内容"]:
            results.append(f"【{name}】月額{info['月額']} / {info['内容']} / {info['特徴']}")
    if not results:
        for name, info in PRODUCT_DATA.items():
            results.append(f"【{name}】月額{info['月額']} / {info['内容']} / {info['特徴']}")
    return "\n".join(results)

# OpenAI function calling 用のツール定義
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_products",
            "description": "商品・サービスをキーワードで検索する。",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {
                        "type": "string",
                        "description": "検索キーワード",
                    }
                },
                "required": ["keyword"],
            },
        },
    }
]

2.2 エージェントの実装

エージェントの処理フローは以下の通りです。

システムプロンプト + ユーザー質問をLLMに送信
LLMがツール呼び出しを返した場合、ツールを実行
ツール結果をLLMに返して最終回答を生成

@mlflow.trace(name="customer_support_agent")
def run_agent(user_question: str):
    """ガードレール付きエンドポイントでツール呼び出しを行う簡易エージェント"""
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_question},
    ]

    # ステップ1: LLM呼び出し（ガードレールがこの時点で入力を評価）
    with mlflow.start_span(name="llm_call_1_tool_selection") as span:
        span.set_inputs({"user_question": user_question})
        response = client.chat.completions.create(
            model=ENDPOINT_NAME, messages=messages, tools=tools, max_tokens=500,
        )
        assistant_msg = response.choices[0].message
        span.set_outputs({"tool_calls": bool(assistant_msg.tool_calls)})

    # ステップ2: ツールを実行
    if assistant_msg.tool_calls:
        messages.append(assistant_msg)
        for tool_call in assistant_msg.tool_calls:
            args = json.loads(tool_call.function.arguments)
            result = search_products(**args)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

        # ステップ3: 最終回答を生成
        with mlflow.start_span(name="llm_call_2_final_answer") as span:
            response = client.chat.completions.create(
                model=ENDPOINT_NAME, messages=messages, max_tokens=500,
            )
            assistant_msg = response.choices[0].message

    return assistant_msg.content

ポイントは、エージェント自体はガードレールのことを一切知らない点です。ガードレールはAIゲートウェイ層で透過的に適用されるため、エージェントのコードにセキュリティロジックを埋め込む必要がありません。

# 正常な問い合わせ → ツールを呼んで回答
run_agent("おすすめのプランを教えてください")

# マイナンバーを含む問い合わせ → ガードレールでブロック
run_agent("マイナンバー1234-5678-9012で本人確認をお願いします")

3. トレースによる可観測性

セクション0でmlflow.tracing.enable()とmlflow.openai.autolog()を設定済みのため、セクション2のエージェント実行はすべて自動的にトレースが記録されています。

3.1 トレースの確認

MLflow UIでトレース結果を確認します。

左サイドバーの エクスペリメント > llmops_workshop を開く
トレース タブをクリック
トレースを開くと以下のスパン構造が見える:

customer_support_agent
├── llm_call_1_tool_selection  ← 1回目のLLM呼び出し（ツール選択）
├── search_products            ← ツール実行（商品検索）
└── llm_call_2_final_answer    ← 2回目のLLM呼び出し（最終回答生成）

各スパンの入出力データ、レイテンシ、トークン使用量が確認できます。

3.2 トークン使用量の確認

mlflow.openai.autolog()により、トークン使用量はTraceオブジェクトのinfo.token_usageに集計されます。

_exp = mlflow.get_experiment_by_name(f"/Users/{_current_user}/llmops_workshop")
recent_traces = mlflow.search_traces(locations=[_exp.experiment_id], max_results=5)

for _, row in recent_traces.iterrows():
    trace_id = row["trace_id"]
    trace = mlflow.get_trace(trace_id=trace_id)
    token_usage = trace.info.token_usage
    print(f"Trace: {trace_id[:20]}...")
    if token_usage:
        print(f"  input_tokens:  {token_usage.get('input_tokens', 'N/A')}")
        print(f"  output_tokens: {token_usage.get('output_tokens', 'N/A')}")
        print(f"  total_tokens:  {token_usage.get('total_tokens', 'N/A')}")

>>> 直近のトレースのトークン使用量

Trace: tr-598ddfdfea9f1af03...
  token_usage: なし

Trace: tr-aeae50805c770fecc...
  input_tokens:  1337
  output_tokens: 161
  total_tokens:  1498

Trace: tr-dd3c7c0ff84c60186...
  token_usage: なし

Trace: tr-0c3d42cb14dff2eb9...
  input_tokens:  1336
  output_tokens: 138
  total_tokens:  1474

Trace: tr-a01a0538f7b15257f...
  input_tokens:  46
  output_tokens: 17
  total_tokens:  63

4. プロンプトの登録・バージョン管理

セクション2ではシステムプロンプトをコードにハードコードしていました。MLflow Prompt Registryを使うと、エージェントのコードを変更せずにプロンプトを更新できるようになります。

プロンプトはUnity Catalogの関数（Function） として保存
バージョンは自動採番（1, 2, 3, ...）で不変（イミュータブル）
エイリアス（production, staging）は特定バージョンを指すミュータブルなポインタ

4.1 プロンプトの登録（v1）

テンプレート変数は{{変数名}}の形式で定義します。

PROMPT_NAME = f"{CATALOG}.{SCHEMA}.{USERNAME}_customer_support"

template_v1 = """\
あなたは{{company_name}}のカスタマーサポートAIアシスタントです。
敬語を使い、お客様に寄り添った丁寧な口調で回答してください。

【回答ルール】
- お客様の質問に正確かつ丁寧に回答する
- 商品やサービスの質問にはsearch_productsツールを使って最新情報を検索する
- 不明な点がある場合は正直に伝え、適切な窓口を案内する
- 個人情報の取り扱いには細心の注意を払う
- 回答は簡潔にまとめる

お客様からの質問: {{question}}"""

prompt_v1 = mlflow.genai.register_prompt(
    name=PROMPT_NAME,
    template=template_v1,
    commit_message="v1: 基本的なカスタマーサポートプロンプト",
    tags={"author": USERNAME, "use_case": "customer_support"},
)

4.2 プロンプトの更新（v2）

ビジネスルールを追加した新バージョンを登録します。register_prompt()を同じ名前で再度呼び出すと、自動的に新しいバージョンが作成されます。

template_v2 = """\
あなたは{{company_name}}のカスタマーサポートAIアシスタントです。
敬語を使い、お客様に寄り添った丁寧な口調で回答してください。

【回答ルール】
- お客様の質問に正確かつ丁寧に回答する
- 商品やサービスの質問にはsearch_productsツールを使って最新情報を検索する
- 不明な点がある場合は正直に伝え、適切な窓口を案内する
- 個人情報の取り扱いには細心の注意を払う
- 競合他社の製品・サービスについて言及しない
- 料金に関する具体的な金額は、最新の公式情報を確認するよう案内する
- 回答は200文字以内で簡潔にまとめる

お客様からの質問: {{question}}"""

prompt_v2 = mlflow.genai.register_prompt(
    name=PROMPT_NAME,
    template=template_v2,
    commit_message="v2: 競合他社言及禁止、料金案内ルール、文字数制限を追加",
)

4.3 エイリアスの設定

mlflow.genai.set_prompt_alias(name=PROMPT_NAME, alias="production", version=prompt_v1.version)
mlflow.genai.set_prompt_alias(name=PROMPT_NAME, alias="staging", version=prompt_v2.version)

4.4 Prompt Registry連携エージェント

ハードコードのプロンプトを@productionエイリアスからの動的取得に切り替えます。

@mlflow.trace
def run_agent_with_registry(user_question: str, alias: str = "production"):
    """Prompt Registry連携エージェント"""
    prompt = mlflow.genai.load_prompt(f"prompts:/{PROMPT_NAME}@{alias}")
    system_msg = prompt.format(company_name=COMPANY_NAME, question=user_question)

    messages = [
        {"role": "system", "content": system_msg},
        {"role": "user", "content": user_question},
    ]
    response = client.chat.completions.create(
        model=ENDPOINT_NAME, messages=messages, tools=tools, max_tokens=500,
    )
    # ... ツール呼び出しの処理（省略）
    return assistant_msg.content

productionエイリアスの向き先をv2に切り替えると、コード変更なしで新しいルールが適用されます。

# v2に切り替え → 競合他社言及禁止・200文字制限が適用される
mlflow.genai.set_prompt_alias(name=PROMPT_NAME, alias="production", version=prompt_v2.version)

# 同じ関数をそのまま再実行 → v2のルールが反映
run_agent_with_registry("おすすめのプランを教えてください")

5. まとめ

このハンズオンでは、Databricksの機能を組み合わせて生成AIガバナンスの基盤を構築しました。

課題	Databricksの解決策	ハンズオンで実装したもの
コンプライアンスリスク	AIゲートウェイ + ガードレール	PII検出、有害コンテンツブロック、カスタムポリシー
可視性の欠如	MLflowトレーシング	エージェント処理フローの可視化、トークン使用量の記録
プロンプト管理	MLflow Prompt Registry	バージョン管理、エイリアスによる本番切り替え
監査・コスト管理	推論テーブル + Usage Tracking	リクエスト/レスポンスの記録（参考紹介）

次のステップ

エージェントのUnity Catalog登録とModel Servingへのデプロイ
RAGアプリケーションへのガードレール統合
プロンプト管理のCI/CDパイプライン設計
監査ログのダッシュボード構築

参考ドキュメント

トピック	リンク
Unity AIゲートウェイ	概要 / エンドポイント設定
ガードレール	設定ガイド
推論テーブル	設定ガイド
MLflow Prompt Registry	概要 / プロンプトの作成
MLflowトレーシング	概要 / トークン使用量の追跡
Agent Framework	エージェント構築

はじめてのDatabricks

Databricks無料トライアル

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up