renue

ARTICLE

MySQL BLOB+text-embedding-3-small で作る RAG 基盤の実装ガイド【2026年版】— 専用ベクトルDB不要のセマンティック検索アーキテクチャ

公開日: 2026/4/6

ChatGPT/Claude以降、社内ドキュメント・議事録・ガイドラインをLLMに与えるRAG(Retrieval-Augmented Generation)の実装ニーズが急増している。しかし多くの実装は「pgvectorを使う」「FAISSを使う」といった専用ベクトルDB前提で語られ、既存のMySQL/PostgreSQL運用にRAGを組み込みたい現場のニーズに応えていない。本記事ではrenueが自社で運用しているRAG基盤の実装コードをもとに、OpenAI text-embedding-3-small + MySQL BLOBカラム + Python struct.packという「追加インフラゼロ」のRAGアーキテクチャを解説する。

なぜ専用ベクトルDBを使わないのか

RAGチュートリアルの多くはpgvector/Pinecone/Weaviate/Chroma等の専用ベクトルDBを前提とするが、実際の中小規模RAG(〜10万件)では以下の理由で既存DBへの埋め込みが現実解となる。

  • 運用コスト: 新規DBの導入は監視・バックアップ・アップグレードを増やす
  • データ整合性: 本体データ(ガイドライン/議事録)と埋め込みを別DBで管理すると削除・更新の整合性が崩れる
  • トランザクション: 本体レコードと埋め込みを同一トランザクションで書き込みたい
  • スケール閾値: 10万件程度ならPythonでの総当たりcosine計算でも数百ms

renueでは会社ガイドライン(110件超)・議事録(数千件)・プロジェクトメモ等を既存MySQLにVARBINARYカラムで格納し、Python側でcosine計算している。この設計が1536次元×float32 = 6144バイト/行で済み、専用DB不要で十分な性能を出している。

レイヤー1: Embedding生成(text-embedding-3-small)

OpenAIのtext-embedding-3-smallは1536次元・$0.02/1M tokensで現時点で最もコスパが良い。L2正規化済み(単位ベクトル)なのでcosine類似度はドット積だけで計算できる。

from litellm import embedding

EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIMENSION = 1536

def generate_embedding_sync(text: str) -> List[float]:
    response = embedding(model=EMBEDDING_MODEL, input=[text])
    return response.data[0]["embedding"]

注目点は litellm 経由で呼んでいる点。OpenAI SDK直叩きではなくLiteLLMを噛ませることで、後でAzure OpenAI/Vertex AI等に切り替え可能なベンダーロックイン回避レイヤーを確保している。

バッチ生成の実装

def batch_generate_embeddings_sync(texts: List[str]) -> List[List[float]]:
    # OpenAI APIは一度に最大2048個まで処理可能
    batch_size = 100
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        response = embedding(model=EMBEDDING_MODEL, input=batch)
        all_embeddings.extend([item["embedding"] for item in response.data])
    return all_embeddings

1回のAPIコールで複数テキストをまとめて送ることでレイテンシとコストを10倍以上削減できる。バッチサイズ100は「API上限2048」と「1リクエストのタイムアウトリスク」のバランス値。

レイヤー2: MySQL BLOBへの保存(struct.pack)

埋め込みをDBに保存する際、多くのチュートリアルはJSON.stringifyでテキスト化して保存する。これは6倍以上サイズが膨らむかつ読み込み時にparse負荷がかかる。

import struct

def embedding_to_bytes(embedding_data: List[float]) -> bytes:
    # float32で1536次元 = 6144バイト
    return struct.pack(f"{len(embedding_data)}f", *embedding_data)

def bytes_to_embedding(data: bytes) -> List[float]:
    count = len(data) // 4  # float32 = 4 bytes
    return list(struct.unpack(f"{count}f", data))

サイズ比較

保存方式1行あたりサイズ10万件の総サイズ読み込みコスト
JSON text約35,000 bytes約3.5 GBJSON.parse必要
struct.pack (float32)6,144 bytes約614 MBunpack一発
float16量子化3,072 bytes約307 MB精度低下あり

MySQLではVARBINARY(6144)またはBLOBカラムを使えば十分。PostgreSQLならbytea。SQLAlchemyモデル側ではColumn(LargeBinary)で受け取る。

レイヤー3: コサイン類似度検索

embedding-3-smallは正規化済みなので、厳密にはドット積で良いのだが、保守性のため明示的にcosine計算しておく。

import numpy as np

def cosine_similarity(a: List[float], b: List[float]) -> float:
    a_np = np.array(a)
    b_np = np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

def find_similar_embeddings(
    query_embedding: List[float],
    embeddings: List[Tuple[int, bytes]],
    threshold: float = 0.7,
    limit: int = 10,
) -> List[Tuple[int, float]]:
    results = []
    for item_id, embedding_bytes in embeddings:
        if embedding_bytes is None:
            continue
        item_embedding = bytes_to_embedding(embedding_bytes)
        similarity = cosine_similarity(query_embedding, item_embedding)
        if similarity >= threshold:
            results.append((item_id, similarity))
    results.sort(key=lambda x: x[1], reverse=True)
    return results[:limit]

性能特性

10万件でPython単純ループだと数秒かかるが、NumPyベクトル化で数十msに短縮できる。具体的には一括デシリアライズ後にnp.dot(matrix, query)を一発で計算する。

