token消化ではなく、記事のご縁を可視化！さくらのAI Engine RAGでチャレンジ参加記事の相関図を描いてみた🌸

TL;DR: 「さくらのAIチャレンジ」のQiita参加記事を、さくらのAI EngineのRAG(ドキュメント検索)に投入し、意味的に近い記事同士をMermaidの相関図でつないでみました。22件を収集し、イベントページへのリンクがない10件を除外した残り12件をRAGに投入して15本のエッジが生まれ、いいね数ランキングとは違う「話題のハブ」が見えてきました。実行後はRAG上のドキュメントを(埋め込みが間に合わなかった分も含めて)削除し、保持課金を残さないようにしています。

対象読者: さくらのAI EngineでRAGを実際に試してみたいエンジニア、記事や文書同士の「意味的な近さ」を手を動かして可視化してみたい人。(注: 基盤モデル無償プランでもRAGのドキュメント保管・ベクトル化には無償枠がなく、従量課金であることに注意。Chat Completions・音声・Embeddingsには無償枠があるが、RAGは対象外。参考までに、アカウント全体の今月のembeddings利用料金は(前日時点の集計で)3円・603リクエストだった。この集計は前日までの反映のため、今回の検証分はまだ含まれていない可能性が高いが、それでも小規模な検証であれば金額面のハードルは低そうだ)

RAGを、回答生成ではなく「近さ」の計測に使う

RAG(Retrieval-Augmented Generation)は、LLMに質問を投げる前に関連しそうな文書を検索し、その内容も一緒に渡すことで回答の根拠を広げる仕組みだ。社内ドキュメントや最新情報のように、LLMが元々知らない知識を答えさせたいときによく使われる。

今回はこの「関連文書を検索する」という部分だけを借りた。検索で得られる距離(近いか遠いか)を、回答生成のためではなく「この記事とあの記事は意味的にどれくらい近いか」を測るために使っている。RAGの「検索」を、記事同士のご縁を測る虫眼鏡として使ってみた、というのが今回の趣旨だ。

きっかけは、毎日育つランキング記事

発端は、@torifukukaiou さんのこの記事だった。

【毎日自動更新】さくらのAI Engine 3,000リクエスト使い切りチャレンジ いいねランキング！
https://qiita.com/torifukukaiou/items/c4f654f0dad56e5da409

Elixir/Nervesで書いた自動更新プログラムが、Qiita APIから「さくらのAIチャレンジ」の参加記事を収集し、いいね数ランキングやタグ集計を毎日更新してくれる。しかもこの記事自体がキャンペーンへの応募作品でもあり、「さくらのAI Engine自身が応募している構図」を楽しんでいる。この入れ子になった構図と、「token消化ではなく、闘魂昇華」という言葉遊びに、すっかり心を掴まれてしまった。

ランキングが記事の熱量を見せてくれるなら、いいね数やタグとは別の切り口で参加記事を眺めたら、どんな景色になるのだろうか。そう思い、本文の意味的な近さを使って、参加記事の相関図を作ってみることにした。これもまた、一種のキャンペーン参加作品になっている。

処理の流れ

やったことは次の5段階になる。

1. Qiita API v2で「さくらのAI」タグの記事を収集し、イベントページへのリンクを含む記事に絞り込む
2. 各記事本文をさくらのAI EngineのRAGドキュメントAPI(/v1/documents/upload/)へアップロード(embeddingモデルは multilingual-e5-large)
3. 埋め込み完了(status: available)を待ってから、各記事本文を使って /v1/documents/query/ に問い合わせ、コサイン距離が近い他の記事を検索
4. 記事ごとに近い上位k件を「エッジ」としてつなぎ、Mermaidの相関図を生成
5. 保持課金を避けるため、実行後にRAGドキュメントは /v1/documents/{id}/ へのDELETEで削除

実装で踏んだ三つの小石

手を動かしてみると、APIドキュメントを読んだだけでは気づきにくい点がいくつかあった。

