5
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@tms-ducvu(Mr. 403)in株式会社TOMOSIA VIET NAM

RAGのコストが高い？Intent Filterで70%のトークンを節約する方法

5
Posted at

はじめに

すべてのユーザーメッセージに対してRAGパイプラインを実行していませんか？「こんにちは」や「ありがとう」といった挨拶にも？

それは私が見てきた中で最も早くお金を燃やす方法です。

この記事では、**Intent Filter（Smart Routing）**というシンプルだが効果的なパターンを紹介します。レスポンス品質に影響を与えずにRAGコストを大幅に削減できます。

対象読者: RAGベースのチャットボットを構築・運用しているエンジニア

前提知識: RAGパイプラインとLangGraph/LangChainの基本的な理解

背景 / 課題

RAGパイプラインの実際のコスト

ユーザーがメッセージを送信するたびに、典型的なRAGパイプラインは3つの高コストなステップを実行します：

User Query → Embedding → Vector Search → LLM Generation
              ↓              ↓                ↓
           $0.0001        $0.0002          $0.002+
           (per query)    (per search)     (per 1K tokens)
コスト項目 説明 推定コスト
Embedding Cost クエリをベクトルに変換 ~$0.0001/query
Vector DB Read top-kドキュメントの検索 ~$0.0002/query
LLM Context Window ドキュメント+クエリをプロンプトに挿入 ~$0.002-0.02/query

問題：40%のクエリはRAGを必要としない

実際のカスタマーサポートチャットボットのログ分析結果：

総クエリ数: 10,000/日
├── 雑談（挨拶）: 25% → 「こんにちは」「ありがとう」「さようなら」
├── オフトピック: 15% → 「今日の天気は？」
└── 実際のナレッジクエリ: 60% → 本当にRAGが必要

40%のクエリが無意味にRAGを実行しています。

10,000クエリ/日の場合、無駄になっているコスト：

  • 4,000 × $0.003 = $12/日 = $360/月

$360/月は小さく聞こえるかもしれませんが、100Kクエリ/日にスケールすると**$3,600/月**になります。これはエンジニア1人を追加雇用できる金額です。

なぜ単純なアプローチは失敗するのか？

ナイーブな方法 #1: キーワードマッチング

# ❌ ナイーブなアプローチ - バイパスされやすく、false positiveが多い
CHITCHAT_KEYWORDS = ["こんにちは", "hello", "ありがとう", "thanks"]

def is_chitchat(query):
    return any(kw in query.lower() for kw in CHITCHAT_KEYWORDS)

問題点：

  • 「こんにちは、返金ポリシーについて質問があります」→ False positive
  • 「Hi there! What's your return policy?」→ 検出漏れ
  • タイポ、スラング、自然言語を処理できない

ナイーブな方法 #2: 事前にLLMで分類

# ❌ RAGと同じくらい高コスト
classification = llm.invoke("Classify this query: " + query)

問題点：LLM呼び出しのコストがかかり、ベクトル検索の部分しか節約できない。

解決策 / 実装

Conditional Routingアーキテクチャ

解決策：パイプラインの先頭に軽量なClassifierノードを配置し、本当に必要な場合のみRAGにルーティングする。

設計判断

制約 決定 根拠
レイテンシ < 100ms 小型分類モデルを使用 分類に大規模LLMは使用不可
精度 > 95% ドメインデータでファインチューン 汎用モデルでは精度不足
コスト < $0.0001/query ローカルモデルまたは安価なAPI RAGより大幅に安くする必要あり

LangGraphでの実装

Step 1: State Schemaの定義

from typing import TypedDict, Literal

class ChatState(TypedDict):
    user_input: str
    intent_type: Literal["CHITCHAT", "QUERY", "OFF_TOPIC"]
    response: str
    context: list[str]  # 取得したドキュメント（QUERYの場合のみ）

Step 2: Intent Classifierノード

Classifierには3つのオプションがあります：

オプションA: 小型LLMでのZero-shot（MVP向け推奨）

from langchain_openai import ChatOpenAI

classifier_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

CLASSIFIER_PROMPT = """ユーザーの意図を以下のカテゴリのいずれかに分類してください：
- CHITCHAT: 挨拶、お礼、さようなら、雑談
- QUERY: 知識・情報の検索が必要な質問
- OFF_TOPIC: サービスドメインに関係ない内容

ユーザーメッセージ: {user_input}

カテゴリ名のみを回答してください。"""

def classify_intent(state: ChatState) -> ChatState:
    response = classifier_llm.invoke(
        CLASSIFIER_PROMPT.format(user_input=state["user_input"])
    )
    state["intent_type"] = response.content.strip().upper()
    return state

gpt-4o-miniのコストは約$0.00015/1K入力トークン - フルRAGパイプラインの約100分の1です。

オプションB: ファインチューン済みローカルモデル（本番環境向け）

from transformers import pipeline

# ファインチューン済みBERT分類器 - 推論約5ms、APIコストゼロ
intent_classifier = pipeline(
    "text-classification",
    model="your-org/intent-classifier-v1",
    device=0  # GPU
)

def classify_intent(state: ChatState) -> ChatState:
    result = intent_classifier(state["user_input"])[0]
    state["intent_type"] = result["label"]
    return state

オプションC: ルールベース + MLハイブリッド

import re

# 高速な正規表現プレフィルター
CHITCHAT_PATTERNS = [
    r"^(hi|hello|hey|こんにちは|おはよう|こんばんは)[\s!.]*$",
    r"^(thanks?|thank you|ありがとう|ありがとうございます).*$",
    r"^(bye|goodbye|さようなら|またね).*$",
]

