Embedding APIのコストが月10万円に達していた個人プロジェクトを、アーキテクチャの見直しで月3000円まで削減した。本記事ではその具体的な手法を、Before/Afterのコスト比較とともに解説する。
Before: 月10万円の内訳
問題のプロジェクトは、社内ドキュメント検索のRAGシステムだった。
- ドキュメント数: 約50,000件
- 1日あたりの検索クエリ: 約2,000件
- ドキュメント更新頻度: 1日約500件
コストの内訳はこうだった。
| 項目 | 月額コスト（概算） |
|---|---|
| ドキュメントの再Embedding（日次全件） | 約70,000円 |
| クエリのEmbedding | 約15,000円 |
| ベクトルDB（Pinecone） | 約15,000円 |
| 合計 | 約100,000円 |
最大のコスト要因は「ドキュメントの再Embedding」だった。ドキュメント更新のたびに全件を再Embeddingしていたのが原因である。
最適化1: 差分Embeddingの導入
全件再Embeddingをやめ、変更があったドキュメントのみを再Embeddingする。
import hashlib
def compute_content_hash(content: str) -> str:
return hashlib.sha256(content.encode()).hexdigest()
def update_embeddings(documents: list[dict]):
for doc in documents:
new_hash = compute_content_hash(doc["content"])
stored_hash = db.get_hash(doc["id"])
if new_hash == stored_hash:
continue # 変更なし: スキップ
embedding = get_embedding(doc["content"])
vector_db.upsert(
id=doc["id"],
vector=embedding,
metadata={"hash": new_hash, "title": doc["title"]}
)
db.update_hash(doc["id"], new_hash)
これだけで、1日あたりのEmbedding対象が50,000件から500件に減少した。
効果: 月70,000円 → 月700円
最適化2: チャンク戦略の見直し
当初は1ドキュメントを500トークンごとに機械的に分割していたが、これにより1ドキュメントあたり平均8チャンクが生成され、トークン消費量が膨れていた。
改善策として、セクション単位での分割に切り替えた。
import re
def smart_chunking(content: str, max_tokens: int = 1000) -> list[str]:
sections = re.split(r'\n#{1,3}\s', content)
chunks = []
current_chunk = ""
for section in sections:
if estimate_tokens(current_chunk + section) > max_tokens:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = section
else:
current_chunk += "\n" + section
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def estimate_tokens(text: str) -> int:
return len(text) // 3 # 日本語の大まかな概算
効果: 平均チャンク数が8から3に減少。Embedding対象のトークン数が約60%削減。
最適化3: Embeddingモデルの変更
OpenAIの text-embedding-ada-002 から text-embedding-3-small に切り替えた。
| モデル | 次元数 | 1Mトークンあたりの料金 |
|---|---|---|
| text-embedding-ada-002 | 1536 | $0.10 |
| text-embedding-3-small | 1536 | $0.02 |
料金が1/5になった。検索精度についても、このプロジェクトのユースケースでは体感できる差はなかった。
効果: API単価が80%削減。
最適化4: クエリのEmbeddingキャッシュ
同じ検索クエリが繰り返し発生する傾向があったため、Embeddingの結果をキャッシュする。
import json
import hashlib
class EmbeddingCache:
def __init__(self, cache_file: str = "embedding_cache.json"):
self.cache_file = cache_file
try:
with open(cache_file, 'r') as f:
self.cache = json.load(f)
except FileNotFoundError:
self.cache = {}
def get_embedding(self, text: str) -> list[float]:
key = hashlib.md5(text.encode()).hexdigest()
if key in self.cache:
return self.cache[key]
embedding = call_embedding_api(text)
self.cache[key] = embedding
if len(self.cache) % 100 == 0:
self._save()
return embedding
def _save(self):
with open(self.cache_file, 'w') as f:
json.dump(self.cache, f)
実測で、クエリの約40%がキャッシュヒットした。
効果: クエリEmbeddingコストが40%削減。
最適化5: ベクトルDBの見直し
Pinecone（マネージド）からローカルで動くChromaDBに切り替えた。50,000件規模であればローカルDBで十分なパフォーマンスが出る。
import chromadb
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection("documents")
def search(query_embedding: list[float], top_k: int = 10):
results = collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
return results
効果: ベクトルDB月額15,000円 → 0円（自前サーバーのリソースのみ）
After: 月3000円の内訳
| 項目 | 最適化前 | 最適化後 |
|---|---|---|
| ドキュメントEmbedding | 70,000円 | 700円 |
| クエリEmbedding | 15,000円 | 1,800円 |
| ベクトルDB | 15,000円 | 0円 |
| サーバーリソース増分 | 0円 | 500円 |
| 合計 | 100,000円 | 3,000円 |
注意点
- ローカルDBへの移行はスケールの限界がある。数百万件を超えるならマネージドDBが必要
- キャッシュのサイズが膨らむと、ディスクとメモリの管理が必要になる
- モデル変更時は検索精度のベンチマークを必ず取ること
- 差分Embeddingは「変更検知の正確性」に依存するため、ハッシュ計算の対象を適切に選ぶ必要がある
まとめ
Embedding APIのコスト最適化は、大きな技術的チャレンジではない。差分Embedding、チャンク戦略の見直し、モデル変更、キャッシュ、DB見直しという5つの施策の組み合わせで、月10万円を月3000円に削減できた。特に差分Embeddingの効果が最も大きく、これだけで全体コストの7割を削減できた。