一つ目は、documents/query/ が記事単位ではなくチャンク単位でマッチを返すことだった。長い記事は512トークンごとに分割されてアップロードされるため、そのまま上位k件を取ると、同じ記事の複数チャンクで埋まってしまう。ドキュメントID(記事)単位で重複排除してから、記事単位のtop-kを選ぶようにした。ただしdocuments/query/自体が返すチャンク数にも上限があるため、重複排除した後にtop_k件ちょうどの異なる記事を確保できるとは限らない — 実際には「上位チャンクから重複等を除いた、その時点で拾えた近傍記事」という位置づけになる。

二つ目は、埋め込み処理が非同期であることだ。アップロード直後はまだ検索対象にならないため、ドキュメント一覧APIの status が available になるまでポーリングして待つ作りが必要だった。

三つ目は、Mermaidラベルのエスケープだ。記事タイトルをそのままノードのラベルに使うと、[ や ]、" などの記号が構文を壊すことがある。表示用のラベルだけ記号を置換しておく必要があった。

なお、記事同士の近さを測るクエリには、本文全体ではなく「タイトル＋本文冒頭500文字」を使っている。埋め込み対象(RAGにアップロードされるチャンク側)は本文全体だが、比較の起点となるクエリ側は冒頭部分に限られる、という非対称な比較になっている点は留意しておきたい。

さらに、記事Aから記事Bへのクエリ結果と、記事Bから記事Aへのクエリ結果は、原理的には別の距離になり得る(方向によって異なるチャンクがヒットするため)。相関図では、両方向の候補が得られた場合はその小さい方(＝より近いと判定された方)を採用し、A-B間の無向のエッジ1本にまとめている。つまり表示されている距離は「AとBの本文同士の対称な距離」ではなく、「どちらか近い方向で観測された距離の最小値」だ。

咲いた相関図がこちら

収集できた参加記事は22件。そのうちイベントページへのリンクを含まない10件をサニティフィルタで除外し、残る12件をRAGに投入した(アップロード失敗は0件)。生成されたエッジは15本だった。数値は各エッジのコサイン距離で、小さいほど意味的に近い。

記事タイトル	URL	いいね数	次数
Laravel×さくらのAI Engineで自分専用の記事下書きツールを作る。	https://qiita.com/miyacha/items/f4db3964afad5142113e	1	2
さくらのAI Engineを無料枠で触ってみて、最初につまずいた話	https://qiita.com/TheGateBreaker/items/855ee917db41fbe11fdd	1	1
MoodleのAIプロバイダでさくらのAI Engineをつかいたおす	https://qiita.com/papillon/items/d0d64609fc63d659a7ef	0	1
満開のさくら街道をゆく――さくらのAI Engine無償枠3,000回を従量課金で歩き直してみた	https://qiita.com/torifukukaiou/items/b2e3ec8a4fb986aa786f	0	2
生成AI初心者がAIに聞きながら「さくらのAI Engine」無償プランを契約〜PowerShellで初APIコールしてみた【公式ドキュメント答え合わせ付き】	https://qiita.com/miyacha/items/93a667c6117b1d46db55	2	4
さくらのAI Engineに書いてもらったコードで、さくらのAI Engineの「ずんだもん」に毎朝私は起こしてもらう	https://qiita.com/torifukukaiou/items/3c03c45da77e9cb7cf6b	7	1
【毎日自動更新】さくらのAI Engine 3,000リクエスト使い切りチャレンジ いいねランキング！	https://qiita.com/torifukukaiou/items/c4f654f0dad56e5da409	1	5
Claude Code無課金勢、LINE botを「さくらのAI Engine」のOpenAI互換Responses APIへ移行する作業を、「さくらのAI Engine」自身に実装させたらCodexレビューを通過した	https://qiita.com/torifukukaiou/items/c3adf542824e4d5ae7b2	2	2
OpenAI互換API、名乗るからには本物だった件 ── ai& から さくらのAI Engineへ、ハマりどころゼロで移行完了	https://qiita.com/torifukukaiou/items/747284dcd98157bdfad6	2	2
さくらのAI Engine Playgroundに『「さくらのAI Engine」3,000リクエスト使い切りチャレンジ』キャンペーンの受賞のコツを聞いてみました	https://qiita.com/torifukukaiou/items/38f5cd9fbcd60bcbe627	0	7
さくらのAI Engine Playgroundに記事を書いてもらいました（応答が速い🚀🚀🚀)	https://qiita.com/torifukukaiou/items/5a3fd81908bcdb0ee59e	0	1
さくらのAI Engine無料枠でgpt-oss-120bを使う ― Anthropic SDKはauth_token=必須の罠	https://qiita.com/nomurasan/items/63654a3f9457a8b5ed35	10	2

