【2026年最新版】RAGシステムのコストをグッと下げる具体手法7選

RAGシステムを本番運用していると、じわじわと気になってくるのが生成AIのAPI費用だ。月数万円で済んでいたはずが、利用者が増えるにつれて青天井になっていく。この記事では、コードを交えながら今すぐ実践できるコスト削減手法を7つ紹介する。

前提：RAGのコストはどこで発生するか

まず敵を知ることから始める。RAGのAPI費用は主に以下の3箇所だ。

このうちLLM推論が最もコスト比率が高い。ここを削るのが最優先だ。

手法1：モデルルーティング（コスト削減率：40〜60%）

全クエリを同じモデルに投げるのをやめる。質問の難易度に応じてモデルを使い分けるだけで、コストは劇的に下がる。

ポイントはルーティング自体にLLMを使わないこと。Embeddingの類似度やルールベースで判定すれば、ルーティングコストはほぼゼロだ。

Embeddingベースのルーター実装

import numpy as np
from openai import OpenAI

client = OpenAI()

# 難易度の基準となるサンプル集（事前に用意）
SIMPLE_EXAMPLES = [
    "営業時間を教えてください",
    "住所はどこですか",
    "返品ポリシーを教えて",
    "料金プランは何種類ありますか",
]

COMPLEX_EXAMPLES = [
    "契約書の第3条と第7条が矛盾している場合の対処法を教えてください",
    "去年と今年の売上を比較して季節変動を除いた実績を分析してほしい",
    "複数の法令が絡む場合のコンプライアンス対応手順を教えてください",
]

def get_embedding(text: str) -> np.ndarray:
    res = client.embeddings.create(
        model="text-embedding-3-small",  # largeの約1/5のコスト
        input=text
    )
    return np.array(res.data[0].embedding)

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 起動時に一度だけEmbeddingしておく（リクエストごとには不要）
simple_vecs = [get_embedding(t) for t in SIMPLE_EXAMPLES]
complex_vecs = [get_embedding(t) for t in COMPLEX_EXAMPLES]

def route_query(query: str) -> str:
    q_vec = get_embedding(query)
    sim_simple  = np.mean([cosine_similarity(q_vec, v) for v in simple_vecs])
    sim_complex = np.mean([cosine_similarity(q_vec, v) for v in complex_vecs])

    # 閾値0.05は実運用データで調整する
    if sim_complex > sim_simple + 0.05:
        return "claude-sonnet-4-20250514"   # 複雑 → 高性能モデル
    else:
        return "claude-haiku-4-5-20251001"  # 単純 → 低コストモデル

# 使用例
model = route_query("先月の売上から季節変動を除いた実績を出してほしい")
print(f"使用モデル: {model}")  # → claude-sonnet-4-20250514

手法2：プロンプトキャッシング（コスト削減率：最大90%）

AnthropicのAPIにはプロンプトキャッシング機能がある。システムプロンプトや参照ドキュメントをキャッシュしておくと、2回目以降のリクエストでは入力トークンのコストが約90%オフになる。

RAGのように「長い文書をコンテキストに含める」用途では特に効果が大きい。

import anthropic

client = anthropic.Anthropic()

# ドキュメントの内容（例：社内規程PDF全文など長大なテキスト）
KNOWLEDGE_BASE = """
[ここに参照させたい長いドキュメントを入れる]
...（数千〜数万トークン）
"""

def ask_with_cache(user_query: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "あなたは社内規程に詳しいアシスタントです。以下のドキュメントを参照して回答してください。",
            },
            {
                "type": "text",
                "text": KNOWLEDGE_BASE,
                "cache_control": {"type": "ephemeral"},  # ← ここがキモ
            }
        ],
        messages=[
            {"role": "user", "content": user_query}
        ]
    )
    return response.content[0].text

# 2回目以降はKNOWLEDGE_BASEのトークンがキャッシュから読まれる
answer = ask_with_cache("有給休暇の申請手順を教えてください")

キャッシュの料金体系（2026年3月時点）

同じコンテキストを繰り返し使うユースケースなら、早期に導入を検討すべき機能だ。

手法3：セマンティックキャッシュ（コスト削減率：20〜50%）

「営業時間は何時ですか？」と「何時から営業してますか？」は意味が同じだ。このような意味的に重複するクエリをキャッシュで返すことで、LLM呼び出しを丸ごとスキップできる。

import numpy as np
import json
from datetime import datetime, timedelta

class SemanticCache:
    def __init__(self, threshold: float = 0.92, ttl_hours: int = 24):
        self.threshold = threshold  # 類似度の閾値（高いほど厳しく判定）
        self.ttl = timedelta(hours=ttl_hours)
        self.cache: list[dict] = []  # 本番はRedisやFaissを使うこと

    def _embed(self, text: str) -> np.ndarray:
        res = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return np.array(res.data[0].embedding)

    def get(self, query: str) -> str | None:
        q_vec = self._embed(query)
        now = datetime.now()

        for entry in self.cache:
            # TTL切れはスキップ
            if now - entry["created_at"] > self.ttl:
                continue
            sim = cosine_similarity(q_vec, entry["vec"])
            if sim >= self.threshold:
                print(f"[Cache HIT] similarity={sim:.3f}")
                return entry["answer"]
        return None

    def set(self, query: str, answer: str):
        self.cache.append({
            "vec": self._embed(query),
            "answer": answer,
            "created_at": datetime.now(),
        })

# 使用例
cache = SemanticCache(threshold=0.92)

