ミツモアでAI関連の開発を担当している増田（@xmasudahiroto）です。

最近、社内でZendeskのヘルプページの記事をRAG化し、MCP経由でAIエージェントから利用可能にする開発を行いました。今回は、既存のデータ基盤であるdbtとBigQueryを活用し、短期間でRAGを構築した際の手順を共有します。

背景と課題

ミツモアでは顧客対応にZendeskを導入しており、プロダクトの仕様をまとめたヘルプページもZendeskのArticle機能で運用しています。

開発中のAIエージェントにおいて、「プロダクトの仕様に基づいた正確な回答」や「FAQを参照した回答」をさせたいという要望がありました。そのため、Zendeskのヘルプページの記事をRAG化し、それを参照するMCPツールを作成することになりました。

当初はZendesk自体が提供する「Zendesk AI」や他の外部RAGサービスの利用も検討しましたが、以下の理由から自前での構築を選択しました。

MCPとして統合: MCPとして実装することで、開発中のAIエージェントや社内のDifyなど、複数のエージェントから統一して利用できる環境を目指しました。
拡張性: 将来的に独自の検索ロジックやプロンプト調整が必要になった際、自社基盤の方が圧倒的に小回りが利きます。
ベンダーロックインの回避: RAG専用サービスに依存しすぎると、将来的なサービスの乗り換えが困難になるリスクを考慮しました。
コストの最適化: 構築済みのデータ基盤（BigQuery / dbt）に載せることで、追加のインフラ費用を最小限に抑えられます。

エージェントに提供する文章をZendeskのヘルプページとは別に作成することも議論しましたが、やはりZendeskを SSoT (Single Source of Truth) として活用することにメリットがあると考え、Zendeskの記事をRAGすることにしました。

RAGとは？

RAG（Retrieval-Augmented Generation）は、ユーザーの質問に関連するドキュメントを外部知識から検索（Retrieval）し、その内容をLLMに渡して回答を生成（Generation）させる手法です。

検索にあたっては、事前にテキストを「埋め込み（Embedding）」し、多次元のベクトル空間上の点として表現します。質問文も同様に埋め込みを行いベクトル化し、質問文のベクトルとドキュメントのベクトルの「距離」を計算することで、意味内容が近い情報を探し出すことができます。

アーキテクチャ

「まずは最小限の工数でクイックに実装する」という方針のもと、以下の構成を採用しました。データの抽出からベクトル化、検索までをすべてBigQueryとdbtの範囲内で完結させています。

技術的なポイント

dbt Pythonモデル: 従来、APIからのデータ取得はワークフローエンジン等の別ジョブとして切り出す必要がありましたが、Pythonモデルを使うことでdbtの実行フロー内で完結できます。
BigQuery ML: SQLから直接Vertex AIのEmbeddingモデルを呼び出せるため、追加の推論サーバーを立てる必要がありません。
スケーラビリティ: サーバーレス構成のため、記事数が増加してもパフォーマンスやインフラコストを過度に気にする必要がありません。

これにより、ミツモアの既存データ基盤に載せる形で、追加インフラなしでの開発を実現しました。

※ミツモアのデータ基盤の詳細については、こちらの記事もご参照ください。

実際の実装手順

基本的には、BigQuery公式のRAGチュートリアルに沿って進めていきます。

セマンティック検索と検索拡張生成を行う | BigQuery | Google Cloud Documentation

BigQuery側の準備

BigQuery上でAI機能を利用するために、リモートモデルを作成します。これはVertex AIへのコネクションを定義する作業です。

モデルを保存するデータセットを作成します（例：models）。
テキスト埋め込み用のリモートモデルを作成します（例：embedding_model）。

CREATE OR REPLACE MODEL `models.embedding_model`
  REMOTE WITH CONNECTION DEFAULT
  OPTIONS (ENDPOINT = 'gemini-embedding-001');

EmbeddingモデルはVertex AIのModel Gardenから選択可能です。今回は gemini-embedding-001 を使用していますが、text-embedding-005 なども選択肢に入ります。

dbt Pythonモデルで記事を抽出

Zendesk APIを叩いて記事を取得し、テーブルに格納する dbt モデル（zendesk_articles）を作成します。APIキーはSecret Managerに安全に保存してある前提です。