# 高速化版(NumPyベクトル化)
def find_similar_fast(query_embedding, embeddings_list):
    # (N, 1536) の行列を一度に作る
    matrix = np.array([bytes_to_embedding(b) for _, b in embeddings_list])
    query = np.array(query_embedding)
    # 正規化済みならnp.dotだけで類似度
    similarities = matrix @ query
    top_indices = np.argsort(-similarities)[:10]
    return [(embeddings_list[i][0], similarities[i]) for i in top_indices]

レイヤー4: 検索対象テキストの整形(最重要)

RAG実装で最も精度を左右するのが「何を埋め込むか」だ。生のcontent丸ごとは長すぎてノイズを含む。renueではtitle + summary + keywordsの3要素に絞って埋め込む設計を採用している。

def prepare_guideline_text_for_embedding(
    title: str,
    content: str,
    summary: Optional[str] = None,
    keywords: Optional[List[str]] = None,
) -> str:
    parts = [title]
    if summary:
        parts.append(summary)
    else:
        # summaryがない場合はcontentの先頭200文字を使用
        parts.append(content[:200])
    if keywords:
        parts.append(" ".join(keywords))
    return "
".join(parts)

なぜcontent全文を埋め込まないのか

  • Context dilution問題: 長文を埋め込むと主要概念が希釈されて類似度が低下する
  • トークンコスト: embedding-3-smallは$0.02/1M tokenだが全文だと数倍になる
  • chunk境界問題: chunking前提だと「どこで切るか」の設計が複雑化する
  • title/summaryの情報密度: 執筆者が要約時点で情報を圧縮済み — 検索軸としてはこれで十分

議事録版の実装

def prepare_meeting_text_for_embedding(
    title: Optional[str] = None,
    summarized_text: Optional[str] = None,
    memo: Optional[str] = None,
    company_department_name: Optional[str] = None,
) -> str:
    parts = []
    if title:
        parts.append(title)
    if company_department_name:
        parts.append(company_department_name)
    if memo:
        parts.append(memo[:300])
    if summarized_text:
        summary = _extract_json_summary(summarized_text, max_chars=500)
        if summary:
            parts.append(summary)
    return "
".join(parts)

議事録は「会議タイトル + プロジェクト名 + アジェンダ冒頭300字 + LLM要約500字」の順で結合している。プロジェクト名を含めることでテナント/案件を跨いだ混同を防ぐのが実装上の勘所。

レイヤー5: LLMのJSON要約を安全に取り出す

議事録要約はLLMが生成したJSON文字列になっていることが多い。生JSONをそのまま埋め込むと{"summary":のような構造ノイズが混入するため、要約フィールドだけ安全に抽出する必要がある。

def _extract_json_summary(raw: Optional[str], max_chars: int = 500) -> str:
    # JSONテキストからsummaryフィールドを抽出、失敗したらプレーンテキストとして截断
    if not raw or not isinstance(raw, str):
        return ""
    raw = raw.strip()
    if not raw:
        return ""
    try:
        import json
        parsed = json.loads(raw)
        if isinstance(parsed, dict):
            summary = parsed.get("summary", "")
            if isinstance(summary, str) and summary.strip():
                return summary.strip()[:max_chars]
    except (ValueError, TypeError):
        pass
    return raw[:max_chars]

ポイントは「JSON parseに失敗してもプレーンテキストとして截断するフォールバック」。LLM出力の揺らぎに対する防御としてこのパターンは他のRAG実装にも転用できる。

スケール戦略: いつ専用ベクトルDBに移行するか

データ件数推奨構成検索レイテンシ(P95)
〜1万件MySQL BLOB + Python NumPy50-150ms
1-10万件MySQL BLOB + NumPy一括計算100-500ms
10-100万件pgvector + HNSW index10-50ms
100万件超Pinecone/Qdrant/Weaviate5-30ms

10万件までは今回の構成で十分戦える。SaaS向けの社内ドキュメント検索/ガイドライン検索/議事録検索は多くの場合この範囲に収まる。

まとめ: 追加インフラゼロでRAGを始める

  1. litellm + text-embedding-3-smallで1536次元ベクトルを生成
  2. struct.pack(float32)でMySQL BLOBに6KB/行で格納
  3. title + summary + keywordsの3要素に絞って埋め込む(content丸ごとはNG)
  4. NumPy一括計算で10万件を数十msで検索
  5. LLMのJSON出力は安全にsummary抽出してフォールバック

この構成は本番運用中で、110件超の社内ガイドラインと数千件の議事録から瞬時に関連情報をRAG取得できている。専用ベクトルDB導入の前に、まずはこの「既存DB + BLOB + Python」構成を試してから判断してほしい。

よくある質問

Q. text-embedding-3-largeではダメですか?

smallは3072次元・$0.13/1M tokenで精度は確かに高いが、社内ドキュメント検索の精度向上は誤差レベル。6倍のストレージと6倍のコストを正当化するユースケースは少ない。multi-language/multi-lingualな大規模検索なら検討価値あり。

Q. Azure OpenAI版でも動きますか?

litellm経由で書いているためazure/text-embedding-3-smallにモデル指定を変えるだけで動く。認証情報は環境変数切り替えで対応。

Q. 日本語の精度は?

text-embedding-3-smallは多言語対応。日本語の社内ドキュメント検索でも実用的な精度が出る。英日混在ドキュメントも問題なく扱える。

navigate_before記事一覧へ戻る