def rag_with_cache(query: str) -> str:
    # まずキャッシュを確認
    cached = cache.get(query)
    if cached:
        return cached  # LLM呼び出しなし

    # キャッシュミスなら通常のRAGパイプラインへ
    answer = run_rag_pipeline(query)  # 既存のRAG処理
    cache.set(query, answer)
    return answer

実運用ではcacheをRedisのベクトル検索（Redis Stack）に置き換えると、スケールしやすくなる。

手法4：Top-K削減＋リランキング（コスト削減率：20〜35%）

「とりあえず多めに取っておこう」とTop-K=10にしているケースをよく見る。チャンクをLLMに渡すトークンは直接コストに効くため、Top-Kを絞ってリランキングで精度を維持するアプローチが効果的だ。

from sentence_transformers import CrossEncoder

# クロスエンコーダー（ローカル実行可能・無料）
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def search_and_rerank(query: str, vector_db, top_k_retrieve=20, top_k_final=3):
    # まず多めに取得（ベクトル検索は安い）
    candidates = vector_db.search(query, top_k=top_k_retrieve)

    # クロスエンコーダーでリランキング
    pairs = [(query, chunk.text) for chunk in candidates]
    scores = reranker.predict(pairs)

    # スコア順にソートして上位3件だけLLMに渡す
    ranked = sorted(zip(scores, candidates), reverse=True)
    top_chunks = [chunk for _, chunk in ranked[:top_k_final]]

    return top_chunks

# コンテキストトークン数の比較
# Top-K=10のまま → 約3,000トークン
# Top-K=3（リランキング後）→ 約900トークン（▲70%）

手法5：Embeddingのローカル化（コスト削減率：Embedding費用の100%）

Embeddingは生成時と検索時の両方でAPIを叩く。ドキュメント数が多いほど積み上がるコストだ。ローカルモデルに切り替えればEmbedding費用をゼロにできる。

# Ollamaをインストール（macOS/Linux）
curl -fsSL https://ollama.com/install.sh | sh

# 多言語対応の軽量Embeddingモデルを取得
ollama pull nomic-embed-text

import ollama

def embed_local(text: str) -> list[float]:
    response = ollama.embeddings(
        model="nomic-embed-text",
        prompt=text,
    )
    return response["embedding"]

# OpenAI Embeddingと差し替えるだけ
chunks = load_documents()
for chunk in chunks:
    vec = embed_local(chunk.text)  # API費用ゼロ
    vector_db.upsert(chunk.id, vec, chunk.text)

モデル比較（日本語を含む多言語タスク）

手法6：要約型RAG（コスト削減率：30〜60%）

長いドキュメントをそのままチャンキングしてLLMに渡すのは非効率だ。インデックス作成時に要約を生成しておき、推論時はコンパクトな要約を使うことでトークン数を大幅に削減できる。

def build_summary_index(documents: list[str]) -> list[dict]:
    """インデックス作成時（1回だけ実行）"""
    index = []
    for doc in documents:
        # 要約はバッチで生成してコストを管理しやすくする
        summary = client.messages.create(
            model="claude-haiku-4-5-20251001",  # 要約は安いモデルで十分
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"以下のドキュメントを200字以内で要約してください:\n\n{doc}"
            }]
        ).content[0].text

        index.append({
            "summary": summary,        # LLMへの入力に使う（短い）
            "original": doc,           # 必要なら参照
            "vec": embed_local(summary),  # 要約でEmbedding
        })
    return index

def rag_with_summary(query: str, index: list[dict]) -> str:
    """推論時（毎リクエスト）"""
    q_vec = embed_local(query)

    # 要約ベクトルで検索
    results = sorted(index, key=lambda x: cosine_similarity(q_vec, x["vec"]), reverse=True)[:3]

    # LLMには要約だけ渡す（originalではなく）
    context = "\n\n".join([r["summary"] for r in results])

    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"以下の情報をもとに質問に答えてください。\n\n{context}\n\n質問: {query}"
        }]
    )
    return response.content[0].text

手法7:ルールベースルーティング（コスト削減率：即効性あり）

最後はシンプルなルールベースだ。実装コストが最も低く、即日適用できる。

import re

ROUTING_RULES = [
    # (条件, 使用モデル)
    (lambda q: len(q) < 30,                              "claude-haiku-4-5-20251001"),
    (lambda q: re.search(r"(住所|電話|営業時間|料金)", q), "claude-haiku-4-5-20251001"),
    (lambda q: re.search(r"(分析|比較|矛盾|原因|戦略)", q),"claude-sonnet-4-20250514"),
    (lambda q: len(q) > 200,                              "claude-sonnet-4-20250514"),
]

DEFAULT_MODEL = "claude-haiku-4-5-20251001"

def rule_based_route(query: str) -> str:
    for condition, model in ROUTING_RULES:
        if condition(query):
            return model
    return DEFAULT_MODEL

# テスト
print(rule_based_route("住所を教えて"))              # → haiku（安い）
print(rule_based_route("昨年比の売上低迷の原因分析")) # → sonnet（高性能）

実運用ではログを蓄積して「本当はSonnetが必要だったのにHaikuで回答された」ケースを定期的に見直し、ルールを育てていくのがコツだ。

まとめ：優先順位と期待効果

これらを組み合わせると、トータルで50〜80%のコスト削減も現実的だ。まずプロンプトキャッシングとルーティングの2つから始めて、効果を計測しながら積み上げていくのをおすすめする。

コスト削減は「全部一気にやる」より「計測→改善→計測」のサイクルが重要だ。CloudWatchやNew RelicでAPIコストをダッシュボード化しておくと、どの施策が効いているか一目でわかるようになる。