# zendesk_articles.py
def model(dbt, session):
    """
    Zendesk Help Center API (Articles) から記事データを取得する Python モデル。

    - 認証: Secret Manager から取得
      - JSON key: {"email": "...", "token": "..."}
      - リソース: projects/<project_id>/secrets/<secret_id>/versions/latest
    - 出力スキーマ: Zendesk API Article Object の全フィールド
    """
    dbt.config(
        submission_method="bigframes",
        materialized="table",
        tags=["proone"],
        packages=[
            "requests",
            "google-cloud-secret-manager",
            "bigframes",
        ],
    )

    import json
    import base64
    import requests
    import bigframes.pandas as bpd
    from google.cloud import secretmanager

    # 定数
    SECRET_RESOURCE_NAME = "projects/<project_id>/secrets/<secret_name>/versions/latest"
    SUBDOMAIN = "subdomain"

    # 出力カラム定義 (APIドキュメントに基づく)
    columns = [
        "id",
        "url",
        "html_url",
        "author_id",
        "comments_disabled",
        "draft",
        "promoted",
        "position",
        "vote_sum",
        "vote_count",
        "section_id",
        "created_at",
        "updated_at",
        "name",
        "title",
        "source_locale",
        "locale",
        "outdated",
        "outdated_locales",
        "edited_at",
        "user_segment_id",
        "permission_group_id",
        "content_tag_ids",
        "label_names",
        "body",
    ]

    def _get_api_creds():
        """
        Secret Manager から認証情報を取得する。
        
        【Secretの作成方法】
        以下のワンライナーでJSONを作成・エンコードし、Google Cloud Secret Managerに `zendesk_api_creds` という名前で保存してください。
        
        # Mac/Linux
        $ echo -n '{"email": "USER_EMAIL", "token": "API_TOKEN"}' | base64 | pbcopy
        
        (作成されたBase64文字列をSecretのpayloadとして保存)
        """
        client = secretmanager.SecretManagerServiceClient()
        resp = client.access_secret_version(name=SECRET_RESOURCE_NAME)
        payload = resp.payload.data.decode("utf-8")
        try:
            return json.loads(payload)
        except json.JSONDecodeError:
            # Base64 encoded JSON case
            return json.loads(base64.b64decode(payload).decode("utf-8"))

    def _fetch_from_zendesk(creds):
        email = creds.get("email")
        token = creds.get("token")
        
        if not email or not token:
            raise ValueError("Zendesk credentials missing 'email' or 'token'")

        auth = (f"{email}/token", token)
        base_url = f"https://{SUBDOMAIN}.zendesk.com/api/v2/help_center/articles.json"
        
        params = {"page[size]": 100}
        url = base_url
        all_articles = []
        
        while url:
            resp = requests.get(url, auth=auth, params=params)
            resp.raise_for_status()
            data = resp.json()
            
            articles = data.get("articles", [])
            all_articles.extend(articles)
            
            url = data.get("links", {}).get("next")
            params = None 
            
        return all_articles

    # Main Logic
    data = []

    try:
        creds = _get_api_creds()
        data = _fetch_from_zendesk(creds)
    except Exception as e:
        raise RuntimeError(f"Failed to fetch Zendesk data: {e}")

    if not data:
        return bpd.DataFrame({c: [] for c in columns})

    # Normalize data to ensure all columns exist
    normalized_data = []
    for item in data:
        row = {}
        for col in columns:
            # APIのレスポンスにないキーは None (NULL) にする
            row[col] = item.get(col)
        normalized_data.append(row)

    df = bpd.DataFrame(normalized_data)
    
    # 型変換が必要な場合ここで行うが、BigFrames/Pandasがある程度推論する
    # idなどは str にしておくと安全かもしれないが、一旦推論に任せる
    
    return df

加工とEmbeddingの生成（dbt SQLモデル）

取得した記事情報を後続のdbtモデルでEmbedding処理します。実行コストを抑えるため、incrementalモデルを利用し、前回の更新以降に変更・追加された記事のみを対象にします。

-- mart_zendesk_articles.sql
{{
  config(
    materialized='incremental',
    unique_key='id',
    incremental_strategy='merge',
  )
}}

WITH source_articles AS (
  SELECT *
  FROM {{ ref('zendesk_articles') }}
),

-- 現在のターゲットテーブルの状態を取得 (Incremental run時のみ)
{% if is_incremental() %}
current_state AS (
  SELECT id, updated_at
  FROM {{ this }}
),
{% endif %}

-- 1. 新規・更新分の特定 (Upserts)
articles_to_embed AS (
  SELECT
    s.id,
    s.url,
    s.html_url,
    s.title,
    s.body,
    s.updated_at,
    -- Embedding用コンテンツ作成
    CONCAT('Title: ', IFNULL(s.title, ''), '\nBody: ', IFNULL(s.body, '')) AS content
  FROM source_articles s
  {% if is_incremental() %}
  LEFT JOIN current_state c ON s.id = c.id
  -- 新規レコード OR 更新日時が新しいレコード OR 以前削除済みだったが復活したレコード
  WHERE c.id IS NULL 
     OR s.updated_at > c.updated_at
  {% endif %}
),

