TiDBベクトル検索RAG実践：HTAPアーキテクチャにおけるセマンティック検索の5つのコアパターン
数据库
ベクトルデータベースの課題：なぜ純粋なベクトルDBはプロダクションRAGに耐えられないのか？
RAGシステムが本番稼働した。ユーザーが「過去3ヶ月の返金ポリシー」を検索すると、セマンティックに類似しているが期間の合わないドキュメントばかりが返ってくる。時間フィルターを追加したいが、ベクトルデータベースにはトランザクションがなく、リレーショナルクエリとベクトルクエリは別システム、データ同期は手動スクリプト頼み——純粋なベクトルデータベースはプロダクションRAGで4つの致命的な弱点を露呈する：
- トランザクション保証なし：ベクトル書き込みとビジネスデータ更新が同じトランザクションにない、整合性は運任せ
- リレーショナルとベクトルの分断：スカラーフィルタリングはメタデータ事前選別しかできず、真のSQL+セマンティック検索融合が不可能
- 複雑なデータ同期：ビジネスDBとベクトルDBのデュアルライト、CDCパイプラインの保守コストが高く遅延が大きい
- コスト倍増：2つのデータベースは2倍のストレージ、2倍の運用、2倍の障害点を意味する
TiDBベクトル検索はHTAPアーキテクチャで一石三鳥：1つのデータベースでトランザクション、分析、ベクトル検索を同時に処理し、SQLとベクトルクエリがネイティブに融合。
コア概念クイックリファレンス
| 概念 | 説明 | 代表的な実装 |
|---|---|---|
| TiDBベクトル検索 | TiDBネイティブのベクトル列型とHNSWインデックス | TiDB 8.0+ |
| HTAP | ハイブリッドトランザクショナル/アナリティカルプロセッシング、行+列ストレージ | TiDB TiFlash |
| ベクトルインデックス | 高次元ベクトルの近似最近傍インデックス構造 | HNSW、IVF |
| HNSW | 階層型ナビゲーブルスモールワールドグラフベースのANNアルゴリズム | ef_construction、max_level |
| ハイブリッドクエリ | SQLスカラーフィルタリングとベクトル類似度の結合クエリ | WHERE + ORDER BY vec_cosine |
| セマンティック検索 | セマンティック埋め込みに基づく類似度検索 | cosine、L2、inner product |
| 埋め込みモデル | テキストを密ベクトルにマッピングするニューラルネットワーク | text-embedding-3-small、bge-large |
| RAG | 検索拡張生成、LLM+外部知識検索 | LangChain、LlamaIndex |
| 全文検索 | 転置インデックスに基づくキーワード検索 | MATCH AGAINST、FULLTEXT |
| スカラーフィルタリング | ベクトル検索の前/後に条件フィルタを適用 | category='tech'、date > '2026-01' |
問題分析：TiDBベクトルRAGの5つの主要課題
| # | 課題 | 具体的な現れ | 影響 |
|---|---|---|---|
| 1 | ベクトルとリレーショナルデータの融合 | ベクトル列とビジネス列が別テーブル、JOINのパフォーマンスが低い | ハイブリッドクエリの高遅延、UX低下 |
| 2 | クエリパフォーマンス最適化 | HNSWパラメータのチューニングが複雑、大規模データで再現率と遅延のバランスが困難 | 検索が遅いか再現率不足 |
| 3 | 埋め込みモデルの選択 | 次元、言語、ドメイン適合性の差異が大きい | 埋め込み品質の低下による検索精度の低下 |
| 4 | データ更新とインデックス保守 | 増分書き込みがインデックス再構築をトリガー、大規模バッチがオンラインクエリをブロック | 書き込み中のクエリタイムアウト |
| 5 | コスト管理 | ベクトル列のストレージ消費が大きい、HTAPクラスタのリソース消費が高い | ストレージとコンピュートコストが予算超過 |
これら5つの課題は密接に連鎖している：モデル選択がベクトル次元とインデックスパラメータを決定し、インデックスパラメータがクエリパフォーマンスに影響し、データ更新戦略はインデックスメカニズムに依存し、コストはすべての要因に制約される。プロダクション級TiDBベクトルRAGは、これらの課題を体系的に解決しなければならない。
ステップバイステップ実装：5つのコアパターン
パターン1：TiDBベクトルテーブル設計とインデックス作成
RAGの第一歩——ビジネスフィールドとベクトル列を融合したテーブル構造を設計し、HNSWインデックスを作成。
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Index
from sqlalchemy.orm import declarative_base, Session
from tidb_vector.sqlalchemy import VectorType
from datetime import datetime
Base = declarative_base()
class KnowledgeDoc(Base):
__tablename__ = "knowledge_docs"
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String(255), nullable=False)
content = Column(Text, nullable=False)
category = Column(String(50), index=True)
source = Column(String(100))
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
embedding = Column(VectorType(1536))
__table_args__ = (
Index("idx_category_created", "category", "created_at"),
)
TIDB_CONN = "mysql+pymysql://root@127.0.0.1:4000/rag_db"
engine = create_engine(TIDB_CONN)
Base.metadata.create_all(engine)
def create_vector_index():
with engine.connect() as conn:
conn.execute("""
ALTER TABLE knowledge_docs
ADD VECTOR INDEX idx_embedding_vec
USING HNSW (embedding)
WITH (ef_construction = 128, m = 16)
""")
conn.commit()
create_vector_index()
print("ベクトルテーブルとHNSWインデックスの作成完了")
適用シーン：RAGナレッジベースのテーブル設計、複数フィールドのハイブリッドクエリ要件。
パターン2：埋め込み生成とバッチ書き込み
ドキュメント内容を埋め込みモデルでベクトルに変換し、TiDBにバッチ書き込み。
import numpy as np
from openai import OpenAI
from sqlalchemy.orm import Session
class EmbeddingWriter:
def __init__(self, engine, api_key: str,
model: str = "text-embedding-3-small",
dim: int = 1536, batch_size: int = 100):
self.engine = engine
self.client = OpenAI(api_key=api_key)
self.model = model
self.dim = dim
self.batch_size = batch_size
def generate_embedding(self, text: str) -> list[float]:
response = self.client.embeddings.create(
input=text, model=self.model, dimensions=self.dim
)
return response.data[0].embedding
def batch_embed(self, texts: list[str]) -> list[list[float]]:
response = self.client.embeddings.create(
input=texts, model=self.model, dimensions=self.dim
)
return [item.embedding for item in response.data]
def write_documents(self, docs: list[dict]) -> int:
written = 0
for i in range(0, len(docs), self.batch_size):
batch = docs[i : i + self.batch_size]
texts = [d["content"] for d in batch]
embeddings = self.batch_embed(texts)
with Session(self.engine) as session:
for doc, emb in zip(batch, embeddings):
record = KnowledgeDoc(
title=doc["title"],
content=doc["content"],
category=doc.get("category", "general"),
source=doc.get("source", ""),
embedding=emb,
)
session.add(record)
session.commit()
written += len(batch)
return written
writer = EmbeddingWriter(engine, api_key="your-api-key")
docs = [
{"title": "返金ポリシー", "content": "2026年Q2返金ポリシー：購入後30日以内は全額返金可能...", "category": "policy"},
{"title": "APIレート制限", "content": "無料プラン：100リクエスト/分、有料プラン：1000リクエスト/分...", "category": "tech"},
]
count = writer.write_documents(docs)
print(f"{count}件のドキュメントを書き込み")
適用シーン：ナレッジベースの一括インポート、増分ドキュメントの埋め込み書き込み。
パターン3：セマンティック検索とスカラーフィルタリングのハイブリッドクエリ
TiDBのコアアドバンテージ——1つのSQLでセマンティック検索とビジネス条件フィルタリングを同時に完了。
from dataclasses import dataclass
@dataclass
class SearchResult:
id: int
title: str
content: str
category: str
similarity: float
class HybridRetriever:
def __init__(self, engine, api_key: str,
model: str = "text-embedding-3-small"):
self.engine = engine
self.client = OpenAI(api_key=api_key)
self.model = model
def search(self, query: str, category: str | None = None,
date_after: str | None = None,
top_k: int = 5) -> list[SearchResult]:
query_emb = self.client.embeddings.create(
input=query, model=self.model
).data[0].embedding
emb_str = str(query_emb)
sql = """
SELECT id, title, content, category,
1 - vec_cosine_distance(embedding, %s) AS similarity
FROM knowledge_docs
WHERE 1=1
"""
params: list = [emb_str]
if category:
sql += " AND category = %s"
params.append(category)
if date_after:
sql += " AND created_at > %s"
params.append(date_after)
sql += " ORDER BY similarity DESC LIMIT %s"
params.append(top_k)
with self.engine.connect() as conn:
rows = conn.execute(sql, params).fetchall()
return [
SearchResult(
id=r[0], title=r[1], content=r[2],
category=r[3], similarity=round(r[4], 4)
)
for r in rows
]
retriever = HybridRetriever(engine, api_key="your-api-key")
results = retriever.search(
query="返金ポリシー", category="policy", date_after="2026-01-01", top_k=3
)
for r in results:
print(f"[{r.similarity:.3f}] {r.title} ({r.category})")
適用シーン：ビジネス条件付きセマンティック検索、多次元フィルタリング検索。
パターン4：全文検索とベクトル検索の融合
キーワードの完全一致+セマンティック理解のデュアルリコール、RRFアルゴリズムで融合ランキング。
from dataclasses import dataclass
@dataclass
class FusedResult:
id: int
title: str
content: str
rrf_score: float
vector_rank: int
fulltext_rank: int
class FusedRetriever:
def __init__(self, engine, api_key: str, k: int = 60):
self.engine = engine
self.client = OpenAI(api_key=api_key)
self.k = k
def fused_search(self, query: str, top_k: int = 5) -> list[FusedResult]:
query_emb = self.client.embeddings.create(
input=query, model="text-embedding-3-small"
).data[0].embedding
emb_str = str(query_emb)
vec_sql = """
SELECT id, title, content,
ROW_NUMBER() OVER (ORDER BY vec_cosine_distance(embedding, %s)) AS vec_rank
FROM knowledge_docs
ORDER BY vec_cosine_distance(embedding, %s)
LIMIT 50
"""
ft_sql = """
SELECT id, title, content,
ROW_NUMBER() OVER (ORDER BY MATCH(content) AGAINST(%s IN NATURAL LANGUAGE MODE)) AS ft_rank
FROM knowledge_docs
WHERE MATCH(content) AGAINST(%s IN NATURAL LANGUAGE MODE)
LIMIT 50
"""
with self.engine.connect() as conn:
vec_rows = conn.execute(vec_sql, [emb_str, emb_str]).fetchall()
ft_rows = conn.execute(ft_sql, [query, query]).fetchall()
rrf_scores: dict[int, dict] = {}
for row in vec_rows:
doc_id = row[0]
rrf_scores[doc_id] = {
"title": row[1], "content": row[2],
"vec_rank": row[3], "ft_rank": 9999,
"rrf": 1.0 / (self.k + row[3]),
}
for row in ft_rows:
doc_id = row[0]
if doc_id in rrf_scores:
rrf_scores[doc_id]["ft_rank"] = row[3]
rrf_scores[doc_id]["rrf"] += 1.0 / (self.k + row[3])
else:
rrf_scores[doc_id] = {
"title": row[1], "content": row[2],
"vec_rank": 9999, "ft_rank": row[3],
"rrf": 1.0 / (self.k + row[3]),
}
sorted_results = sorted(
rrf_scores.items(), key=lambda x: x[1]["rrf"], reverse=True
)[:top_k]
return [
FusedResult(
id=doc_id, title=v["title"], content=v["content"],
rrf_score=round(v["rrf"], 5),
vector_rank=v["vec_rank"], fulltext_rank=v["ft_rank"],
)
for doc_id, v in sorted_results
]
fused = FusedRetriever(engine, api_key="your-api-key")
results = fused.fused_search("返金ポリシー 2026", top_k=5)
for r in results:
print(f"[RRF:{r.rrf_score:.4f} V:{r.vector_rank} F:{r.fulltext_rank}] {r.title}")
適用シーン：キーワード+セマンティックのデュアルリコール、専門用語の完全一致要件。
パターン5：エンドツーエンドRAGアプリケーション構築
検索、リランキング、生成を完全なRAGパイプラインに連結。
from dataclasses import dataclass
@dataclass
class RAGConfig:
retrieval_top_k: int = 10
rerank_top_n: int = 3
max_context_tokens: int = 3000
llm_model: str = "gpt-4o-mini"
embedding_model: str = "text-embedding-3-small"
class TiDBRAGPipeline:
def __init__(self, engine, api_key: str, config: RAGConfig | None = None):
self.config = config or RAGConfig()
self.engine = engine
self.client = OpenAI(api_key=api_key)
self.retriever = HybridRetriever(
engine, api_key, model=self.config.embedding_model
)
def _rerank(self, query: str, results: list[SearchResult]) -> list[SearchResult]:
scored = []
for r in results:
overlap = sum(
1 for w in query.split() if w in r.content
) / max(len(query.split()), 1)
final_score = 0.7 * r.similarity + 0.3 * overlap
scored.append((r, final_score))
scored.sort(key=lambda x: x[1], reverse=True)
return [r for r, _ in scored[: self.config.rerank_top_n]]
def query(self, question: str, category: str | None = None) -> str:
results = self.retriever.search(
query=question, category=category,
top_k=self.config.retrieval_top_k
)
reranked = self._rerank(question, results)
context = "\n\n".join(
f"【{r.title}】({r.category})\n{r.content}" for r in reranked
)
if len(context) > self.config.max_context_tokens * 4:
context = context[: self.config.max_context_tokens * 4]
prompt = (
"以下の検索結果に基づいて質問に答えてください。コンテキストに十分な情報がない場合は、明確にその旨を述べてください。\n\n"
f"コンテキスト：\n{context}\n\n質問：{question}"
)
response = self.client.chat.completions.create(
model=self.config.llm_model,
messages=[{"role": "user", "content": prompt}],
max_tokens=800,
)
return response.choices[0].message.content
pipeline = TiDBRAGPipeline(engine, api_key="your-api-key")
answer = pipeline.query("2026年Q2の返金ポリシーは何ですか？", category="policy")
print(answer)
適用シーン：プロダクション級RAG Q&Aシステム、エンタープライズナレッジベースアシスタント。
落とし穴ガイド：5つのよくある罠
罠1：ベクトル列にインデックスを作成せずにクエリ
❌ 誤った方法：
SELECT * FROM knowledge_docs
ORDER BY vec_cosine_distance(embedding, @query_emb)
LIMIT 10;
✅ 正しい方法：
ALTER TABLE knowledge_docs
ADD VECTOR INDEX idx_vec USING HNSW (embedding)
WITH (ef_construction = 128, m = 16);
-- クエリ時にef_searchを指定
SET SESSION tidb_vector_ef_search = 64;
SELECT * FROM knowledge_docs
ORDER BY vec_cosine_distance(embedding, @query_emb)
LIMIT 10;
罠2：埋め込み次元とインデックス次元の不一致
❌ 誤った方法：
embedding = Column(VectorType(768)) # テーブル定義は768次元
# しかしtext-embedding-3-smallの1536次元ベクトルを書き込み
✅ 正しい方法：
EMBEDDING_DIM = 1536
class KnowledgeDoc(Base):
embedding = Column(VectorType(EMBEDDING_DIM))
# 埋め込み生成時に次元を明示的に指定
response = client.embeddings.create(
input=text, model="text-embedding-3-small", dimensions=EMBEDDING_DIM
)
罠3：バルク書き込み時にインデックスを無効化しない
❌ 誤った方法：
for doc in large_doc_list:
session.add(KnowledgeDoc(embedding=doc["emb"]))
session.commit() # 各書き込みがインデックス更新をトリガー
✅ 正しい方法：
# 先にベクトルインデックスを無効化し、バルク書き込み後に再構築
with engine.connect() as conn:
conn.execute("ALTER TABLE knowledge_docs ALTER INDEX idx_vec INVISIBLE")
conn.commit()
bulk_write(large_doc_list)
with engine.connect() as conn:
conn.execute("ALTER TABLE knowledge_docs ALTER INDEX idx_vec VISIBLE")
conn.commit()
罠4：ハイブリッドクエリでスカラーフィルタリングをベクトルランキングの後に適用
❌ 誤った方法：
SELECT * FROM knowledge_docs
ORDER BY vec_cosine_distance(embedding, @emb)
LIMIT 100;
-- その後アプリケーション層でcategoryをフィルタリング
✅ 正しい方法：
SELECT * FROM knowledge_docs
WHERE category = 'policy' AND created_at > '2026-01-01'
ORDER BY vec_cosine_distance(embedding, @emb)
LIMIT 10;
罠5：RAGコンテキストを切り詰めずにLLMに直接入力
❌ 誤った方法：
context = "\n".join(r.content for r in all_results)
prompt = f"コンテキスト：{context}\n質問：{question}"
✅ 正しい方法：
MAX_CTX_TOKENS = 3000
def truncate_context(results: list[SearchResult]) -> str:
parts, total = [], 0
for r in results:
est_tokens = len(r.content) // 4
if total + est_tokens > MAX_CTX_TOKENS:
break
parts.append(f"【{r.title}】\n{r.content}")
total += est_tokens
return "\n\n".join(parts)
エラートラブルシューティング：10のよくあるエラー
| # | エラーメッセージ | 原因 | 解決策 |
|---|---|---|---|
| 1 | Vector dimension mismatch: expected 1536 got 768 |
埋め込み次元がテーブル定義と一致しない | 埋め込みモデルの次元を統一、またはVectorTypeパラメータを変更 |
| 2 | HNSW index build OOM |
ef_constructionが大きすぎる、またはデータ量がメモリを超過 | ef_constructionを64に下げ、バッチでインデックスを構築 |
| 3 | vec_cosine_distance function not found |
TiDBバージョンが8.0未満、またはベクトル機能が有効化されていない | TiDBを8.0+にアップグレード、ベクトルプラグインのロードを確認 |
| 4 | FULLTEXT index not found on column content |
全文インデックスが作成されていない | ALTER TABLE knowledge_docs ADD FULLTEXT INDEX ft_content(content) |
| 5 | Query timeout during vector search |
HNSW ef_searchが大きすぎる、またはデータ量が予想を超過 | ef_searchを下げ、スカラーフィルタリングでスコープを絞る |
| 6 | Duplicate entry for key 'PRIMARY' |
バルク書き込み時のID衝突 | auto_incrementを使用、またはID範囲を明示的に指定 |
| 7 | Embedding API rate limit exceeded |
埋め込みAPI呼び出し頻度が制限を超過 | リクエスト間隔を追加、batch APIを使用、バックオフリトライを実装 |
| 8 | TiDB connection pool exhausted |
同時クエリが多すぎる、接続が解放されていない | 接続プールサイズを設定、with文で接続の返却を確実に |
| 9 | Index invisible after bulk load |
バルク書き込み後にインデックスの可視性を復旧し忘れ | ALTER INDEX idx_vec VISIBLEを実行し検証 |
| 10 | RRF fusion returns empty results |
ベクトル検索と全文検索の両方でマッチなし | 類似度閾値を緩和、フォールバック検索戦略を追加 |
高度な最適化：4つのキーテクニック
1. HNSWパラメータの自動チューニング
class HNSWAutoTuner:
def __init__(self, engine):
self.engine = engine
def recommend_params(self, row_count: int,
recall_target: float = 0.95) -> dict:
if row_count < 100_000:
return {"ef_construction": 64, "m": 8, "ef_search": 40}
elif row_count < 1_000_000:
return {"ef_construction": 128, "m": 16, "ef_search": 64}
else:
return {"ef_construction": 256, "m": 24, "ef_search": 128}
def apply(self, table: str, params: dict):
with self.engine.connect() as conn:
conn.execute(
f"ALTER TABLE {table} ADD VECTOR INDEX idx_vec "
f"USING HNSW (embedding) WITH "
f"(ef_construction={params['ef_construction']}, m={params['m']})"
)
conn.execute(
f"SET SESSION tidb_vector_ef_search = {params['ef_search']}"
)
conn.commit()
2. 埋め込みキャッシュで重複計算を回避
import hashlib
import json
class EmbeddingCache:
def __init__(self, engine, ttl_hours: int = 24):
self.engine = engine
self.ttl = ttl_hours
self._local: dict[str, list[float]] = {}
def _cache_key(self, text: str, model: str) -> str:
raw = f"{model}:{text}"
return hashlib.md5(raw.encode()).hexdigest()
def get_or_compute(self, text: str, model: str,
compute_fn) -> list[float]:
key = self._cache_key(text, model)
if key in self._local:
return self._local[key]
emb = compute_fn(text, model)
self._local[key] = emb
return emb
3. クエリルーター：検索戦略の自動選択
class QueryRouter:
def route(self, query: str) -> str:
has_exact_terms = any(
kw in query for kw in ["番号", "ID", "コード", "バージョン"]
)
is_recent = any(
kw in query for kw in ["最新", "最近", "今日", "今週"]
)
if has_exact_terms and is_recent:
return "fused"
elif has_exact_terms:
return "fulltext"
elif is_recent:
return "hybrid"
else:
return "vector"
4. ベクトル検索パフォーマンスモニタリング
class VectorSearchMonitor:
def __init__(self, engine):
self.engine = engine
def get_index_stats(self, table: str) -> dict:
with self.engine.connect() as conn:
count = conn.execute(
f"SELECT COUNT(*) FROM {table}"
).scalar()
null_embs = conn.execute(
f"SELECT COUNT(*) FROM {table} WHERE embedding IS NULL"
).scalar()
return {
"total_rows": count,
"null_embeddings": null_embs,
"coverage": round((count - null_embs) / count, 4) if count else 0,
}
比較分析：4つのベクトル検索ソリューションの包括的比較
| 次元 | TiDB Vector | Pinecone | Milvus | pgvector |
|---|---|---|---|---|
| SQL+ベクトル融合 | ネイティブ対応 | 非対応 | スカラーフィルタリング | ネイティブ対応 |
| ACIDトランザクション | フルACID | なし | 限定 | フルACID |
| HTAP分析 | 行+列ストレージ | なし | なし | なし |
| 水平スケーリング | 自動スケーリング | 自動 | 手動シャーディング | Citusが必要 |
| インデックスタイプ | HNSW | 自動 | IVF/HNSW/DiskANN | HNSW/IVFFlat |
| 最大次元数 | 16384 | 2048 | 32768 | 2000（HNSW） |
| 全文検索 | ネイティブFULLTEXT | なし | なし | tsvector |
| デプロイ方法 | セルフホスト/Serverless | クラウドのみ | セルフホスト/クラウド | セルフホスト |
| 運用複雑さ | 低（マネージド） | 極低 | 高 | 中 |
| 適用シーン | RAG+ビジネス融合 | 純ベクトルSaaS | 大規模ベクトル | PGエコシステム |
TiDB Vectorのコアアドバンテージ：1つのデータベースでSQL+ベクトル+全文+トランザクションを処理——データ同期パイプライン不要。
まとめと展望
TiDBベクトル検索は2026年のRAGプロダクションデプロイにおける重要なインフラになりつつある：
- Serverlessベクトル：TiDB Cloud Serverlessはベクトルクエリ量に基づく課金——ゼロ運用でRAGを開始
- マルチモーダルベクトル：画像、音声の埋め込みをテキスト埋め込みと統合保存、クロスモーダル検索
- ストリーミング埋め込み更新：CDC駆動の自動埋め込み再計算、データ変更をベクトルインデックスにリアルタイム反映
- アダプティブインデックス：クエリパターンに基づいてHNSWパラメータを自動調整、手動チューニング不要
- RAG評価の標準化：TiDB内蔵の検索品質メトリクス、エンドツーエンドRAG評価がすぐに利用可能
TiDBベクトルRAGの選択原則：RAGにSQLフィルタリング、トランザクション保証、またはビジネスデータ融合が必要な場合、TiDBが唯一のワンストップソリューション。純粋なベクトル検索シナリオではPineconeを、超大规模ベクトルストアではMilvusを検討。まずはTiDB Cloud Serverlessの無料枠で効果を検証することをお勧めする。
オンラインツール推奨
- JSONフォーマッター — 埋め込みベクトルと検索結果のJSONデータをフォーマット
- ハッシュ計算 — ドキュメントフィンガープリントと埋め込みキャッシュのMD5/SHAハッシュを計算
- Curl to Code — TiDB APIと埋め込みインターフェースのデバッグcurlをPythonコードに変換
ブラウザローカルツールを無料で試す →
#TiDB向量搜索#RAG生产部署#向量数据库#HTAP#语义检索#2026#数据库