feat: add OceanBase/seekdb vector and graph storage providers

[Evenss]

Description

Add two optional storage providers for OceanBase / seekdb, reusing the existing BaseVecDB / BaseGraphDB contracts. No default behavior changes: unless a user selects the oceanbase / seekdb backend, everything works exactly as before.

• OceanBaseVecDB (vec_dbs/oceanbase.py): built on pyseekdb's Collection / vector API, serving General Memory. Config now enforces a positive vector_dimension.
• OceanBaseGraphDB (graph_dbs/oceanbase.py): ported from the PostgreSQL + pgvector backend (nodes + edges + JSON properties + VECTOR column) over the MySQL-compatible protocol. Includes a thread-safe connection pool (maxconn now effective), atomic multi-step deletes, and identifier whitelisting for table_prefix / search_filter keys.
• Register oceanbase / seekdb aliases in the vector & graph factories and config factories; add the GraphDBError exception.

Dependencies: adds an optional extra ob-mem (containing pyseekdb); imports are guarded with try/except ImportError and are not added to core dependencies.

Related Issue (Required): Fixes #2109

Type of change

• New feature (non-breaking change which adds functionality)

How Has This Been Tested?

• Unit Test

.venv/bin/python -m pytest tests/vec_dbs/test_oceanbase.py tests/graph_dbs/test_oceanbase.py -q
# 31 passed

.venv/bin/ruff check src/memos/vec_dbs/oceanbase.py src/memos/graph_dbs/oceanbase.py
.venv/bin/ruff format --check src/memos/vec_dbs/oceanbase.py src/memos/graph_dbs/oceanbase.py
# All checks passed

External seekdb / OceanBase drivers (pyseekdb / pymysql) are stubbed with mocks; no live server is required. A live-server smoke test is not included in this PR.

Checklist

• I have performed a self-review of my own code | 我已自行检查了自己的代码
• I have commented my code in hard-to-understand areas | 我已在难以理解的地方对代码进行了注释
• I have added tests that prove my fix is effective or that my feature works | 我已添加测试以证明我的修复有效或功能正常
• I have created related documentation issue/PR in MemOS-Docs (if applicable) | 我已在 MemOS-Docs 中创建了相关的文档 issue/PR（如果适用）
• I have linked the issue to this PR (if applicable) | 我已将 issue 链接到此 PR（如果适用）
• I have mentioned the person who will review this PR | 我已提及将审查此 PR 的人

Reviewer Checklist

• closes feat: support OceanBase for relational, vector, and graph storage #2109
• Made sure Checks passed
• Tests have been provided

[Memtensor-AI]

🤖 Open Code Review

Target: PR #2130
Task: 9ed8ee6946107e51
Base: main
Head: main

🔍 OpenCodeReview found 12 issue(s) in this PR.

⚠️ 1 warning(s) occurred during review.

---

1. pyproject.toml (L148)

After discarding a dead connection, the second self._pool.acquire() has no retry limit or timeout. If the database is unreachable, threads pile up waiting for a pool slot that never returns, eventually exhausting the pool under sustained connectivity loss.

---

2. src/memos/api/config.py (L926)

Both get_oceanbase_config and get_postgres_config read from the same EMBEDDING_DIMENSION env var, but with different fallback defaults ("1024" here vs "384" in postgres). In a deployment that runs both backends simultaneously, a single env var silently governs both with conflicting assumed defaults, making it easy to misconfigure one silently.

Consider a dedicated env var such as OCEANBASE_EMBEDDING_DIMENSION (falling back to EMBEDDING_DIMENSION if absent), or at least document the shared variable clearly.

---

3. src/memos/api/handlers/config_builders.py (L44-L45)

All config methods in graph_db_backend_map are called eagerly on every invocation, regardless of which backend is actually selected. If GRAPH_DB_BACKEND is "oceanbase" or "seekdb" but none of the OCEANBASE_* env vars are set, this silently uses defaults — which is fine. However, if GRAPH_DB_BACKEND is set to a backend not in this dict (e.g. a future backend), graph_db_backend_map[graph_db_backend] on line 55 raises an unhandled KeyError at startup.

More pressing: every call to build_graph_db_config() also calls APIConfig.get_polardb_config(), get_postgres_config(), etc., even when they are unused. For the existing backends this is cheap, but it establishes a pattern where any config method that performs I/O or side effects would fire unnecessarily.

Suggestion: look up the backend string first, then call only the relevant config method — or use a lambda/dict-of-callables pattern (consistent with how other backends already work in this file).

---

4. src/memos/graph_dbs/oceanbase.py (L64-L75)

TOCTOU race: _idle.empty() is checked under the lock, but _idle.get() runs outside it. Another thread can drain the queue between the check and the get, causing _idle.get() to block indefinitely even when _maxconn is already reached (no new connection will be created because _created was not incremented).

Suggestion: use get_nowait() with a queue.Empty fallback to create a new connection, keeping _created accounting correct.

💡 Suggested Change