-- Embeddingの生成 (Upsert分のみ)
embedded_articles AS (
  SELECT
    id,
    url,
    html_url,
    title,
    body,
    updated_at,
    content,
    embedding,
    statistics,
    status,
    FALSE AS is_deleted
  FROM AI.GENERATE_EMBEDDING(
    MODEL `models.embedding_model`,
    (
      SELECT
        id,
        url,
        html_url,
        title,
        body,
        updated_at,
        content
      FROM articles_to_embed
    )
  )
)

{% if is_incremental() %}
,
-- 2. 削除分の特定 (Logically Deleted)
-- ソーステーブルに存在しないIDは、ターゲットテーブル(this)から既存レコードを取得し、is_deleted=TRUEにする
deletions AS (
  SELECT
    c.id,
    c.url,
    c.html_url,
    c.title,
    c.body,
    c.updated_at,
    c.content,
    c.embedding,
    c.statistics,
    c.status,
    TRUE AS is_deleted
  FROM {{ this }} c
  WHERE c.id NOT IN (SELECT id FROM source_articles)
  AND c.is_deleted = FALSE -- 既に削除済みのものは更新不要
)

SELECT * FROM embedded_articles
UNION ALL
SELECT * FROM deletions

{% else %}

-- 初回実行時 (Full Refresh)
SELECT * FROM embedded_articles

{% endif %}

ベクトル検索（RAG）の実行

RAGを実行する際は、ユーザーのクエリも同じモデルでEmbedding化し、VECTOR_SEARCH 関数を利用して類似度検索を行います。この処理をMCPで定義してサーバーに載せることで、外部からエージェントがRAGを利用できるようになります。

SELECT
    query.query,
    base.id,
    base.html_url,
    base.title,
    base.body,
    base.updated_at
FROM VECTOR_SEARCH(
    TABLE `dbt.proone_mid_zendesk_articles_embedded`,
    'embedding',
    (
        SELECT embedding, content AS query
        FROM AI.GENERATE_EMBEDDING(
            MODEL `ml.embedding_model`,
            (SELECT @searchQuery AS content)
        )
    ),
    top_k => @topK
)
WHERE NOT base.is_deleted

Indexによる高速化

記事数が増えてパフォーマンスが課題になる場合は、ベクトルインデックスを貼ることで高速化が可能です。

CREATE OR REPLACE VECTOR INDEX my_index
ON `<dbt_dataset>.mart_zendesk_articles`(embedding)
OPTIONS(index_type = 'IVF',
  distance_type = 'COSINE',
  ivf_options = '{"num_lists":500}')

AIエージェント（MCPツール）での活用イメージ

構築したRAGは、MCPサーバーを通じてAIエージェントに提供されます。エージェントはユーザーの質問に応じて、以下のようなツール定義に従ってBigQueryへアクセスします。

{
  "name": "search_help_center",
  "description": "ミツモアのプロダクト仕様やFAQについて検索します。",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": { "type": "string", "description": "検索クエリ" }
    }
  }
}

このツールが呼ばれると、裏側で前述の VECTOR_SEARCH クエリが実行され、上位数件の記事がエージェントのコンテキストに注入されます。これにより、最新の仕様に基づいた正確な回答が可能になります。

運用上の留意点と制約

今回は工数をかけずに実装することを目指したため、以下のようなトレードオフが存在します。

モデルの選択肢: BigQuery MLの制約上、Vertex AIでサポートされているモデルに限定されます。
リアルタイム性: dbtのバッチ実行間隔に依存するため、Zendesk側の変更が即座に反映されるわけではありません。
チャンク分割: 今回は記事単位でEmbeddingしていますが、極端に長いドキュメントを扱う場合は、dbt内で適切なChunking処理を検討する必要があります。
ハイブリッド検索: 今回はベクトル類似度のみの検索です。単語の完全一致を重視したい場合は、全文検索エンジンを組み合わせるなど、より高度なRAG構成が必要です。

まとめ

BigQueryとdbtをそのままRAGのエンジンとして活用することで、追加のインフラ管理コストをほぼゼロに抑えつつ、最小限の工数でRAGを実現できます。今回はZendeskのヘルプページの記事での例でしたが、他のサービスでもAPI連携部分を変えれば同様の方針でできます。BigQueryとdbtをデータ基盤としている会社は多いと思うので、参考になれば幸いです。