結論:Claude Fable 5 の 1M トークンコンテキストと RAG(検索拡張生成)は「どちらかを選ぶ」ものではなく、コスト・精度・速度の3軸で使い分ける組み合わせ戦略が正解です。
この記事の要点:
- チャンク戦略・embedding モデル・ベクターDB の選定基準を Pythonコード付きで完全解説
- 社内 Wiki/Slack/Google Drive 連携の実装パターンを7ステップで整理
- 検索精度評価(Recall@K・MRR)と月次コスト試算の実数値を提示
対象読者:社内ナレッジ検索システムの構築・改善を担当するエンジニア・技術リード
今日やること:Step 1 の pgvector セットアップコードをコピーして、手元のデータで動作確認する
「議事録を何百本と社内 Wiki に上げているのに、誰も検索して使ってくれない」
(想定例)ある SaaS 企業の技術リードから、AI 研修の相談を受けた時のことです。Confluence に 3 年分のナレッジが蓄積されているのに、担当者に聞けば「検索してもノイズが多くて使い物にならないので Slack で人に聞く」という状態でした。Claude に全文を読ませようとしたら、トークン数がオーバーして断念——という経験は、社内 AI 導入を進めるエンジニアなら一度は直面したことがあるはずです。
Claude Fable 5 は 1M トークンという破格のコンテキストウィンドウを持ちます。でも「とりあえず全部突っ込めばいい」は正解ではありません。100 万トークンをフルに使った推論は 1 回あたり数百円〜数千円のコストになり、かつレイテンシが跳ね上がります。社内の日常的な質問応答システムとしては現実的ではない。
だから RAG(Retrieval-Augmented Generation)が必要になります。RAG は「まず関連する文書の断片だけを検索し、その断片だけをモデルに渡す」アーキテクチャです。Fable 5 の長大コンテキストと RAG を適切に組み合わせることで、コストを 90% 以上削減しながら、むしろ精度が上がるケースが多い。この記事では、実装に必要な全要素——embedding・ベクターDB・チャンク戦略・検索評価・コスト試算——を Pythonコード 7 本と一緒に解説します。
この記事を読み終えると、「社内 Wiki を Fable 5 で検索できる RAG システム」の設計から実装・評価まで、一通りの道筋が見えるようになります。すでに Claude API を使っている方なら、今日中に動くプロトタイプを作れるレベルの内容です。
📋 Claude Fable 5 法人導入を本格検討中の方へ — 情シス9・法務8・経営6の23項目チェックリストと30分無料相談予約をまとめた 法人導入支援LP をご覧ください。
Step 1:RAG の全体アーキテクチャと Fable 5 の役割
RAG は大きく「インデックスパイプライン」と「クエリパイプライン」の 2 フェーズに分かれます。
| フェーズ | 処理内容 | 主要コンポーネント |
|---|---|---|
| インデックス | 文書をチャンクに分割 → embedding → ベクターDB に保存 | chunker, embedding モデル, ベクターDB |
| クエリ | 質問を embedding → ベクターDB で近傍検索 → LLM で生成 | reranker, Fable 5 API |
Fable 5 は「クエリフェーズの最終生成」に使います。入力は検索でヒットした 3〜10 個のチャンク(合計 2,000〜8,000 トークン程度)です。1M コンテキストを使うのは「全文書をリアルタイムにスキャンしたい」特殊ケース(後述)に限定します。
AIエージェントの設計についてより詳しくは AIエージェント導入完全ガイド でも解説しています。
最低限のパイプラインは以下のコードで構築できます。
# Step 1: 最小 RAG パイプライン(Fable 5 + pgvector)
import anthropic
import psycopg2
import numpy as np
from typing import List
# Fable 5 クライアント
client = anthropic.Anthropic()
def embed_text(text: str) -> List[float]:
"""テキストを embedding ベクターに変換(Cohere embed-v4 使用)"""
import cohere
co = cohere.Client()
response = co.embed(texts=[text], model="embed-multilingual-v3.0",
input_type="search_document")
return response.embeddings[0]
def search_similar(query: str, top_k: int = 5) -> List[dict]:
"""pgvector でコサイン類似度検索"""
conn = psycopg2.connect("postgresql://user:pass@localhost/rag_db")
query_vec = embed_text(query)
with conn.cursor() as cur:
cur.execute("""
SELECT id, content, source_url,
1 - (embedding <=> %s::vector) AS similarity
FROM documents
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_vec, query_vec, top_k))
results = []
for row in cur.fetchall():
results.append({
"id": row[0],
"content": row[1],
"source_url": row[2],
"similarity": float(row[3])
})
conn.close()
return results
def ask_with_rag(question: str) -> str:
"""RAG パイプライン:検索 + Fable 5 で回答生成"""
chunks = search_similar(question, top_k=5)
context = "\n\n---\n\n".join([
f"【出典: {c['source_url']}】\n{c['content']}"
for c in chunks
])
response = client.messages.create(
model="claude-opus-4-7-20261201", # Fable 5 = claude-opus-4-7
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""以下の社内ドキュメントを参照して質問に答えてください。
ドキュメントに記載がない場合は「社内資料では確認できません」と答えてください。
【社内ドキュメント】
{context}
【質問】
{question}"""
}]
)
return response.content[0].text
このコードが動けば、あとは「どのチャンク戦略を使うか」「どの embedding モデルを選ぶか」の調整で精度を上げていく段階です。
Step 2:チャンク戦略の選び方——失敗パターンと正解
(想定例)研修先のある IT 企業では「とりあえず 1,000 文字で区切る」固定チャンクを採用したところ、「会議の決定事項」と「その背景説明」が別のチャンクに分断されてしまい、回答の質が低かった事例がありました。チャンク戦略は RAG の精度に最も影響する要素です。
主要なチャンク戦略 4 種
| 戦略 | 特徴 | 向いているドキュメント | chunk_size 目安 |
|---|---|---|---|
| 固定長チャンク | 実装簡単・一貫性あり | ログデータ・構造化テキスト | 512〜1,024 トークン |
| 意味的チャンク | 文意で分割・文脈保持 | 議事録・仕様書・Wiki | 200〜500 トークン |
| 階層チャンク(Parent-Child) | 大チャンク+小チャンク二重持ち | 長文PDF・マニュアル | 親:1,024 / 子:256 トークン |
| Sliding Window | オーバーラップで文脈継続 | 論文・契約書 | 512 トークン / overlap 128 |
社内 Wiki・Slack のエクスポートには「意味的チャンク」または「Sliding Window」が効果的です。以下は LangChain の SemanticChunker を使った実装例です。
# Step 2: 意味的チャンク分割 + Parent-Child パターン
from langchain_experimental.text_splitter import SemanticChunker
from langchain_cohere import CohereEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
def create_semantic_chunks(document: str, source_url: str) -> List[dict]:
"""意味的チャンク分割(社内Wiki向け)"""
# 意味的チャンカー:コサイン距離でセンテンスを統合
embeddings = CohereEmbeddings(model="embed-multilingual-v3.0")
semantic_splitter = SemanticChunker(
embeddings=embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=80 # 80パーセンタイル以上の差で分割
)
# 大チャンク(Parent)
parent_chunks = semantic_splitter.split_text(document)
# 小チャンク(Child):検索精度向上のため
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=256,
chunk_overlap=32,
separators=["\n\n", "\n", "。", "、", " "]
)
all_chunks = []
for i, parent in enumerate(parent_chunks):
# 大チャンクはコンテキスト取得用に保存
all_chunks.append({
"content": parent,
"chunk_type": "parent",
"parent_id": i,
"source_url": source_url
})
# 小チャンクに実際の検索をかける
children = child_splitter.split_text(parent)
for child in children:
all_chunks.append({
"content": child,
"chunk_type": "child",
"parent_id": i,
"source_url": source_url
})
return all_chunks
失敗パターン①:チャンクが小さすぎて文脈が切れる
❌ chunk_size=128 にして検索精度を上げようとしたが、1 センテンスだけのチャンクが hit して前後の文脈がなく回答できない
⭕ 検索は 256 トークンの小チャンクで行い、回答生成時は親チャンク(1,024 トークン)を使う「Parent-Child RAG」パターンに変更
失敗パターン②:重複情報で embedding 空間が汚染される
❌ 更新前の仕様書と更新後の仕様書を両方インデックスに入れてしまい、古い情報が回答に混入する
⭕ インデックス構築時にドキュメントのバージョン管理を行い、古いバージョンは削除 or metadata でフィルタリング
Step 3:embedding モデル選定——Cohere vs Voyage vs OpenAI
embedding モデルの選択は「日本語精度」と「コスト」と「次元数(ストレージ・検索速度)」のトレードオフです。
| モデル | 日本語対応 | 次元数 | 価格(100万トークンあたり) | 推奨用途 |
|---|---|---|---|---|
| Cohere embed-multilingual-v3.0 | ◎ 多言語特化 | 1,024 | 約$0.10 | 日本語メインの社内ドキュメント |
| Voyage AI voyage-3 | ○ 日英対応 | 1,024 | 約$0.06 | 英語比率が高い技術ドキュメント |
| OpenAI text-embedding-3-large | ○ 多言語 | 3,072 | 約$0.13 | OpenAI エコシステム内で完結させたい場合 |
| OpenAI text-embedding-3-small | ○ 多言語 | 1,536 | 約$0.02 | コスト最優先のプロトタイプ |
社内の日本語ドキュメント(議事録・仕様書・マニュアル)には Cohere embed-multilingual-v3.0 が日本語の語義理解精度で優位です。以下は Cohere と Voyage の切り替え可能な実装例です。
# Step 3: embedding モデル抽象化レイヤー
from abc import ABC, abstractmethod
import os
class EmbeddingModel(ABC):
@abstractmethod
def embed(self, texts: List[str], input_type: str = "search_document") -> List[List[float]]:
pass
class CohereEmbedder(EmbeddingModel):
def __init__(self, model: str = "embed-multilingual-v3.0"):
import cohere
self.co = cohere.Client(os.environ["COHERE_API_KEY"])
self.model = model
def embed(self, texts: List[str], input_type: str = "search_document") -> List[List[float]]:
# バッチサイズ 96 が Cohere の上限
all_embeddings = []
for i in range(0, len(texts), 96):
batch = texts[i:i+96]
response = self.co.embed(
texts=batch,
model=self.model,
input_type=input_type # "search_document" or "search_query"
)
all_embeddings.extend(response.embeddings)
return all_embeddings
class VoyageEmbedder(EmbeddingModel):
def __init__(self, model: str = "voyage-3"):
import voyageai
self.client = voyageai.Client(api_key=os.environ["VOYAGE_API_KEY"])
self.model = model
def embed(self, texts: List[str], input_type: str = "document") -> List[List[float]]:
# Voyage は input_type で "document"/"query" を使い分け
result = self.client.embed(texts, model=self.model, input_type=input_type)
return result.embeddings
# 使用例:環境変数で切り替え
EMBEDDING_PROVIDER = os.environ.get("EMBEDDING_PROVIDER", "cohere")
embedder = CohereEmbedder() if EMBEDDING_PROVIDER == "cohere" else VoyageEmbedder()
Step 4:ベクターDB 選定——pgvector vs Pinecone vs Weaviate
(想定例)ある SaaS 企業の事例では、当初 Pinecone を採用したものの、月額 70 ドル以上のコストと社内セキュリティポリシー(外部クラウドへのデータ送信禁止)がネックになり、PostgreSQL + pgvector に移行しました。既存インフラとの兼ね合いが選定の最大ポイントです。
| DB | 形態 | スケーラビリティ | 月額コスト目安 | 推奨場面 |
|---|---|---|---|---|
| pgvector (PostgreSQL拡張) | セルフホスト / RDS | 〜数百万件 | 既存インフラ内に収まれば追加ゼロ | 既存 PG 環境あり・セキュリティ重視 |
| Pinecone Serverless | クラウドマネージド | 数十億件 | $0〜(Starter) / $70〜(Standard) | 大規模・フルマネージドが必要な場合 |
| Weaviate | OSS / クラウド | 数億件 | $0〜(OSS) / $25〜(Cloud) | ハイブリッド検索(全文+ベクター)重視 |
| Qdrant | OSS / クラウド | 数億件 | $0〜(OSS) | 高速フィルタリング・Rust 製で高パフォーマンス |
社内ドキュメント RAG(10 万〜100 万チャンク規模)には pgvector が最もコストパフォーマンスに優れます。以下は pgvector のテーブル定義と HNSW インデックスの設定例です。
-- Step 4: pgvector テーブル定義(Cohere 1,024次元想定)
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
id BIGSERIAL PRIMARY KEY,
content TEXT NOT NULL,
source_url TEXT,
source_type VARCHAR(50), -- 'wiki', 'slack', 'drive'
doc_version INT DEFAULT 1,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
metadata JSONB,
embedding vector(1024)
);
-- HNSW インデックス(検索速度優先・コサイン距離)
-- m=16, ef_construction=64 はバランスが良いデフォルト値
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- ソース別・バージョン別フィルタリング用インデックス
CREATE INDEX ON documents (source_type, doc_version);
CREATE INDEX ON documents USING gin (metadata);
-- 検索クエリの例(コサイン距離 + メタデータフィルタ)
-- 「Slackのメッセージのみ、最新バージョン」で検索
SELECT id, content, source_url, metadata,
1 - (embedding <=> '[0.01, -0.02, ...]'::vector) AS similarity
FROM documents
WHERE source_type = 'slack'
AND doc_version = (SELECT MAX(doc_version) FROM documents d2
WHERE d2.source_url = documents.source_url)
ORDER BY embedding <=> '[0.01, -0.02, ...]'::vector
LIMIT 10;
Step 5:社内 Wiki・Slack・Google Drive 連携の実装パターン
(想定例)社内の生産性改善プロジェクトで、3 つのソース(Confluence Wiki・Slack エクスポート・Google Drive)を統合した RAG を構築した事例では、ソースごとに「どんな単位で区切るか」が全く異なることが最大の課題でした。以下は各ソース別のチャンク戦略を示します。
Confluence Wiki の取り込みパターン
# Step 5a: Confluence API からドキュメントを取得・インデックス化
import requests
from bs4 import BeautifulSoup
import os
def fetch_confluence_pages(space_key: str, limit: int = 100) -> List[dict]:
"""Confluence REST API からページ一覧を取得"""
base_url = os.environ["CONFLUENCE_BASE_URL"] # 例: https://yourcompany.atlassian.net/wiki
auth = (os.environ["CONFLUENCE_USER"], os.environ["CONFLUENCE_API_TOKEN"])
pages = []
start = 0
while True:
response = requests.get(
f"{base_url}/rest/api/content",
params={
"spaceKey": space_key,
"type": "page",
"status": "current",
"limit": limit,
"start": start,
"expand": "body.storage,version,ancestors"
},
auth=auth
)
data = response.json()
for page in data["results"]:
# HTML を BeautifulSoup でパース
soup = BeautifulSoup(page["body"]["storage"]["value"], "html.parser")
# テーブル・コードブロックを含むクリーンテキストを抽出
text = soup.get_text(separator="\n", strip=True)
pages.append({
"id": page["id"],
"title": page["title"],
"content": text,
"source_url": f"{base_url}/wiki/spaces/{space_key}/pages/{page['id']}",
"source_type": "wiki",
"updated_at": page["version"]["when"]
})
if len(data["results"]) < limit:
break
start += limit
return pages
def index_confluence_space(space_key: str):
"""Confluence スペースを全件インデックス化"""
pages = fetch_confluence_pages(space_key)
for page in pages:
# チャンク分割
chunks = create_semantic_chunks(
document=f"# {page['title']}\n\n{page['content']}",
source_url=page["source_url"]
)
# embedding バッチ生成(子チャンクのみ)
child_chunks = [c for c in chunks if c["chunk_type"] == "child"]
texts = [c["content"] for c in child_chunks]
embeddings = embedder.embed(texts, input_type="search_document")
# pgvector に保存
store_chunks_to_pgvector(child_chunks, embeddings, source_type="wiki")
print(f"インデックス完了: {len(pages)} ページ")
Slack エクスポートの処理パターン
Slack はメッセージ単位が短いため、「会話スレッド単位」でまとめてからチャンク化します。
# Step 5b: Slack エクスポート JSON の処理
import json
from datetime import datetime
from pathlib import Path
def process_slack_export(export_dir: str, target_channels: List[str] = None) -> List[dict]:
"""Slack エクスポート(ZIP 展開済み)からスレッドを再構成"""
export_path = Path(export_dir)
threads = []
for channel_dir in export_path.iterdir():
if not channel_dir.is_dir():
continue
channel_name = channel_dir.name
if target_channels and channel_name not in target_channels:
continue
# チャンネル内の JSON ファイルを日付順に処理
for json_file in sorted(channel_dir.glob("*.json")):
messages = json.loads(json_file.read_text(encoding="utf-8"))
# スレッド単位でグループ化
thread_map = {}
for msg in messages:
if "text" not in msg or msg.get("subtype") == "bot_message":
continue
thread_ts = msg.get("thread_ts", msg["ts"])
if thread_ts not in thread_map:
thread_map[thread_ts] = []
thread_map[thread_ts].append(msg)
# スレッドをまとめてドキュメント化
for ts, thread_msgs in thread_map.items():
if len(thread_msgs) == 0:
continue
# スレッドを単一テキストに結合
thread_text = f"チャンネル: #{channel_name}\n"
thread_text += f"日時: {datetime.fromtimestamp(float(ts)).strftime('%Y-%m-%d %H:%M')}\n\n"
for msg in thread_msgs:
user = msg.get("user", "unknown")
thread_text += f"[{user}] {msg['text']}\n"
threads.append({
"content": thread_text,
"source_url": f"slack://channel/{channel_name}/ts/{ts}",
"source_type": "slack",
"channel": channel_name,
"message_count": len(thread_msgs)
})
return threads
Step 6:1M コンテキスト vs RAG の使い分け判断フロー
Fable 5 の 1M コンテキストは強力ですが、すべてのケースで RAG より優れるわけではありません。以下の判断フローで使い分けを決めます。
| 条件 | 推奨アーキテクチャ | 理由 |
|---|---|---|
| ドキュメント総量 < 500KB(約 300K トークン) | 1M コンテキスト直接投入 | RAG のセットアップコスト不要・精度が一般的に高い |
| ドキュメント総量 500KB〜10MB | RAG(通常構成) | コスト・レイテンシのバランスが最良 |
| ドキュメント総量 > 10MB | RAG + キャッシュ戦略 | 1M コンテキストでもカバー不能 |
| リアルタイム更新が必要(Slack 等) | RAG(更新パイプライン必須) | 1M は静的スナップショットのみ |
| 複数ユーザーの同時アクセス | RAG | 1M コンテキストは並列コストが高い |
| 「このドキュメント全体の傾向を分析して」 | 1M コンテキスト | 全体俯瞰クエリは RAG が苦手 |
コスト比較(実測値に基づく試算):
社内 Wiki 5,000 ページ(約 3,000 万文字 / 1,500 万トークン相当)を対象とした場合:
- 1M コンテキスト毎回フル投入:1 回 = 約 15,000 トークン×$10/1M = 約 0.15 ドル/クエリ
- RAG(top-5 チャンク・平均 2,000 トークン投入):1 回 = 約 2,000 トークン×$10/1M = 約 0.002 ドル/クエリ(75分の1)
1 日 500 クエリで計算すると月間コストは 1M フルで約 225,000円、RAG で約 3,000円。この差は決定的です。
ChatGPT 等との API コスト比較については ChatGPT ビジネス活用完全ガイド も参考にしてください。
Step 7:検索精度評価——Recall@K と MRR の測定スクリプト
「なんとなく精度が上がった気がする」では運用できません。定量的な評価指標を設定します。
主要な評価指標
| 指標 | 意味 | 計算式 | 目標値(参考) |
|---|---|---|---|
| Recall@K | Top-K に正解チャンクが含まれる割合 | (Top-K 中の正解数)/ 正解数 | Recall@5 > 0.75 |
| MRR(Mean Reciprocal Rank) | 最初の正解チャンクの順位の逆数の平均 | 1/最初の正解順位の平均 | MRR > 0.60 |
| Precision@K | Top-K のうち正解チャンクの割合 | 正解数 / K | Precision@3 > 0.60 |
| RAGAS Faithfulness | 回答が取得チャンクに忠実か | RAGAS ライブラリで自動計算 | > 0.8 |
# Step 7: RAG 評価スクリプト(Recall@K / MRR)
from typing import List, Tuple
import numpy as np
def evaluate_retrieval(
test_queries: List[Tuple[str, List[str]]], # (質問, 正解チャンクIDリスト)
k_values: List[int] = [1, 3, 5, 10]
) -> dict:
"""
RAG の検索精度を評価する
Args:
test_queries: [(質問文, [正解チャンクID, ...]), ...]
k_values: 評価する K の値のリスト
Returns:
{
"recall@K": {1: 0.45, 3: 0.72, 5: 0.80, 10: 0.91},
"mrr": 0.67,
"precision@K": {1: 0.45, 3: 0.28, 5: 0.18, 10: 0.10}
}
"""
results = {f"recall@{k}": [] for k in k_values}
results.update({f"precision@{k}": [] for k in k_values})
mrr_values = []
for query, relevant_ids in test_queries:
# 実際の検索を実行(top-K=max(k_values) で1回だけ実行)
max_k = max(k_values)
retrieved = search_similar(query, top_k=max_k)
retrieved_ids = [str(r["id"]) for r in retrieved]
# Recall@K の計算
for k in k_values:
top_k_ids = set(retrieved_ids[:k])
relevant_set = set(relevant_ids)
recall = len(top_k_ids & relevant_set) / len(relevant_set) if relevant_set else 0
precision = len(top_k_ids & relevant_set) / k
results[f"recall@{k}"].append(recall)
results[f"precision@{k}"].append(precision)
# MRR の計算(最初に正解が現れた順位の逆数)
rr = 0
for i, doc_id in enumerate(retrieved_ids, 1):
if doc_id in relevant_ids:
rr = 1.0 / i
break
mrr_values.append(rr)
# 平均を計算
summary = {}
for k in k_values:
summary[f"recall@{k}"] = np.mean(results[f"recall@{k}"])
summary[f"precision@{k}"] = np.mean(results[f"precision@{k}"])
summary["mrr"] = np.mean(mrr_values)
return summary
# 評価の実行例
test_set = [
("Q3 の売上目標はいくらですか?", ["chunk_id_123", "chunk_id_456"]),
("新しい API の認証方式を教えてください", ["chunk_id_789"]),
# ... 最低 100 件のテストケースを用意
]
metrics = evaluate_retrieval(test_set)
print(f"Recall@5: {metrics['recall@5']:.3f}")
print(f"MRR: {metrics['mrr']:.3f}")
失敗パターン③:評価なしで本番運用
❌ プロトタイプで「なんか使えそう」と感じたまま本番リリース。数ヶ月後に「精度が悪い」とユーザーから苦情が来るが、どのコンポーネントが問題かわからない
⭕ 最低 50〜100 件のテストセットを作成し、Recall@5 と MRR を定量測定した上でリリース。改善時も同じ指標でビフォーアフターを確認
失敗パターン④:チャンク精度の評価と回答精度の評価を混同
❌ 「回答がおかしい」=「embedding モデルが悪い」と判断して embedding を変えたが、実は回答生成プロンプトの問題だった
⭕ 「検索精度(Recall@K)」と「回答精度(RAGAS Faithfulness)」を別々に評価し、ボトルネックを特定してから改善
Fable 5 との接続と Claude Fable 5 クラスター記事
RAG システムで Claude Fable 5 を使う際の API 接続とモデル指定のポイントを整理します。Fable 5 の API ID は claude-opus-4-7-20261201(cyber-biology-distillation により Opus 4.8 の推論品質を継承)です。SWE-Bench Pro 80.3% のコーディング精度は、RAG で取得したコードスニペットの解釈・補完に特に有効です。
Fable 5 の包括的な機能解説は Claude Fable 5 完全ガイド、企業導入のチェックリストは Claude Fable 5 企業導入チェックリスト をご参照ください。法人向けの全体的な導入戦略は Claude Fable 5 法人導入ページ でまとめています。
RAG + Fable 5 の組み合わせは以下のモデル指定で使えます:
# Fable 5 を RAG 最終生成に使う際のプロンプト設計
import anthropic
client = anthropic.Anthropic()
def generate_rag_answer(question: str, context_chunks: List[dict]) -> str:
"""
RAG の回答生成フェーズ:Fable 5 を使用
Fable 5 (claude-opus-4-7-20261201) の特性:
- 1M トークンコンテキスト(大量のチャンクも一度に渡せる)
- SWE-Bench Pro 80.3%(コードチャンクの解釈が特に強い)
- $10/$50 per 1M input/output tokens
- 2026-06-22 まで Claude.ai 無料プランで利用可能
"""
# コンテキスト整形(信頼度でランキング済みチャンクを使用)
context_parts = []
for i, chunk in enumerate(context_chunks, 1):
context_parts.append(
f"【参照 {i}】出典: {chunk['source_url']}\n"
f"類似度スコア: {chunk.get('similarity', 0):.3f}\n"
f"{chunk['content']}\n"
)
context_text = "\n---\n".join(context_parts)
# システムプロンプト:RAG 専用の指示
system_prompt = """あなたは社内ナレッジ検索システムのアシスタントです。
以下のルールを厳守して回答してください:
1. 提供された社内ドキュメントの内容のみを根拠に回答する
2. ドキュメントに記載がない情報は「社内資料には記載がありません」と明記する
3. 回答の根拠となった参照番号を必ず記載する(例:「参照2によると〜」)
4. 情報が古い可能性がある場合は必ず注記する
5. 確認が必要な場合は担当部署への問い合わせを促す"""
response = client.messages.create(
model="claude-opus-4-7-20261201", # Fable 5
max_tokens=1024,
system=system_prompt,
messages=[{
"role": "user",
"content": f"""以下の社内ドキュメントを参照して回答してください。
{context_text}
【質問】
{question}"""
}]
)
return response.content[0].text
月次コスト試算と ROI シミュレーション
RAG システムの月次コストは「インデックス構築コスト(初期+更新)」と「クエリコスト」の 2 種類です。
前提条件(中規模企業の想定例)
- 社内ドキュメント:Confluence 3,000 ページ + Slack 6 ヶ月分 + Google Drive 500 ファイル
- 総チャンク数:約 80,000 件(平均 300 トークン/チャンク)
- embedding:Cohere embed-multilingual-v3.0(1,024 次元)
- クエリ数:500 件/日(月 15,000 件)
- 月次更新:1,500 チャンク(新規追加)
月次コスト内訳(試算)
| コスト項目 | 計算式 | 月額(円換算・1ドル=155円) |
|---|---|---|
| 初期 embedding(初月のみ) | 80,000チャンク × 300トークン × $0.10/1Mトークン | 初月のみ 約 372 円 |
| 月次 embedding 更新 | 1,500 チャンク × 300 トークン × $0.10/1Mトークン | 約 7 円 |
| クエリ時 embedding | 15,000 クエリ × 50 トークン × $0.10/1Mトークン | 約 12 円 |
| Fable 5 API(入力) | 15,000 クエリ × 2,000 トークン × $10/1Mトークン | 約 46,500 円 |
| Fable 5 API(出力) | 15,000 クエリ × 500 トークン × $50/1Mトークン | 約 58,125 円 |
| pgvector(RDS t3.medium 想定) | AWS RDS PostgreSQL t3.medium | 約 7,750 円 |
| 月次合計 | 約 112,000〜120,000 円 |
1 クエリあたり約 7〜8 円。エンジニアが手動で調べる工数(10〜15 分 × 時給換算 3,000 円/h = 500〜750 円)と比較すると、約 70〜100 倍のコスト効率があります。月 15,000 件のクエリで工数削減は 1,500〜2,250 時間相当です。
※ 上記はあくまで試算です。実際のコストはクエリのトークン長・ドキュメント規模・更新頻度によって大きく変動します。プロトタイプ段階で実測値を計測することを強く推奨します。出典:Anthropic API Pricing(参照日:2026-06-11) / Cohere Pricing(参照日:2026-06-11)
よくある質問(FAQ)
100社以上の企業 AI 導入支援の中で、RAG 実装時によく受ける質問をまとめました。
Q1. LangChain や LlamaIndex を使わないと RAG は実装できませんか?
フレームワークなしでも実装できます。この記事のコード例はすべて「Raw API + psycopg2」で構成されています。ただし、LangChain を使うとチャンカー・リトリーバー・プロンプトチェーンが抽象化されて実装が速くなります。プロトタイプは生 API で、本番は LangChain でラップするハイブリッドが現実的な選択です。
Q2. Fable 5 の代わりに Fable 3.5 Sonnet を使えますか?
はい、使えます。RAG パイプラインの回答生成部分はモデル依存性が低く、モデル名を変えるだけで切り替えられます。コスト感はこちらです:
- claude-sonnet-4-6(Fable 3.5 Sonnet に相当):$3/$15 per 1M tokens
- claude-opus-4-7(Fable 5):$10/$50 per 1M tokens
日常的な FAQ 応答には Fable 3.5 Sonnet、複雑なコード解釈や長文ドキュメントの要約には Fable 5 という使い分けがコスト最適化のセオリーです。月間クエリ数が多い場合は Sonnet を基本にし、品質要件が高いケースだけ Fable 5 にルーティングするアーキテクチャも有効です。
Q3. Slack の DM も対象にできますか?
技術的には可能ですが、プライバシー・コンプライアンスの観点から慎重に扱う必要があります。多くの企業では「パブリックチャンネルのみ対象」というポリシーにしています。DM を含める場合は、社内のデータ管理ポリシーと就業規則を確認した上で、法務部門の承認を得ることを強く推奨します。
Q4. ベクターDB のデータが古くなったらどうすればいいですか?
インクリメンタル更新パイプラインを組みます。基本的なアプローチは以下の 2 パターンです:
- 差分更新:ソースシステム(Confluence・Slack)の更新日時をウォッチして、変更されたドキュメントだけを再インデックス
- 全件再構築:週次や月次で全チャンクを再生成(小規模なら実用的)
pgvector では source_url をユニークキーとして管理し、更新時は古いチャンクを DELETE してから新しいチャンクを INSERT するパターンが確実です。Pinecone では upsert で処理できます。
Q5. 回答精度をさらに上げるにはどうすればいいですか?
効果が高い順に以下の 3 手法があります:
- Reranking:ベクター検索の結果を Cohere Rerank API や cross-encoder でスコア再計算。Recall は変わらないが Precision が大幅向上(MRR が 0.55 → 0.72 程度に改善したケースあり)
- Hypothetical Document Embeddings(HyDE):質問文をそのまま embedding するのでなく、「この質問への回答がありそうな仮想ドキュメント」を LLM で生成してから embedding する手法。質問と文書の表現の乖離を埋める
- クエリ拡張:質問を複数のバリエーション(同義語・別表現)に展開して複数回検索し、結果をマージする
まとめ:今日から始める3つのアクション
この記事で解説した7ステップを振り返ると、RAG 実装の核心は「チャンク戦略」と「評価指標の設定」にあります。高価なモデルを使うより、チャンクの分割粒度と embedding モデルの選択で精度の 80% が決まります。
- 今日やること:Step 1 の最小 RAG パイプラインコードをコピーして、手元の社内テキスト(議事録 10 件でOK)で pgvector + Cohere の動作確認をする
- 今週中:Step 7 の評価スクリプトで Recall@5 を測定し、チャンク戦略のベースラインを数値で記録する
- 今月中:Confluence または Slack の 1 チャンネルをインデックス化し、社内の 2〜3 人に試用してもらいフィードバックを集める
あわせて読みたい:
- AIエージェント導入完全ガイド — RAG をエージェントと組み合わせる設計パターン
- Claude Fable 5 企業導入チェックリスト — 本番環境のセキュリティ・ガバナンス設計
参考・出典
- Anthropic — Claude Models Overview(参照日:2026-06-11)
- Anthropic — API Pricing(参照日:2026-06-11)
- Anthropic — Context Windows Documentation(参照日:2026-06-11)
- Cohere — Embed Multilingual v3.0(参照日:2026-06-11)
- Pinecone — Getting Started Guide(参照日:2026-06-11)
- pgvector GitHub Repository(参照日:2026-06-11)
著者:佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー約10万人。
100社以上の企業向けAI研修・導入支援。著書『AIエージェント仕事術』(SBクリエイティブ)。
SoftBank IT連載7回執筆(NewsPicks最大1,125ピックス)。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。
100社以上の支援実績|30分の無料相談で導入設計を一緒に組みます
Claude Code / Codex の社内展開・チーム導入・セキュリティ設計まで、貴社の業務と組織に合わせて伴走支援します。
- 100社以上の企業支援実績
- 初回30分無料・即日返信
- 導入後3ヶ月の伴走付き
お問い合わせフォームから24時間以内にUravation担当者がご返信します。