Before:

        with self._lock:
            should_create = self._created < self._maxconn and self._idle.empty()
            if should_create:
                self._created += 1
        if should_create:
            try:
                return self._connect_fn()
            except Exception:
                with self._lock:
                    self._created -= 1
                raise
        return self._idle.get()

After:

        with self._lock:
            should_create = self._created < self._maxconn and self._idle.empty()
            if should_create:
                self._created += 1
        if should_create:
            try:
                return self._connect_fn()
            except Exception:
                with self._lock:
                    self._created -= 1
                raise
        try:
            return self._idle.get_nowait()
        except queue.Empty:
            # Lost the race: another thread took the last idle connection after our
            # check. Create a new one if capacity allows, otherwise block.
            with self._lock:
                if self._created < self._maxconn:
                    self._created += 1
                    should_create = True
                else:
                    should_create = False
            if should_create:
                try:
                    return self._connect_fn()
                except Exception:
                    with self._lock:
                        self._created -= 1
                    raise
            return self._idle.get()

---

5. src/memos/graph_dbs/oceanbase.py (L264-L266)

_execute already operates under autocommit=True (set in _conn_kwargs), so this conn.commit() is redundant. Wrapping it in suppress(Exception) makes it worse: if commit ever does fail (e.g. a future caller switches to manual-commit mode), the error is silently swallowed and the caller falsely believes the write succeeded. Remove the commit here; explicit commit/rollback belongs only in _transaction().

💡 Suggested Change

Before:

                with suppress(Exception):
                    conn.commit()
                return rowcount

After:

                return rowcount

---

6. src/memos/graph_dbs/oceanbase.py (L381-L392)

if embedding: evaluates to False for a valid all-zeros vector [0.0, 0.0, ...], silently inserting the node without its embedding. Use if embedding is not None: instead.

💡 Suggested Change