同じ相関図を、円の大きさ・濃さ(次数)と線の太さ(近さ)で見やすくした画像がこちら。

ドラッグやホバーで各記事のタイトル・いいね数・次数を確認できるインタラクティブ版も作った(GitHub上では実行できないため、ダウンロードしてブラウザで開いてほしい): correlation_network_interactive.html

いいね数と「つながる力」は、どうやら別の指標だった

最も多くの記事につながったのは、いいね0・次数7の「Playgroundに受賞のコツを聞いてみました」だった。Playground、無料枠、キャンペーン攻略という話題は、技術スタックを問わず他の記事とも共鳴しやすいのかもしれない。「毎日自動更新のいいねランキング」記事も次数5で、もう一つのハブになっている。どちらも特定のAPI用途というより、キャンペーンそのものを話題にしたメタ的な記事同士が近くなるのは、考えてみれば自然な結果だ。

一方、いいね10の「gpt-oss-120bを使う」記事は次数2にとどまった。人気と意味的な接続数は、別の指標らしい。「ずんだもん」に起こしてもらう記事や、Playgroundに記事を書いてもらった記事は次数1で、音声合成や実演ネタという内容の独自性が、線の少なさとしてそのまま現れているように見える。

いいね数(横軸)と次数(縦軸)を並べてみると、この「別の指標らしさ」がはっきりする。

副産物として、記事本文の代わりに、さくらのAI Engineの各機能を説明した文章(機能名そのものではなく、機能の内容を数十〜百数十字で説明したテキスト)をRAGにクエリしてみる実験もしてみた。

機能	近かった記事(コサイン距離)
Chat Completions	gpt-oss-120bを使う記事(0.11)、PowerShell初APIコール記事(0.12)、Moodle記事(0.12)
RAG / ドキュメント検索	PowerShell初APIコール記事(0.14)
Embeddings	受賞のコツ記事(0.15)、ずんだもん記事(0.16)、Moodle記事(0.16)
音声文字起こし	ずんだもん記事(0.14)
音声読み上げ	ずんだもん記事(0.14)
AI Engine Playground	PowerShell初APIコール記事(0.11)、Playgroundに記事を書いてもらった記事(0.11)
無料枠・従量課金	受賞のコツ記事(0.13)、ずんだもん記事(0.13)、PowerShell初APIコール記事(0.14)
OpenAI/Anthropic互換API	LINE bot移行記事(0.11)、gpt-oss-120b記事(0.11)

記事対記事の比較だけでなく、興味のある機能の説明文から記事をたどる入口としても、RAGは使えそうだ。

動かしたサンプルコード

環境変数 SAKURA_AI_ACCOUT_KEY にさくらのAI Engineで発行したAPIキーを設定し(ACCOUT はスペルミスに見えるが、このスクリプトが読み込む変数名としてそのまま使う)、pip install requests してから実行できる。

#!/usr/bin/env python3
"""Qiitaのタグ検索結果を、さくらのAI EngineのRAG(ドキュメント検索)に投入して、
記事同士の類似度から相関図(Mermaid)を作るサンプルスクリプト。

前提:
- 環境変数 SAKURA_AI_ACCOUT_KEY にさくらのAI Engineで発行したAPIキーを設定しておく
  (「ACCOUT」はスペルミスに見えるが、このスクリプトが読み込む変数名としてそのまま使う)
- pip install requests
"""

from __future__ import annotations

import hashlib
import os
import time
import uuid

import requests

QIITA_API_BASE = "https://qiita.com/api/v2"
SAKURA_API_BASE = "https://api.ai.sakura.ad.jp/v1"
EVENT_URL = "https://qiita.com/official-events/bd14d28b53326d318fec"