def classify_intent(state: ChatState) -> ChatState:
    user_input = state["user_input"].lower().strip()
    
    # 高速パス: 正規表現マッチング
    for pattern in CHITCHAT_PATTERNS:
        if re.match(pattern, user_input, re.IGNORECASE):
            state["intent_type"] = "CHITCHAT"
            return state
    
    # 低速パス: 曖昧なケースはML分類
    result = classifier_llm.invoke(...)
    state["intent_type"] = result.content.strip().upper()
    return state

Step 3: ルーティング関数

def route_by_intent(state: ChatState) -> str:
    """分類されたintentに基づいて適切なノードにルーティング"""
    intent = state["intent_type"]
    
    # 雑談 → 直接応答パス（低コスト）
    if intent == "CHITCHAT":
        return "general_chat_node"
    
    # オフトピック → フォールバックパス（RAG不要）
    if intent == "OFF_TOPIC":
        return "fallback_node"
    
    # クエリ → フルRAGパイプライン（高コストだが必要）
    return "rag_pipeline_node"

Step 4: レスポンスノード

def general_chat_node(state: ChatState) -> ChatState:
    """シンプルなプロンプトで雑談を処理 - RAGなし"""
    response = cheap_llm.invoke(
        f"フレンドリーに応答してください: {state['user_input']}"
    )
    state["response"] = response.content
    return state

def fallback_node(state: ChatState) -> ChatState:
    """オフトピッククエリを処理"""
    state["response"] = (
        "申し訳ございません。当社の製品・サービスに関する"
        "ご質問のみサポートしております。他にお手伝いできることはありますか？"
    )
    return state

def rag_pipeline_node(state: ChatState) -> ChatState:
    """フルRAGパイプライン - 必要な場合のみ呼び出し"""
    # 高コストな処理
    docs = vector_store.similarity_search(state["user_input"], k=5)
    state["context"] = [doc.page_content for doc in docs]
    
    response = llm.invoke(
        RAG_PROMPT.format(
            context="\n".join(state["context"]),
            question=state["user_input"]
        )
    )
    state["response"] = response.content
    return state

Step 5: グラフの構築

from langgraph.graph import StateGraph, END

# グラフを構築
workflow = StateGraph(ChatState)

# ノードを追加
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("general_chat_node", general_chat_node)
workflow.add_node("fallback_node", fallback_node)
workflow.add_node("rag_pipeline_node", rag_pipeline_node)

# エントリポイントを設定
workflow.set_entry_point("classify_intent")

# 条件付きルーティングを追加
workflow.add_conditional_edges(
    "classify_intent",
    route_by_intent,
    {
        "general_chat_node": "general_chat_node",
        "fallback_node": "fallback_node",
        "rag_pipeline_node": "rag_pipeline_node",
    }
)

# すべてのパスはENDに到達
workflow.add_edge("general_chat_node", END)
workflow.add_edge("fallback_node", END)
workflow.add_edge("rag_pipeline_node", END)

# コンパイル
app = workflow.compile()

結果

コスト比較

指標 導入前（ルーティングなし） 導入後（Smart Routing） 改善率
RAG呼び出し/日 10,000 6,000 -40%
平均コスト/クエリ $0.003 $0.0012 -60%
月間コスト $900 $360 -$540/月

レイテンシの改善

導入前: ユーザー入力 → RAGパイプライン → レスポンス
        平均: 2.5秒

導入後: ユーザー入力 → Classifier (50ms) → 雑談レスポンス (200ms)
        雑談の平均: 250ms（10倍高速）
        
        ユーザー入力 → Classifier (50ms) → RAGパイプライン → レスポンス
        クエリの平均: 2.55秒（オーバーヘッドはわずか）

雑談レスポンスは2.5秒ではなく250msで返却されるようになり、ユーザー体験が大幅に向上しました。

トレードオフ

メリット ✅

  • 40-60%のコスト削減 - RAG操作の削減
  • 低レイテンシ - シンプルなクエリに対して
  • 拡張が容易 - intentカテゴリの追加が簡単
  • 可観測性の向上 - どのクエリがRAGを使用したか正確に把握

デメリット ⚠️

  • 複雑性の増加 - パイプラインが複雑になる
  • Classifierのメンテナンス - 時間とともにドリフトする可能性
  • 誤分類のリスク - QUERYがCHITCHATに分類されると誤った回答になる

リスク軽減策

# 継続的改善のために誤分類をログ
def log_classification(state: ChatState, user_feedback: str):
    if user_feedback == "wrong_answer":
        logger.warning(
            f"誤分類の可能性: "
            f"input='{state['user_input']}' "
            f"classified_as='{state['intent_type']}'"
        )
        # 再学習データセットに追加
        save_to_training_data(state, correct_label=None)

使用すべきでないケース

以下の場合はSmart Routingを適用しないでください：

  1. クエリ量が少ない（< 1000/日）- オーバーヘッドに見合わない
  2. 100%のクエリがナレッジを必要とする - 例：法律リサーチツール
  3. ラベル付きデータがない - Classifier精度の検証ができない
  4. レイテンシが重要でない - バッチ処理システム
  5. ドメインが特殊すぎる - Classifierが雑談と実際のクエリを区別できない

まとめ

Intent Filterはシンプルだが効果的なパターンです：

  • 問題: 40%のクエリはRAGを必要としないが、フルパイプラインを実行している
  • 解決策: 軽量なClassifierノードがRAGに入る前にトラフィックをルーティング
  • 結果: 40-60%のコスト削減、シンプルなクエリに対するレイテンシ改善

次のステップ:

  1. クエリログを分析して雑談/オフトピックの割合を特定
  2. gpt-4o-miniで基本的なClassifierを実装
  3. モニタリングして誤分類データを収集
  4. 十分なデータが集まったらローカルモデルをファインチューン

参考文献

5
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?