Before:

        if embedding:
            self._execute(
                f"""
                INSERT INTO {self.nodes_table}
                    (id, memory, properties, embedding, user_name, created_at, updated_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                    memory = VALUES(memory),
                    properties = VALUES(properties),
                    embedding = VALUES(embedding),
                    updated_at = VALUES(updated_at)
                """,

After:

        if embedding is not None:
            self._execute(
                f"""
                INSERT INTO {self.nodes_table}
                    (id, memory, properties, embedding, user_name, created_at, updated_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                    memory = VALUES(memory),
                    properties = VALUES(properties),
                    embedding = VALUES(embedding),
                    updated_at = VALUES(updated_at)
                """,

---

7. src/memos/graph_dbs/oceanbase.py (L446-L452)

Same falsy-vector bug as in add_node: if embedding: skips the embedding update for an all-zeros vector. Use if embedding is not None:.

💡 Suggested Change

Before:

        if embedding:
            self._execute(
                f"""
                UPDATE {self.nodes_table}
                SET memory = %s, properties = %s, embedding = %s, updated_at = NOW()
                WHERE id = %s AND user_name = %s
                """,

After:

        if embedding is not None:
            self._execute(
                f"""
                UPDATE {self.nodes_table}
                SET memory = %s, properties = %s, embedding = %s, updated_at = NOW()
                WHERE id = %s AND user_name = %s
                """,

---

8. src/memos/graph_dbs/oceanbase.py (L116-L118)

if embedding and isinstance(embedding, list): skips float-conversion for an all-zeros embedding list [0.0, ...] because an empty list is also falsy. Use if embedding is not None and isinstance(embedding, list): to catch the all-zeros case (the isinstance guard already rejects None and non-list types, but the leading truthiness check incorrectly excludes [] and zero-filled vectors).

💡 Suggested Change

Before:

    embedding = metadata.get("embedding")
    if embedding and isinstance(embedding, list):
        metadata["embedding"] = [float(x) for x in embedding]

After:

    embedding = metadata.get("embedding")
    if embedding is not None and isinstance(embedding, list):
        metadata["embedding"] = [float(x) for x in embedding]

---

9. src/memos/graph_dbs/oceanbase.py (L738-L744)

Method name delete_node_by_prams is a misspelling of params. This is a new public method (not inherited from BaseGraphDB), so it can be renamed without breaking the abstract interface. All callers of this method should also be updated.

💡 Suggested Change

Before:

    def delete_node_by_prams(
        self,
        writable_cube_ids: list[str] | None = None,
        memory_ids: list[str] | None = None,
        file_ids: list[str] | None = None,
        filter: dict | None = None,
    ) -> int:

After:

    def delete_node_by_params(
        self,
        writable_cube_ids: list[str] | None = None,
        memory_ids: list[str] | None = None,
        file_ids: list[str] | None = None,
        filter: dict | None = None,
    ) -> int:

---

10. src/memos/vec_dbs/oceanbase.py (L148-L149)

Bug: flat vector passed instead of a list of query vectors.

query_embeddings=query_vector passes a flat list[float], but the chromadb-style API expects list[list[float]] (a batch of query vectors). The response is already unpacked with [0] (e.g. response.get("ids")[0]), confirming the intent was to wrap the vector in a list. Passing a flat list will likely cause a shape/dimension mismatch error or incorrect results.

Fix:

query_embeddings=[query_vector],

---

11. src/memos/vec_dbs/oceanbase.py (L287-L295)

Bug: scalar id string passed instead of a list.

The chromadb-style collection.update() API requires ids to be a list[str]. Passing a bare string causes the client to iterate over the individual characters of the ID, resulting in wrong behavior or an error.

Fix:

self.collection.update(
    ids=[id],
    embeddings=item.vector,
    documents=self._document_of(item.payload),
    metadatas=metadata,
)

---

12. src/memos/vec_dbs/oceanbase.py (L238)

Performance: fetching full items (including embeddings) just to count them.

get_by_filter retrieves all matching records including their embeddings and metadata, then len() is called on the result. For large collections this wastes significant memory and network bandwidth.

If pyseekdb supports a where parameter on count(), prefer:

return int(self.collection.count(where=filter))

Otherwise, fetch only IDs to avoid transferring embedding data:

response = self.collection.get(where=filter, include=[])
return len(response.get("ids") or [])

---

🧹 Filtered 18 low-confidence OCR finding(s) before posting/fix-loop (existing_code_mismatch: 18).

Generated by cloud-assistant via Open Code Review.

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (31/31 executed). memos_python_core/changed-repo-python: 31/31. Duration: 5s

Branch: main

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (31/31 executed). memos_python_core/changed-repo-python: 31/31. Duration: 6s

Branch: main

[wustzdy]

新增的db不符合

木项目的代码结构

[Evenss]

@wustzdy 感谢 review！想跟您确认一下「不符合本项目代码结构」具体指的是哪方面，以便我针对性调整：

本次改动的文件布局是参照现有 polardb / postgres 后端的接入方式来做的：

• 实现：src/memos/graph_dbs/oceanbase.py、src/memos/vec_dbs/oceanbase.py（与 polardb.py、postgres.py、milvus.py一样平铺在各自目录下）
• 配置：src/memos/configs/graph_db.py / vec_db.py 中新增 pydantic 配置并注册到 ConfigFactory
• 工厂：graph_dbs/factory.py / vec_dbs/factory.py 注册 oceanbase / seekdb 别名
• 依赖：pyseekdb 放在可选 extras ob-mem 中，import 有 try/except 保护，不影响核心依赖
• 测试：tests/graph_dbs/ 与 tests/vec_dbs/ 下，与源码目录一一对应

如果您指的是其他结构问题，麻烦指明一下期望的结构，我尽快调整，谢谢！

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (31/31 executed). memos_python_core/changed-repo-python: 31/31. Duration: 8s [advisory, non-gating] AI-generated tests on branch test/auto-gen-abad4fed10e4fccd-20260727164640: 181/186 passed, 5 failed — these do NOT affect the PR verdict; review the branch manually.

Branch: main

[Evenss]

@wustzdy PTAL

[WeiminLee]

Thanks for adding the OceanBase/seekdb backends. I don't think this is ready to merge yet; there are a few provider-contract and tenant-safety issues that need to be fixed first.

Blocking findings:

1. src/memos/graph_dbs/oceanbase.py:890 - search_by_embedding() accepts filter, knowledgebase_ids, and return_fields, but never applies or returns them. Existing graph backends use these parameters, and API context recall calls search_by_embedding(..., return_fields=["memory", "key", ...]); with this implementation OceanBase returns only id/score, so recall results are filtered out because memory is missing. Ignoring filter / knowledgebase_ids also breaks caller-provided metadata and multi-cube filtering.

2. src/memos/graph_dbs/oceanbase.py:669 - delete_node_by_prams() allows file_ids or filter deletion without any writable_cube_ids / tenant scope, then selects and deletes matching nodes from the whole shared table. The API model allows writable_cube_ids=None, and the Neo4j community backend explicitly rejects file_ids deletes without writable_cube_ids; OceanBase should either require the scope for these modes or default safely to the configured user_name, with regression tests.

3. src/memos/configs/vec_db.py:66 - the vector_dimension validator does not run when the inherited default None is omitted in Pydantic v2, so OceanBaseVecDBConfig can still be constructed without a positive dimension and only fail later inside HNSWConfiguration. Please make the field required or validate the default value at config-construction time.

Please also make sure the new OceanBase tests cover these contracts, especially return_fields, structured filter / knowledgebase_ids, and tenant-scoped deletes.

[Evenss]

Thanks for the review. All three blocking issues have been addressed:

1. search_by_embedding() now supports filter, knowledgebase_ids, and return_fields.
2. Delete operations are tenant-scoped, and deleting by file_ids requires writable_cube_ids.
3. vector_dimension is now required and must be a positive integer.

Regression tests have been added: all OceanBase tests pass, and Ruff checks pass. Please take another look when convenient. Thanks!

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (48/48 executed). memos_python_core/changed-repo-python: 48/48. Duration: 5s [advisory, non-gating] AI-generated tests on branch test/auto-gen-9ed8ee6946107e51-20260804173940: 185/199 passed, 14 failed — these do NOT affect the PR verdict; review the branch manually.

Branch: main