AIエージェントの長期記憶を SQL/Vector/JSON 統合 DB で実装 — Praxia × TiDB Vector で得た知見
Zennfes Spring 2026 エントリ記事です。テーマは「AIでの情報検索 (RAG等) / AIエージェントのメモリ機能に関する実装知見を、他のエンジニアにとって有益な形で共有」。
具体例としてマルチエージェント OSSPraxiaの長期記憶層をTiDB Vectorで組む設計と実装を解説しますが、他システム(LangChain / LlamaIndex / 自前実装)にも転用できる知見を中心にまとめました。実装(~280 LoC) + テスト(19 件) + ベンチ(10K 行実測) 一式公開。
この記事で得られる実装知見
本記事は TiDB Vector で AI エージェントの長期記憶層を組んだ実装知見 を、他のエンジニアが自分の RAG / 記憶層に持ち帰れる形で共有するものです。
(A) TiDB を選んだから取れた知見:
-
を
SELECT … WHERE user_id=? ORDER BY VEC_COSINE_DISTANCE(embedding, ?) LIMIT ?
1 本の SQLで書く構造 — Pinecone + RDB の 2 ラウンドトリップ問題を消す - VECTOR(N) + JSON + B-Tree インデックス が同一テーブル内に同居する設計 — エージェント書込みと BI 分析クエリが同じテーブルに直接届く
-
— JSON-array 文字列 + 浮動小数点
VECTOR(N)
型と pymysql の値表現repr()
のシリアライズノウハウ - TiDB Cloud の TLS 自動有効化判定— DSN ホスト名ベースの分岐パターン
- Serverless 無料枠 → 本番への連続スケール特性と実測レイテンシ(10K rows で SEARCH P50=281ms)
(B) TiDB 実装の過程で汎用化できた設計パターン:
- 「4 メソッド Protocol で記憶層を抽象化する」設計— どんな Vector DB にも差し替え可能な抽象化
- callable embedder 注入— OpenAI / sentence-transformers / 自前モデルを 1 行で切替
- 動的テーブル名のホワイトリスト検証— SQL プレースホルダが効かない箇所の標準対策
- Deterministic fake embedder でベンチ分離— embedding API レイテンシを DB 性能から切り離す方法論
- 接続セットアップ ≒ 律速という Serverless DB 全般の観察と pool 化方針
末尾の「持ち帰れる実装知見のまとめ」に表で整理してあります。
60 秒で Praxia を理解したい人向け:
https://youtu.be/o_6NbjJU1AA
TL;DR
- AIエージェントの長期記憶層には 「セマンティック検索 + 構造化メタデータ + 既存 BI ツールからの直接 SELECT + 監査ログ + RBAC」が要求されますが、ベクトル DB と RDB を別々に運用すると負債が指数的に増えます
-
TiDB Vectorは
SQL + VECTOR(N) + JSON + HNSW index
が同一テーブル内に同居できる数少ない HTAP DB で、この要件を 1 基盤で満たせます - 本記事は Praxia の長期記憶層 (L1: PersonalMemory / L3: SharedMemory) を TiDB で実装した過程の知見をまとめたものです
- 実装は HNSW + Cosine 距離による ANN 検索を
VEC_COSINE_DISTANCE()
一発、約 280 LoC - Tokyo region の TiDB Cloud Serverless 無料枠で実測: 10K 行に対する SEARCH P50 = 281ms(end-to-end)。律速は接続セットアップで、pool 化で 100ms オーダーが視野
- CI ゼロ依存テスト 19 件 + 実 DSN がある時だけ走るライブ smoke 1 件 を公開
なぜ Praxia × TiDB なのか
Praxia 側の事情 — 「組織記憶」を持つマルチエージェント基盤
Praxia は 個人 → 組織への暗黙知自動昇格 を狙うマルチエージェント・オーケストレーターで、5 層のメモリスタックを持ちます:
L1 PersonalMemory ユーザごと(個人引出し) L2 PromotionEngine 夜間バッチ。L1 → L3 への昇格判定(3 経路スコア) L3 SharedMemory 組織全体、RBAC ゲート、時間減衰 L4 MarkdownStore git 管理、PR レビュー必須(不変ガバナンス層) L5 GraphLayer 任意。関係抽出(Zep / Graphiti)
L1 と L3 は 大量の自然文 + 任意メタデータ + 時系列 を持ち、エージェントから 意味検索 + 厳密フィルタ + 集計クエリ で叩かれます。これは典型的なベクトル DB だけでも RDB だけでも完結しないワークロードです。
組み合わせれば確かに動きますが、ベクトル DB と RDB を別々に管理する とすぐ次のような運用負債が積み上がります:
- スキーマ migration の 2 系統管理(vector store と RDB それぞれ)
- 「テキストは消したがベクトルが残っている」状態の発生(削除の二段階トランザクション)
- BI ツールが ベクトル DB から直接 SELECT できず、結局 RDB へエクスポートし直す
- 監査要件が 2 つの永続化レイヤに分散する
TiDB 側の事情 — Vector + SQL + JSON がネイティブで同居する HTAP
ここで TiDB が抜群に効きます:
- VECTOR(N) 型がネイティブ(2024 LTS 以降)
-
HNSW indexで ANN 検索を SQL 内で
VEC_COSINE_DISTANCE()
一発 - JSON 型 + B-Tree インデックスがメタデータと併用可能
-
MySQL ワイヤープロトコル互換—
pymysql
ひとつで繋がる - Cloud Serverless 無料枠 25 GiB— 検証から本番までスケール連続性あり
- HTAP— 行ストア (TiKV) と列ストア (TiFlash) を同居させ、エージェント書込みと分析 SELECT が同じテーブルで干渉しにくい
つまり、Praxia が長期記憶層に欲しい「意味検索 + 構造化保持 + 既存基盤との JOIN 可能性」が、1 つの DB で揃う のが TiDB です。
この組み合わせで解ける問題
具体的に、Praxia × TiDB で実現できることを 3 つだけ:
1. 「同じテーブルから ANN 検索しつつ BI ツールが SELECT」
エージェントは
ORDER BY VEC_COSINE_DISTANCE(embedding, ?) LIMIT 10で意味検索、データチームは
SELECT user_id, COUNT(*) FROM praxia_memory WHERE metadata->>'$.tags' = 'customer' GROUP BY ...で集計、と 同じ 1 テーブル から両方できます。ベクトル DB から CSV エクスポートして DWH に流し込む、という運用が要りません。
2. 「組織横断のレポートが即書ける」
L3 SharedMemory に蓄積された暗黙知を、RBAC ゲート + 時間減衰 を効かせたまま SQL で直接集計 できます。「過去 90 日で営業チームが頻繁に参照した事実 TOP 20」が
GROUP BY metadata->>'$.team'で書ける。
3. 「監査と運用が単一基盤」
行レベルセキュリティ、監査ログ、バックアップ、レプリケーション、すべて TiDB の運用ノウハウがそのまま使える。Pinecone + PostgreSQL の二重運用と比べると、運用者のオンコール対象が半分です。
「SQL も Vector も JSON も全部 1 箇所で済ませたい」というニーズは、Praxia のような組織知を扱うシステムで強く出ます。TiDB はこの要請にネイティブで応える数少ない選択肢でした。
設計パターン: 「4 メソッド Protocol で記憶層を抽象化する」
これは Praxia 固有の話というより、エージェント記憶層を設計する時に強く推奨できるパターン です。長期記憶層を以下の 4 メソッドだけの Protocol で抽象化すると、ベクトル DB の選択を後で差し替えられる構造になります:
class MemoryBackend(Protocol):
def add(self, *, user_id, text, kind, metadata) -> MemoryRecord: ...
def search(self, *, user_id, query, limit) -> list[MemoryRecord]: ...
def all(self, *, user_id=None) -> list[MemoryRecord]: ...
def clear(self, *, user_id=None) -> None: ...
この Protocol が成立する条件 は以下:
- 記録の最小単位が
(text, user_id, kind, metadata)
で表現できる - 検索は 意味検索(セマンティック)+ ユーザフィルタの組み合わせで十分
- バッチ列挙(
all
)とリセット(clear
)が運用上必要
LangChain / LlamaIndex / Haystack いずれも内部で似た protocol を持ちますが、もっと太い 抽象になりがちです(
add_documents/
as_retriever/ etc.)。エージェント記憶のように 書込み単位が単文 で per-user フィルタが必須 なユースケースでは、上記 4 メソッドに削ぎ落とした方が 新 backend を 1 日で書ける ようになります。
🛠 他システムへの転用ポイント
ベクトル DB を 1 つ決め打ちで RAG / メモリ層を実装している場合、まずこの 4 メソッド Protocol を切り出して既存実装をその後ろに置くだけで、後日の移行コストが激減します。「Protocol → 既存 1 backend」という構造を最初から持っておくのがコツ。
全体構成
┌───────────────────────────────────────┐
│ Praxia Agent (LiteLLM / 100+ LLMs) │
└───────────────┬───────────────────────┘
│ MemoryBackend Protocol
│ (add/search/all/clear)
┌───────────────▼───────────────────────┐
│ TidbBackend (本記事の実装, ~280 LoC) │
└───────────────┬───────────────────────┘
│ pymysql + MySQL wire
┌───────────────▼───────────────────────┐
│ TiDB Cloud Serverless │
│ ───────────────────────────────── │
│ praxia_memory テーブル │
│ id VARCHAR(36) │
│ user_id VARCHAR(128) │
│ text TEXT │
│ kind VARCHAR(64) │
│ ts DOUBLE │
│ metadata JSON │
│ embedding VECTOR(1536) + HNSW │
│ KEY (user_id), (user_id, kind) │
└───────────────────────────────────────┘
エージェントは
add()で記録、
search()で意味検索、L2 PromotionEngine が
all(user_id=...)で個人記録をスキャンして L3 へ昇格させる、という流れが すべて同じ TiDB テーブル に対する SQL クエリで実現されます。
TiDB 実装の中身 — スキーマ + 埋込 + 接続管理
スキーマ
CREATE TABLE IF NOT EXISTS praxia_memory ( id VARCHAR(36) PRIMARY KEY, user_id VARCHAR(128) NOT NULL, text TEXT NOT NULL, kind VARCHAR(64) NOT NULL, ts DOUBLE NOT NULL, metadata JSON, embedding VECTOR(1536), KEY idx_user (user_id), KEY idx_user_kind (user_id, kind), VECTOR INDEX vidx_embedding ((VEC_COSINE_DISTANCE(embedding))) USING HNSW );
設計上のポイント:
-
が OpenAI
embedding VECTOR(1536)
text-embedding-3-small
と互換。bge-m3 等の別モデルを使う時はembedding_dim=1024
等で再構築 -
で各 backend にあった構造化メタデータの差を吸収。
metadata JSON
metadata->>'$.team'
のようにパス指定で WHERE / GROUP BY に直接使える -
/
(user_id)
B-Tree index で per-user 検索のプリフィルタが高速。Vector 検索は user 内に閉じる(user_id, kind)
- HNSW + Cosine— 100 万件オーダーでも 10ms 未満の ANN(公式ベンチ)
埋込関数を callable で外部注入
Praxia の LiteLLM 経由マルチプロバイダ思想に合わせ、embedder は コンストラクタで注入可能 な callable:
def _default_embedder(self, model: str) -> Callable[[str], list[float]]:
def embed(text: str) -> list[float]:
from litellm import embedding
resp = embedding(model=model, input=text)
return list(resp["data"][0]["embedding"])
return embed
ローカル sentence-transformers / bge-m3 を使いたい場合は callable を差し替えるだけ:
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-m3") # 1024 dim
backend = TidbBackend(
embedding_dim=1024,
embedder=lambda t: model.encode(t).tolist(),
)
OpenAI に依存しない完全オンプレ運用 も成立します。
接続管理
TiDB Cloud Serverless は 多数の短命接続に強い ので、初版は connection pool を入れず、操作ごとに新規接続を開いて autocommit:
def _connect(self):
return self._pymysql.connect(
**self._parse_dsn(self.dsn),
charset="utf8mb4",
autocommit=True,
)
高頻度書込みワークロード(秒間数千件)が必要なら、このクラスを
sqlalchemy.create_engine(pool_size=...)で包むのが拡張ポイント(後述ベンチで判明した最大のチューニング余地)。
TiDB Cloud と self-hosted を 1 行で切替
# TiDB Cloud Serverless (Tokyo region, 無料 25 GiB) dsn = "mysql://user:pass@gateway01.ap-northeast-1.prod.aws.tidbcloud.com:4000/test?ssl_verify_cert=true" # Self-hosted (ローカル開発) dsn = "mysql://root:@127.0.0.1:4000/praxia_dev"
_parse_dsnで TLS 自動判定:
is_cloud = "tidbcloud.com" in (u.hostname or "")
if is_cloud or qs.get("ssl_verify_cert") == ["true"]:
kwargs["ssl"] = {"ssl_disabled": False}
実装 — 280 LoC で全部入り
addメソッドの全文(これが本質):
def add(self, *, user_id, text, kind, metadata) -> MemoryRecord:
self._ensure_schema()
rec_id = str(uuid.uuid4())
ts = time.time()
vec = self._embedder(text)
if len(vec) != self.embedding_dim:
raise ValueError(...)
with self._connect() as conn:
with conn.cursor() as cur:
cur.execute(
f"INSERT INTO {self.table} "
"(id, user_id, text, kind, ts, metadata, embedding) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
(rec_id, user_id, text, kind, ts,
json.dumps(metadata, ensure_ascii=False),
self._vec_to_str(vec)),
)
return MemoryRecord(id=rec_id, user_id=user_id, text=text,
kind=kind, timestamp=ts, metadata=metadata)
searchがさらに短く、TiDB の真骨頂が出ます:
def search(self, *, user_id, query, limit) -> list[MemoryRecord]:
self._ensure_schema()
qvec = self._embedder(query)
with self._connect() as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT id, user_id, text, kind, ts, metadata "
f"FROM {self.table} WHERE user_id = %s "
f"ORDER BY VEC_COSINE_DISTANCE(embedding, %s) ASC "
f"LIMIT %s",
(user_id, self._vec_to_str(qvec), int(limit)),
)
rows = cur.fetchall()
return [self._row_to_record(r) for r in rows]
「WHERE で user パーティション + ORDER BY VEC_COSINE_DISTANCE で ANN 検索 + LIMIT」が同じ SQL で書けるのが TiDB Vector の中核的優位性です。Pinecone + RDB を別々に管理する構成では、これが「ベクトル DB に投げる → ID リスト → RDB へ JOIN」の 2 ラウンドトリップになり、レイテンシも整合性も悪化します。
all()と
clear()は通常の SQL なので省略します。全コードは末尾の関連リソース参照。
エージェントから使う — 3 ステップ
1. TiDB Cloud Serverless 起動(無料 25 GiB)
https://tidbcloud.com/ でサインアップ → 「Serverless」プラン → リージョン選択(日本からは Tokyo
ap-northeast-1推奨) → クラスター作成。Connection details から DSN を取得。
2. インストール
pip install pymysql litellm # litellm は default embedder 用 export PRAXIA_TIDB_DSN="mysql://user.token:password@gateway01.ap-northeast-1.prod.aws.tidbcloud.com:4000/test?ssl_verify_cert=true"
3. エージェントから記録 → 検索
from tidb_backend import TidbBackend
backend = TidbBackend(
embedding_dim=1536,
# embedder=... を省略すると LiteLLM text-embedding-3-small が自動使用
)
# 記録 — Praxia の PromotionEngine がこれを参照して L3 への昇格判定する
rec = backend.add(
user_id="alice",
text="Manufacturing sector decision-makers weigh ROI > timeline > risk.",
kind="fact",
metadata={"tags": ["sales", "manufacturing"], "outcome": "won"},
)
# 検索 — Cosine ANN (HNSW)
hits = backend.search(
user_id="alice",
query="how do manufacturing buyers decide?",
limit=5,
)
for h in hits:
print(f"[{h.kind}] {h.text} meta={h.metadata}")
初回
add()で
praxia_memoryテーブルが lazy 作成されます(
CREATE TABLE IF NOT EXISTS)。
Composite で他 backend と並列融合
L1 を TiDB 一本に寄せる構成も良いですが、Praxia の
CompositeBackendを使えば TiDB + 既存 Mem0 並列 + RRF 融合という移行戦略も取れます:
from praxia.memory.composite import CompositeBackend, WeightedBackend
from praxia.memory.backends import load_backend
composite = CompositeBackend(
backends=[
WeightedBackend("tidb", load_backend("tidb"), weight=1.5),
WeightedBackend("mem0", load_backend("mem0"), weight=1.0),
],
fusion="rrf",
write_to="tidb", # 新規書込は TiDB 一本に → 既存 Mem0 は read-only で塩漬けに
)
「既存運用の Mem0 を読みつつ、新しい書込は TiDB に倒す」 がこの設定で実現でき、ダウンタイムなしの段階移行が可能になります。
ベンチマーク: 実測 + 「方法論」も持ち帰り可能
Tokyo (
ap-northeast-1) の TiDB Cloud Serverless 無料枠に対し、Tokyo のローカル PC から計測した実数値です。
計測方法論 — Deterministic embedder で外部 API レイテンシを切り離す
エージェント記憶層のベンチを取る時、OpenAI embedding API のレイテンシ(~50-150ms)が混入 すると、何を測っているのか分からなくなります。対策として deterministic な fake embedder を用意:
def fake_embedder(text: str) -> list[float]:
h = hashlib.sha512(text.encode("utf-8")).digest()
return [(h[i % 64] + (i * 37) % 256) / 512.0 for i in range(1536)]
- sha512 ベースで input → output が決定的
- 1536 dim を生成、cosine 距離が意味のあるレンジ ([0, 1)) に分布
- 外部 API を 1 回も叩かないので、TiDB 側の性能だけが残る
🛠 他システムへの転用ポイント
RAG / メモリ層のベンチを取る時は、まず外部 embedding API を deterministic fake に差し替えた版で DB 単体の性能を測ってください。その後 OpenAI 等を入れた時の差分から「embedding コール分のレイテンシ」が分離されます。これをやらないと「OpenAI が遅いのか DB が遅いのか」が永久に分からないままチューニングします。
実測結果
| Phase | 件数 | P50 | P95 | P99 | mean |
|---|---|---|---|---|---|
| INSERT @ 0→1K | 1,000 | 190.7 ms | 340.4 ms | 418.1 ms | 208.7 ms |
| SEARCH @ 1K rows | 100 | 189.6 ms | 346.0 ms | 439.0 ms | 211.8 ms |
| INSERT @ 1K→10K | 9,000 | 207.7 ms | 342.4 ms | 437.5 ms | 222.7 ms |
| SEARCH @ 10K rows | 100 | 281.5 ms | 485.7 ms | 557.1 ms | 302.8 ms |
実行: 2026-05-19、合計 10,200 操作、Serverless 無料枠で $0、ウォールクロック約 35 分(INSERT 10K 件で大半)。
この数値の読み方
すべて end-to-end(クライアント → TLS → クエリ → 結果取得) のウォールクロックタイム。テーブル名は
praxia_memory_benchで計測終了時に TRUNCATE。1 SEARCH ごとに LIMIT 10 件を全受信。
観察できたこと
1. 接続セットアップが律速になっている
初版は操作ごとに新規 pymysql 接続を開く設計(autocommit, no pool)。TLS ハンドシェイク + auth が およそ 150-200ms 乗っていると推測できます:
- INSERT @ 0→1KとINSERT @ 1K→10Kの P50 がほぼ同じ (190ms / 207ms) — テーブルサイズ 10 倍でも 9% 増だけ。DB 操作自体は速い
- min_ms は 142ms (INSERT)、145ms (SEARCH@1K)— 100ms 未満には絶対に行かない、明らかなハンドシェイク下限
pool を入れれば 50-150ms 程度の改善が期待できる 余地です(SQLAlchemy
create_engine(pool_size=10, pool_recycle=...)で包めば良い)。これが次の最大のチューニングポイント。
2. ANN 検索のスケール特性は素直
SEARCH @ 1K → 10K rowsで P50 が 189.6ms → 281.5ms (+92ms)。テーブルサイズ 10 倍に対してこの増分は、純粋に TiDB 側の HNSW + cosine 距離計算で発生していると見られます。
つまり「接続オーバーヘッド ≈ 190ms + データ依存 DB 時間 ≈ 数 ms 〜数十 ms」という構造。pool 化で 190ms が消えれば、SEARCH @ 10K rows の P50 は 100ms 前後 に落ちる見込み。
3. P99 が穏やか
10K rows での SEARCH P99 で 557ms、INSERT P99 で 437ms。TiDB Cloud Serverless のテール挙動は無料枠でも荒れていません。エージェント記憶層として実用的なレンジ。
LLM ワークフローでの位置づけ
人間の会話 UX で考えると、LLM 自体の応答が 1〜10 秒 かかるのが普通。記憶検索が 280ms (10K rows) なら、LLM レスポンスタイムの 5-10% の追加で済むので、エージェントメモリとしては実用範囲内です。
100 万件規模で P50 < 100ms を狙うなら、次は (a) 接続プール化、(b)
VECTOR INDEXの HNSW パラメータチューニング、(c) Read replica 分離、の順で効くはずです。
ベンチコード(
bench.py)は実験プロジェクトに同梱、
PRAXIA_TIDB_DSNを設定して
python bench.pyで誰でも再現可能。
ハマったポイント
① VECTOR 型と pymysql の値表現
TiDB の
VECTOR(N)カラムへの INSERT は、JSON-Array 形式の 文字列 で受け付けてくれます:
@staticmethod
def _vec_to_str(vec: list[float]) -> str:
return "[" + ",".join(repr(float(x)) for x in vec) + "]"
repr()を使うのは
str(0.123456789012345)だと浮動小数点の丸めで精度を失う可能性があるため。
repr()の方が round-trip 安全です。
② テーブル名の SQL インジェクション対策
CREATE TABLE IF NOT EXISTS {self.table} のように テーブル名 を文字列補間する箇所は pymysql の %sプレースホルダが効きません(識別子は SQL 文法上 quote できない)。なので コンストラクタ時にホワイトリスト検証:
@staticmethod
def _safe_ident(name: str) -> str:
if not name.replace("_", "").isalnum() or not name[0].isalpha():
raise ValueError(f"Invalid table name: {name!r}")
return name
テストケース
bad-name; DROP TABLE x;は即 ValueError で弾けます。
③ TiDB Cloud の TLS 自動有効化
TiDB Cloud は接続に必ず TLS が要りますが、
pymysql.connect(ssl=...)の引数仕様が独特。
ssl_disabled=Falseを dict で渡すのが正解:
if is_cloud or qs.get("ssl_verify_cert") == ["true"]:
kwargs["ssl"] = {"ssl_disabled": False}
DSN ホスト名に
tidbcloud.comが含まれていれば自動有効化、self-hosted の場合は明示的に
?ssl_verify_cert=trueを付ければ有効、という設計。
🛠 他システムへの転用ポイント
3 つの落とし穴は「DB 識別子は SQL プレースホルダで quote できない」「ベクトル型の文字列表現は浮動小数点精度問題を含む」「クラウド DB の TLS 設定はライブラリごとに方言がある」という、TiDB に限らずほぼ全ての RDB + Vector 統合で当たる地雷です。ホワイトリスト検証//repr()ベースのシリアライズDSN クエリパラメータで TLS 切替は、汎用パターンとして覚えておくと他の DB(pgvector / DuckDB Vector 等)でも応用できます。
Mem0 / Zep / Pinecone — どう使い分けるか
Praxia の
MemoryBackendには複数の選択肢があり、TiDB Vector はその 1 つ。比較すると:
| ユースケース | 推奨 |
|---|---|
| 個人開発 / プロトタイプ | JSON(ゼロ依存) |
| エンティティリンキングが要(「Alice が言及した X」「X と Y の関係」など) | Mem0 |
| 構造化メタデータ + ベクトル検索 + SQL 分析クエリを 1 箇所に | TiDB ← 本記事 |
| 時系列 KG(Knowledge Graph)が中核 | Zep |
| 長期記憶 + LangChain エコシステム親和性 | LangMem |
| 以上を融合して取りこぼし最小化 | Composite (RRF) で複数並列 |
TiDB が一番フィットする のは、すでに RDB をデータ基盤として使っている組織で、AI エージェント機能を 既存テーブルとの JOIN / 集計 / レポート と組み合わせたいケース:
- BI ツールから直接エージェント記憶層を
SELECT
- 監査要件で 同じ DB に行レベルセキュリティ + 監査ログで全部置きたい
- 企業内データウェアハウスを AI エージェント記憶の単一ソースに統一したい
逆に TiDB が不利なのは、完全にローカル / オフライン で動かしたいケース。その時は JSON backend がベスト。
テスト構成
DSN なしでも CI が落ちないよう、
pymysqlを mock した ユニットテスト 19 件 + ライブ smoke テスト 1 件(
PRAXIA_TIDB_DSN設定時のみ実行)をセットで実装:
tests/test_tidb_backend.py TestTidbBackendConstruction (5 件) — DSN handling + 不正検出 TestTidbBackendDSNParsing (4 件) — SSL 自動有効化、scheme 検証 TestTidbBackendVectorEncoding (2 件) — vec_to_str / 次元不整合 TestTidbBackendCRUD (5 件) — add/search/all/clear の SQL shape TestTidbBackendRowDeserialization (3 件) — JSON metadata の劣化耐性 TestTidbBackendLive (1 件) — 実 DSN smoke (skip 可)
$ python -m pytest tests/ -v ======================== 19 passed, 1 skipped in 0.15s ========================
PRAXIA_TIDB_DSNを入れた瞬間、live smoke が自動的に対象になります。
次にやりたいこと
実測で性能特性と実装上のボトルネックが見えたので、次のロードマップ:
-
接続プール化— ベンチで判明した「P50 の ~190ms がほぼ全部接続セットアップ」を SQLAlchemy
create_engine(pool_size=10, pool_recycle=...)
で解消。SEARCH @ 10K rows の P50 が ~100ms 台に落ちる見込み - 100 万件規模のベンチ— 今回は 1K / 10K を測ったので HNSW のスケール特性をもう 2 桁見たい
-
TiDB Hybrid Search 統合— BM25 + Vector の builtin 融合がベータ版で出ているので、
Composite(fusion="rrf")
のTiDB 側 push-downとして実装 - mem9 統合— TiDB Vector の上に mem9 系列の操作を載せると、メモリ層の expressiveness がさらに上がりそう
- TiDB Vector Quantization— 1536 dim → 512 dim 圧縮でストレージ 1/3 + 速度向上(精度は若干トレードオフ)
- TiFlash 連携— 行ストア(エージェント書込み)と列ストア(分析クエリ)を同居させて HTAP らしさを引き出す
持ち帰れる実装知見のまとめ
本記事の核心は 「TiDB Vector を選んだからこそ取れた知見」 です。それを実装する過程で 汎用的に転用できる副産物的知見 も得られた、という構造を以下に整理します。
(A) TiDB Vector を使ったから取れた知見
これが TiDB を選んだ価値そのもの です。他のベクトル DB / RDB 単独構成ではこの形では取れません。
| 知見 | TiDB が解決した課題 |
|---|---|
が SELECT … WHERE user_id=? ORDER BY VEC_COSINE_DISTANCE(embedding, ?) LIMIT ?1 本の SQL で書ける |
Pinecone + RDB 構成だと「ベクトル DB に問合せ → ID 一覧 → RDB へ JOIN」と 2 ラウンドトリップ。整合性とレイテンシが二重に劣化 |
| VECTOR(N) + HNSW + JSON + B-Tree インデックス が同一テーブル内 に同居 | スキーマ migration が 1 系統で完結、削除時の「テキストは消えたがベクトルが残る」状態が原理的に発生しない |
MySQL ワイヤープロトコル互換 で pymysql一つで接続 |
既存の DBA / 監視 / バックアップ運用がそのまま流用可能。新しいクライアントライブラリの学習・運用ノウハウ蓄積が不要 |
| 同じテーブルに BI ツールが SQL で直接 SELECT | DWH へのエクスポート/同期が不要。エージェントが書いた直後のデータが即分析対象に |
| Cloud Serverless 無料 25 GiB → 本番までスキーマ連続 | 検証用クラスタと本番クラスタで「型が違う」「機能差がある」が発生しない。スケール時のリスクが低い |
| HTAP(TiKV 行ストア + TiFlash 列ストア) で書込みと分析クエリが同居 | エージェント記憶の高頻度書込み中でも、大規模 GROUP BY 集計が干渉しにくい |
| VECTOR(N) 型の JSON-array 文字列シリアライズ という具体的な値表現ノウハウ | pymysql から INSERT INTO ... VALUES (..., '[0.123, ...]')の形で渡す必要がある、という TiDB 固有の実装手順 |
DSN ホスト名 パターンtidbcloud.comベースの TLS 自動有効化 |
Cloud / Self-hosted の混在運用で接続コード分岐を避けられる |
(B) 副産物的に得られた汎用パターン
TiDB 実装の過程で抽出できた、他システム(LangChain / LlamaIndex / pgvector / 自前 RAG など)にも応用できる パターン。コンテストテーマの「他エンジニアに有益」軸での価値です:
| パターン | 適用範囲 |
|---|---|
| 4 メソッド Protocol で記憶層を抽象化 | あらゆる Vector DB 切替に有効。後日の TiDB 移行・撤退コストも下がる |
| callable embedder 注入 | OpenAI / sentence-transformers / Voyage / 自前モデルを 1 行で切替 |
| 動的テーブル名のホワイトリスト検証 | あらゆる SQL DB で必要(プレースホルダが効かない箇所の標準対策) |
repr()ベースのベクトル文字列化 |
float 精度問題に当たる全シーン(独自シリアライズフォーマット全般) |
| Deterministic fake embedder でベンチ分離 | RAG / メモリ層ベンチ全般。embedding API レイテンシを DB 性能から切り離す方法論 |
| 接続セットアップ ≒ 律速 という観察 | Serverless DB 全般(Neon / PlanetScale 等でも同じ構造)。pool 化が次の自然な一手 |
(A) が TiDB を選んだ核心的価値、(B) は TiDB を実装したからこそ言語化できた汎用パターン という関係です。両方を持ち帰れる形にしておくのが、TiDB × エージェント記憶の実装記事として一番役に立つかな、と判断しました。
マルチエージェント基盤の長期記憶を「1 つの DB」に寄せる意味
Praxia は「個人 → 組織への暗黙知昇格」というミッションを掲げます。シニアの頭の中にしかない試行錯誤の文脈を、退職と同時に蒸発させずに組織知へ転写する — これは本質的に データの長期保存 に依存した取り組みです。
その長期保存層を、ベクトル DB + RDB + JSON ストア の三層構成で組むと運用負債が指数的に増えます。TiDB Vector はそれを 1 つの DB に統合する数少ない選択肢 で、しかも MySQL ワイヤープロトコル互換なので既存の DBA / 監視 / バックアップ運用がそのまま使えます。
Pinecone / Weaviate / Qdrant は素晴らしいベクトル DB ですが、「全社の SQL 基盤を新しいベクトル DB に移行する」は現実的にほぼ不可能。一方「既存の MySQL 互換クラスタを TiDB に置き換える / 横置きする」は、SQL の方言は MySQL のまま、
VECTOR型が追加されるだけなので学習コストが極めて低い。
組織側のデータ基盤に寄り添える backend を持っているか — これがマルチエージェント基盤が エンタープライズで採用されるか否か を分ける、と感じています。Praxia × TiDB の組み合わせは、その問いへの 1 つの具体的な回答です。
現状ステータス
本記事の TiDB backend 実装は、Praxia の
MemoryBackendプロトコルを満たす形で 検証用リポジトリ として公開中です。entry-point ベースのプラグイン機構(
praxia.memory_backends)が用意されているので、ベンチ結果と運用シナリオを踏まえつつ、 praxia-tidb パッケージとして PyPI へ正式リリース していくロードマップ。
⭐ Praxia 本体への Star + PR + Issue 大歓迎: github.com/praxia-dev/praxia
🎬 60 秒デモ: https://youtu.be/o_6NbjJU1AA
📦 PyPI:
pip install praxia
関連リソース
- TiDB Vector docs— VECTOR 型と HNSW の正式仕様
- Praxia 5 層メモリスタック解説(全体記事)— 春に始めた話
- praxia/memory/backends/base.py— 4 メソッド Protocol(本記事で実装した protocol)