def search_qiita_by_tag(
    tag: str, since: str | None = None, per_page: int = 100, max_pages: int = 10
) -> list[dict]:
    """Qiita APIをタグ検索し、必要なフィールドだけを取り出す(認証なしでも動く)。
    イベントページへのリンクを本文に含む記事だけに絞り込む(誤検出防止)。"""
    query = f"tag:{tag}"
    if since:
        query += f" created:>={since}"
    articles: list[dict] = []
    for page in range(1, max_pages + 1):
        response = requests.get(
            f"{QIITA_API_BASE}/items",
            params={"query": query, "page": page, "per_page": per_page},
            timeout=30,
        )
        response.raise_for_status()
        items = response.json()
        if not items:
            break
        articles.extend(
            {"id": str(item["id"]), "title": item["title"], "url": item["url"], "body": item["body"]}
            for item in items
        )
    return [article for article in articles if EVENT_URL in article["body"]]


def _sakura_headers() -> dict[str, str]:
    api_key = os.environ["SAKURA_AI_ACCOUT_KEY"]
    return {"Authorization": f"Bearer {api_key}"}


def upload_document(text: str, name: str, tag: str) -> str:
    """記事本文をさくらのAI EngineのRAGドキュメントとしてアップロードする。"""
    response = requests.post(
        f"{SAKURA_API_BASE}/documents/upload/",
        headers=_sakura_headers(),
        files={"file": (name, text.encode("utf-8"), "text/plain")},
        data={"name": name, "tags": tag, "model": "multilingual-e5-large", "chunk_size": "512"},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["id"]


def wait_until_available(document_id: str, tag: str, timeout_sec: int = 120) -> bool:
    """埋め込み処理が終わり、ドキュメントが検索可能(status=available)になるまで待つ。
    一覧は複数ページに分かれることがあるため、全ページを見てから次のポーリングに進む。"""
    deadline = time.monotonic() + timeout_sec
    while time.monotonic() < deadline:
        page = 1
        while True:
            response = requests.get(
                f"{SAKURA_API_BASE}/documents/",
                headers=_sakura_headers(),
                params={"tag": tag, "page": page},
                timeout=30,
            )
            if response.status_code == 404:
                break  # ページ範囲外は「これ以上ページがない」とみなし、例外にしない
            response.raise_for_status()  # 404以外のエラー(401/429/500等)は通常どおり例外にする
            results = response.json().get("results", [])
            if not results:
                break
            for doc in results:
                if doc["id"] == document_id and doc["status"] == "available":
                    return True
            page += 1
        time.sleep(3)
    return False


def query_similar(text: str, tag: str) -> list[dict]:
    """指定タグ内のドキュメントから、テキストに距離の近いチャンクを検索する(cosine距離、小さいほど類似)。
    返るチャンク数自体がAPI側の仕様に従うため、重複排除した後にtop_k件ちょうどの
    異なる記事が確保できるとは限らない。"""
    response = requests.post(
        f"{SAKURA_API_BASE}/documents/query/",
        headers=_sakura_headers(),
        json={"model": "multilingual-e5-large", "tags": [tag], "query": text, "distance_type": "cosine"},
        timeout=30,
    )
    response.raise_for_status()
    matches = [
        {"document_id": r["document"]["id"], "distance": r["distance"]} for r in response.json()["results"]
    ]
    return sorted(matches, key=lambda m: m["distance"])


def delete_document(document_id: str) -> bool:
    """204が返れば削除成功。それ以外のステータスや通信エラーはFalseを返す(呼び出し側で警告し、
    他のドキュメントの削除は継続する)。"""
    try:
        response = requests.delete(
            f"{SAKURA_API_BASE}/documents/{document_id}/", headers=_sakura_headers(), timeout=30
        )
    except requests.RequestException:
        return False
    return response.status_code == 204


def _node_id(article_id: str) -> str:
    return f"n_{hashlib.sha1(article_id.encode()).hexdigest()[:8]}"


def _mermaid_label(title: str) -> str:
    """Mermaidのノードラベル構文を壊す記号(`\\` `"` `[` `]`)をエスケープ/置換する。"""
    return title[:20].replace("\\", "\\\\").replace('"', "'").replace("[", "(").replace("]", ")")


def build_correlation_graph(tag: str, corpus_tag: str, top_k: int = 3) -> str:
    """記事同士のRAG類似度からMermaidの相関図(graph LR)を組み立てる。"""
    if top_k <= 0:
        return "graph LR\n"

    articles = search_qiita_by_tag(tag)

    uploaded_ids: list[str] = []
    doc_to_article: dict[str, str] = {}
    try:
        for article in articles:
            document_id = upload_document(article["body"], name=f"{article['id']}.md", tag=corpus_tag)
            uploaded_ids.append(document_id)
            if wait_until_available(document_id, corpus_tag):
                doc_to_article[document_id] = article["id"]

        edges: dict[tuple[str, str], float] = {}
        for article in articles:
            own_document_id = next(
                (doc_id for doc_id, art_id in doc_to_article.items() if art_id == article["id"]), None
            )
            if own_document_id is None:
                continue  # 埋め込みがタイムアウトした記事はスキップ

            matches = query_similar(f"{article['title']}\n{article['body'][:500]}", corpus_tag)
            seen_articles: set[str] = set()
            for match in matches:
                other_id = doc_to_article.get(match["document_id"])
                if other_id is None or other_id == article["id"] or other_id in seen_articles:
                    continue  # 自分自身・埋め込み未完了・同じ記事の別チャンクは除外
                seen_articles.add(other_id)
                key = tuple(sorted((article["id"], other_id)))
                if key not in edges or match["distance"] < edges[key]:
                    edges[key] = match["distance"]
                if len(seen_articles) >= top_k:
                    break

        lines = ["graph LR"]
        for article in articles:
            lines.append(f'    {_node_id(article["id"])}["{_mermaid_label(article["title"])}"]')
        for (left_id, right_id), distance in sorted(edges.items()):
            lines.append(f"    {_node_id(left_id)} ---|{distance:.2f}| {_node_id(right_id)}")

        return "\n".join(lines) + "\n"
    finally:
        # available にならなかったドキュメントも含め、アップロード済み全件の削除を試みる。
        # 1件の失敗でループ全体を止めないよう、失敗しても次のドキュメントの削除に進む。
        for document_id in uploaded_ids:
            if not delete_document(document_id):
                print(f"Warning: failed to delete document {document_id}")


if __name__ == "__main__":
    # 実行のたびにユニークなタグを使う。固定タグだと、削除に失敗した過去実行分や
    # 並行実行分のドキュメントが同じタグの検索結果に混ざってしまう。
    unique_corpus_tag = f"rag-correlation-demo-{uuid.uuid4().hex[:8]}"
    print(build_correlation_graph(tag="さくらのAI", corpus_tag=unique_corpus_tag))

実行翌日、利用状況ダッシュボードを再確認したところ、利用料金は3円のままだった。この結果を見る限り、削除しても課金からは外れていない可能性が高そうだ。おそらく、その月に保存した最大チャンク数に対して課金される仕組みなのではないかと今は推測している。正確な課金ルールは公式には確認できていないので、引き続き観察していきたい。

まとめ――ランキングの次は、関係性が咲く？

いいねランキングが記事の盛り上がりを見せてくれる一方、RAGの相関図は、記事同士に流れる共通の話題を浮かび上がらせてくれた。任意の機能の説明文をクエリにすれば、「音声読み上げに近い記事」「互換APIに近い記事」といった、別の入口も作れそうだ。

記事がさらに増えたとき、いまのハブはハブのままなのだろうか。それとも新しいテーマを中心に、別の枝が伸びていくのだろうか。3,000リクエストをただ消化するのではなく、記事のご縁へ昇華してみるのも、悪くない選択かもしれない。

関連URL

• さくらのAI Engine(サービスページ)
• 「さくらのAI Engine」3,000リクエスト使い切りチャレンジ(Qiita公式イベントページ) — 本記事もこのキャンペーンへの参加作品
• さくらのAI Engine ドキュメント・RAG APIの使い方(公式マニュアル) — 本記事のコードが使っている/v1/documents/upload/・/v1/documents/query/等のAPI仕様
• 【毎日自動更新】さくらのAI Engine 3,000リクエスト使い切りチャレンジ いいねランキング！(@torifukukaiouさん、インスパイア元の記事)