refactor: minor update

docs/vector_store_api_guide.md

@@ -116,7 +116,7 @@ The asynchronous interface is particularly useful in the following scenarios:
 - **password**: Database password (seekdb Docker images commonly set this via `ROOT_PASSWORD`).
 - **database**: Logical database name (default: `test`).
 - **index_type**: Vector index family (default: `HNSW`).
-- **index_metric**: Distance metric for the vector index: `cosine`, `l2`, or `ip` (inner product); default `cosine`.
+- **index_metric**: Distance metric for the vector index: `cosine` or `ip` (inner product); default `cosine`.
 - **index_ef_search**: HNSW `ef_search` parameter passed to pyobvector (default: `100`).
 - **collection_name**: Table name for the collection (from `VectorStoreConfig`, default `reme`). Use lowercase names if your deployment restricts identifiers.
 
@@ -132,8 +132,6 @@ docker compose -f docker-compose.obvec.yml up -d
 OBVEC_PASSWORD=<your_root_password> python tests/test_vector_store.py --obvec
 ```
 
-**Dependencies**: `pyobvector` is declared in ReMe’s `pyproject.toml`. A compatible `sqlglot` range is pinned so the pyobvector client imports cleanly.
-
 ## Configuration File Examples
 
 Configure Vector Store in `flowllm/config/default.yaml` under the `vector_store` section. The basic structure is as follows:

reme/core/vector_store/obvec_vector_store.py

@@ -12,27 +12,39 @@ from pathlib import Path
 from typing import Any
 
 from loguru import logger
-from sqlalchemy import text as sa_text
+from sqlalchemy import Column, JSON, String, text as sa_text
+from sqlalchemy.dialects.mysql import LONGTEXT
 
 from .base_vector_store import BaseVectorStore
 from ..embedding import BaseEmbeddingModel
 from ..schema import VectorNode
 
 _OBVECTOR_IMPORT_ERROR: Exception | None = None
+_DISTANCE_BY_METRIC: dict[str, Callable[..., Any]] | None = None
 
 try:
     from pyobvector import IndexParams, ObVecClient, VecIndexType, VECTOR
+    from pyobvector import cosine_distance, inner_product
 except Exception as e:
     _OBVECTOR_IMPORT_ERROR = e
-    IndexParams = None
-    ObVecClient = None
-    VecIndexType = None
-    VECTOR = None
+    IndexParams = None  # type: ignore[misc, assignment]
+    ObVecClient = None  # type: ignore[misc, assignment]
+    VecIndexType = None  # type: ignore[misc, assignment]
+    VECTOR = None  # type: ignore[misc, assignment]
+else:
+    _DISTANCE_BY_METRIC = {
+        "cosine": cosine_distance,
+        "ip": inner_product,
+    }
+
+# ann_search with ``with_dist=True`` yields id, content, metadata, distance.
+_ANN_ROW_MIN_COLUMNS = 4
+
+_COL_SELECT = "id, content, vector, metadata"
 
 # HNSW index metric strings for pyobvector ``IndexParam`` (metric_type / distance).
 _METRIC_TO_INDEX_DISTANCE: dict[str, str] = {
     "cosine": "cosine",
-    "l2": "l2_distance",
     "ip": "inner_product",
 }
 
@@ -111,20 +123,15 @@ def _normalize_embedding_for_ann(raw: Any) -> list[float]:
     return [float(x) for x in raw]
 
 
-def _get_distance_function(metric: str) -> Callable[..., Any]:
-    from pyobvector import cosine_distance, inner_product, l2_distance
-
-    registry: dict[str, Callable[..., Any]] = {
-        "cosine": cosine_distance,
-        "l2": l2_distance,
-        "ip": inner_product,
-    }
-    return registry.get(metric.lower(), cosine_distance)
+def _distance_function(metric: str) -> Callable[..., Any]:
+    if _DISTANCE_BY_METRIC is None:
+        raise RuntimeError("pyobvector distance functions are unavailable")
+    return _DISTANCE_BY_METRIC.get(metric.lower(), _DISTANCE_BY_METRIC["cosine"])
 
 
 def _similarity_from_distance(metric: str, distance: float) -> float:
     m = metric.lower()
-    if m in ("cosine", "l2"):
+    if m == "cosine":
         return max(0.0, 1.0 - distance / 2.0)
     return max(0.0, float(distance))
 
@@ -138,6 +145,22 @@ def _vector_node_from_db_row(row: tuple[Any, ...]) -> VectorNode:
     )
 
 
+def _normalize_nodes(nodes: VectorNode | list[VectorNode]) -> list[VectorNode]:
+    return [nodes] if isinstance(nodes, VectorNode) else list(nodes)
+
+
+def _search_result_metadata(metadata_raw: Any, score: float, distance: Any) -> dict[str, Any]:
+    meta = _coerce_db_metadata(metadata_raw) if metadata_raw is not None else {}
+    meta["score"] = score
+    meta["_score"] = score
+    meta["_distance"] = distance
+    return meta
+
+
+def _sql_table(name: str) -> str:
+    return f"`{name}`"
+
+
 class ObVecVectorStore(BaseVectorStore):
     """OceanBase or seekdb vector store for dense vectors and kNN search."""
 
@@ -178,81 +201,83 @@ class ObVecVectorStore(BaseVectorStore):
         self.client: ObVecClient | None = None
         self.embedding_model_dims = embedding_model.dimensions
 
+    def _require_client(self) -> ObVecClient:
+        if self.client is None:
+            raise RuntimeError("ObVecVectorStore.start() must be called before this operation")
+        return self.client
+
+    async def _nodes_with_filled_embeddings(
+        self,
+        nodes: list[VectorNode],
+        embed_if: Callable[[VectorNode], bool],
+    ) -> list[VectorNode]:
+        need = [n for n in nodes if embed_if(n)]
+        if not need:
+            return nodes
+        filled = await self.get_node_embeddings(need)
+        by_id = {n.vector_id: n for n in filled}
+        return [by_id.get(n.vector_id, n) if embed_if(n) else n for n in nodes]
+
     async def list_collections(self) -> list[str]:
         if self.client is None:
             return []
         try:
-            result = self.client.perform_raw_text_sql(
-                f"SHOW TABLES FROM `{self.database}`",
-            )
+            result = self.client.perform_raw_text_sql(f"SHOW TABLES FROM {_sql_table(self.database)}")
             rows = result.fetchall()
             return [row[0] for row in rows if row]
         except Exception as e:
             logger.warning("Failed to list collections: {}", e)
             return []
 
-    async def create_collection(self, collection_name: str, **kwargs):
-        dimensions = kwargs.get("dimensions", self.embedding_model_dims)
-
-        if self.client is None:
-            logger.warning("Client not initialized, skipping collection creation")
-            return
-
-        if self.client.check_table_exists(collection_name):
-            logger.info("Collection {} already exists", collection_name)
-            return
-
-        from sqlalchemy import Column, JSON, String
-        from sqlalchemy.dialects.mysql import LONGTEXT
-
-        columns = [
+    def _table_columns_for_create(self, dimensions: int) -> list[Any]:
+        return [
             Column("id", String(255), primary_key=True),
             Column("content", LONGTEXT),
             Column("vector", VECTOR(dimensions)),
             Column("metadata", JSON),
         ]
 
-        vidxs: IndexParams | None = None
-        if self.index_type == "HNSW":
-            metric = _METRIC_TO_INDEX_DISTANCE.get(self.index_metric, "cosine")
-            vidxs = IndexParams()
-            vidxs.add_index(
-                "vector",
-                VecIndexType.HNSW,
-                f"{collection_name}_vidx",
-                metric_type=metric,
-                params={"efSearch": self.index_ef_search},
-            )
+    def _hnsw_index_params(self, collection_name: str) -> IndexParams | None:
+        if self.index_type != "HNSW":
+            return None
+        metric = _METRIC_TO_INDEX_DISTANCE.get(self.index_metric, "cosine")
+        vidxs = IndexParams()
+        vidxs.add_index(
+            "vector",
+            VecIndexType.HNSW,
+            f"{collection_name}_vidx",
+            metric_type=metric,
+            params={"efSearch": self.index_ef_search},
+        )
+        return vidxs
 
-        try:
-            self.client.create_table_with_index_params(
-                table_name=collection_name,
-                columns=columns,
-                vidxs=vidxs,
-            )
-            logger.info("Created collection {} with dimensions={}", collection_name, dimensions)
-        except Exception as e:
-            logger.error("Failed to create collection {}: {}", collection_name, e)
-            raise
+    async def create_collection(self, collection_name: str, **kwargs):
+        client = self._require_client()
+        dimensions = kwargs.get("dimensions", self.embedding_model_dims)
+
+        if client.check_table_exists(collection_name):
+            logger.info("Collection {} already exists", collection_name)
+            return
+
+        columns = self._table_columns_for_create(dimensions)
+        vidxs = self._hnsw_index_params(collection_name)
+
+        client.create_table_with_index_params(
+            table_name=collection_name,
+            columns=columns,
+            vidxs=vidxs,
+        )
+        logger.info("Created collection {} with dimensions={}", collection_name, dimensions)
 
     async def delete_collection(self, collection_name: str, **kwargs):
-        if self.client is None:
-            logger.warning("Client not initialized, skipping collection deletion")
-            return
-
-        try:
-            self.client.drop_table_if_exist(collection_name)
-            logger.info("Deleted collection {}", collection_name)
-        except Exception as e:
-            logger.error("Failed to delete collection {}: {}", collection_name, e)
-            raise
+        client = self._require_client()
+        client.drop_table_if_exist(collection_name)
+        logger.info("Deleted collection {}", collection_name)
 
     async def copy_collection(self, collection_name: str, **kwargs):
-        if self.client is None:
-            logger.warning("Client not initialized, skipping collection copy")
-            return
+        client = self._require_client()
 
-        if not self.client.check_table_exists(self.collection_name):
+        if not client.check_table_exists(self.collection_name):
             raise ValueError(f"Source collection {self.collection_name} does not exist")
 
         await self.create_collection(collection_name)
@@ -261,53 +286,38 @@ class ObVecVectorStore(BaseVectorStore):
             source_data = await self.list(limit=None)
             if source_data:
                 await self.insert(source_data, collection_name=collection_name)
-
             logger.info("Copied collection {} to {}", self.collection_name, collection_name)
         except Exception:
             try:
-                self.client.drop_table_if_exist(collection_name)
+                client.drop_table_if_exist(collection_name)
             except Exception as cleanup_err:
                 logger.warning("Cleanup after failed copy failed: {}", cleanup_err)
             raise
 
     async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs):
-        if isinstance(nodes, VectorNode):
-            nodes = [nodes]
-
+        nodes = _normalize_nodes(nodes)
         if not nodes:
             return
 
-        nodes_without_vectors = [node for node in nodes if node.vector is None]
-        if nodes_without_vectors:
-            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
-            vector_map = {n.vector_id: n for n in nodes_with_vectors}
-            nodes_to_insert = [
-                vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes
-            ]
-        else:
-            nodes_to_insert = nodes
+        client = self._require_client()
 
-        data = []
-        for node in nodes_to_insert:
-            vector_list = node.vector if node.vector is not None else []
-            data.append({
+        nodes_to_insert = await self._nodes_with_filled_embeddings(
+            nodes,
+            embed_if=lambda n: n.vector is None,
+        )
+
+        data = [
+            {
                 "id": node.vector_id,
                 "content": node.content,
-                "vector": vector_list,
+                "vector": node.vector if node.vector is not None else [],
                 "metadata": node.metadata if node.metadata else {},
-            })
-
-        target_collection = kwargs.get("collection_name", self.collection_name)
-
-        try:
-            self.client.insert(
-                table_name=target_collection,
-                data=data,
-            )
-            logger.info("Inserted {} documents into {}", len(nodes_to_insert), target_collection)
-        except Exception as e:
-            logger.error("Failed to insert documents: {}", e)
-            raise
+            }
+            for node in nodes_to_insert
+        ]
+        target = kwargs.get("collection_name", self.collection_name)
+        client.insert(table_name=target, data=data)
+        logger.info("Inserted {} documents into {}", len(nodes_to_insert), target)
 
     async def search(
         self,
@@ -316,157 +326,125 @@ class ObVecVectorStore(BaseVectorStore):
         filters: dict | None = None,
         **kwargs,
     ) -> list[VectorNode]:
+        client = self._require_client()
         raw_vec = await self.get_embedding(query)
         query_vector = _normalize_embedding_for_ann(raw_vec)
-        distance_func = _get_distance_function(self.index_metric)
+        dist_fn = _distance_function(self.index_metric)
 
         filter_sql = _build_metadata_filter_sql(filters)
         where_parts = [sa_text(filter_sql)] if filter_sql else None
 
-        try:
-            results = self.client.ann_search(
-                table_name=self.collection_name,
-                vec_data=query_vector,
-                vec_column_name="vector",
-                distance_func=distance_func,
-                with_dist=True,
-                topk=limit,
-                output_column_names=["id", "content", "metadata"],
-                where_clause=where_parts,
-            )
+        results = client.ann_search(
+            table_name=self.collection_name,
+            vec_data=query_vector,
+            vec_column_name="vector",
+            distance_func=dist_fn,
+            with_dist=True,
+            topk=limit,
+            output_column_names=["id", "content", "metadata"],
+            where_clause=where_parts,
+        )
 
-            search_results: list[VectorNode] = []
-            score_threshold = kwargs.get("score_threshold")
-
-            for row in results:
-                if len(row) < 4:
-                    continue
-                vector_id, content, metadata_raw, distance = row[0], row[1], row[2], row[3]
-                score = _similarity_from_distance(self.index_metric, float(distance))
-
-                if score_threshold is not None and score < score_threshold:
-                    continue
-
-                metadata: dict[str, Any] = {}
-                if metadata_raw:
-                    metadata = _coerce_db_metadata(metadata_raw)
-                metadata["score"] = score
-                metadata["_distance"] = distance
-
-                search_results.append(
-                    VectorNode(
-                        vector_id=vector_id,
-                        content=content or "",
-                        vector=None,
-                        metadata=metadata,
-                    ),
+        score_threshold = kwargs.get("score_threshold")
+        out: list[VectorNode] = []
+        for row in results:
+            if len(row) < _ANN_ROW_MIN_COLUMNS:
+                logger.warning(
+                    "ann_search row has unexpected width: len={} (expected >= {})",
+                    len(row),
+                    _ANN_ROW_MIN_COLUMNS,
                 )
-
-            return search_results
-        except Exception as e:
-            logger.error("Search failed: {}", e)
-            return []
+                continue
+            vid, content, metadata_raw, distance = row[0], row[1], row[2], row[3]
+            score = _similarity_from_distance(self.index_metric, float(distance))
+            if score_threshold is not None and score < score_threshold:
+                continue
+            meta = _search_result_metadata(metadata_raw, score, distance)
+            out.append(
+                VectorNode(
+                    vector_id=vid,
+                    content=content or "",
+                    vector=None,
+                    metadata=meta,
+                ),
+            )
+        return out
 
     async def delete(self, vector_ids: str | list[str], **kwargs):
         if isinstance(vector_ids, str):
             vector_ids = [vector_ids]
-
         if not vector_ids:
             return
-
-        try:
-            self.client.delete(self.collection_name, ids=vector_ids)
-            logger.info("Deleted {} documents from {}", len(vector_ids), self.collection_name)
-        except Exception as e:
-            logger.error("Failed to delete documents: {}", e)
-            raise
+        client = self._require_client()
+        client.delete(self.collection_name, ids=vector_ids)
+        logger.info("Deleted {} documents from {}", len(vector_ids), self.collection_name)
 
     async def delete_all(self, **kwargs):
-        try:
-            self.client.delete(self.collection_name)
-            logger.info("Deleted all documents from {}", self.collection_name)
-        except Exception as e:
-            logger.error("Failed to delete all documents: {}", e)
-            raise
+        client = self._require_client()
+        client.delete(self.collection_name)
+        logger.info("Deleted all documents from {}", self.collection_name)
 
     async def update(self, nodes: VectorNode | list[VectorNode], **kwargs):
-        if isinstance(nodes, VectorNode):
-            nodes = [nodes]
-
+        nodes = _normalize_nodes(nodes)
         if not nodes:
             return
 
-        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
-        if nodes_without_vectors:
-            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
-            vector_map = {n.vector_id: n for n in nodes_with_vectors}
-            nodes_to_update = [
-                vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes
-            ]
-        else:
-            nodes_to_update = nodes
+        client = self._require_client()
+        nodes_to_update = await self._nodes_with_filled_embeddings(
+            nodes,
+            embed_if=lambda n: n.vector is None and bool(n.content),
+        )
 
-        try:
-            from sqlalchemy import text
+        for node in nodes_to_update:
+            updates: list[str] = []
+            params: dict[str, Any] = {}
 
-            for node in nodes_to_update:
-                updates: list[str] = []
-                params: dict[str, Any] = {}
+            if node.content is not None:
+                updates.append("content = :content")
+                params["content"] = node.content
 
-                if node.content is not None:
-                    updates.append("content = :content")
-                    params["content"] = node.content
+            if node.vector is not None:
+                updates.append("vector = :vector")
+                params["vector"] = _format_vector_sql_literal(node.vector)
 
-                if node.vector is not None:
-                    updates.append("vector = :vector")
-                    params["vector"] = _format_vector_sql_literal(node.vector)
+            if node.metadata is not None:
+                updates.append("metadata = :metadata")
+                params["metadata"] = json.dumps(node.metadata)
 
-                if node.metadata is not None:
-                    updates.append("metadata = :metadata")
-                    params["metadata"] = json.dumps(node.metadata)
+            if not updates:
+                continue
 
-                if not updates:
-                    continue
+            params["vid"] = node.vector_id
+            update_sql = f"UPDATE {_sql_table(self.collection_name)} SET {', '.join(updates)} WHERE id = :vid"
+            with client.engine.connect() as conn:
+                with conn.begin():
+                    conn.execute(sa_text(update_sql), params)
 
-                params["vid"] = node.vector_id
-                update_sql = (
-                    f"UPDATE `{self.collection_name}` SET {', '.join(updates)} WHERE id = :vid"
-                )
-                with self.client.engine.connect() as conn:
-                    with conn.begin():
-                        conn.execute(text(update_sql), params)
-
-            logger.info("Updated {} documents in {}", len(nodes_to_update), self.collection_name)
-        except Exception as e:
-            logger.error("Failed to update documents: {}", e)
-            raise
+        logger.info("Updated {} documents in {}", len(nodes_to_update), self.collection_name)
 
     async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode] | None:
-        single_result = isinstance(vector_ids, str)
-        if single_result:
+        single = isinstance(vector_ids, str)
+        if single:
             vector_ids = [vector_ids]
-
         if not vector_ids:
-            return [] if not single_result else None
+            return [] if not single else None
 
+        client = self._require_client()
         try:
             ids_str = "', '".join(vector_ids)
             select_sql = (
-                f"SELECT id, content, vector, metadata FROM `{self.collection_name}` "
+                f"SELECT {_COL_SELECT} FROM {_sql_table(self.collection_name)} "
                 f"WHERE id IN ('{ids_str}')"
             )
-
-            result = self.client.perform_raw_text_sql(select_sql)
+            result = client.perform_raw_text_sql(select_sql)
             rows = result.fetchall()
-
-            results = [_vector_node_from_db_row(row) for row in rows if row]
-
-            if single_result:
-                return results[0] if results else None
-            return results
+            parsed = [_vector_node_from_db_row(row) for row in rows if row]
+            if single:
+                return parsed[0] if parsed else None
+            return parsed
         except Exception as e:
             logger.error("Failed to get documents: {}", e)
-            return [] if not single_result else None
+            return [] if not single else None
 
     async def list(
         self,
@@ -475,39 +453,32 @@ class ObVecVectorStore(BaseVectorStore):
         sort_key: str | None = None,
         reverse: bool = False,
     ) -> list[VectorNode]:
+        client = self._require_client()
         try:
-            select_sql = f"SELECT id, content, vector, metadata FROM `{self.collection_name}`"
-
+            select_sql = f"SELECT {_COL_SELECT} FROM {_sql_table(self.collection_name)}"
             where_clause = _build_metadata_filter_sql(filters)
             if where_clause:
                 select_sql += f" WHERE {where_clause}"
-
             if sort_key and _is_safe_metadata_key(sort_key):
                 order = "DESC" if reverse else "ASC"
                 select_sql += f" ORDER BY JSON_EXTRACT(metadata, '$.{sort_key}') {order}"
-
             if limit is not None:
                 select_sql += f" LIMIT {limit}"
-
-            result = self.client.perform_raw_text_sql(select_sql)
+            result = client.perform_raw_text_sql(select_sql)
             rows = result.fetchall()
-
             return [_vector_node_from_db_row(row) for row in rows if row]
         except Exception as e:
             logger.error("Failed to list documents: {}", e)
             return []
 
     async def collection_info(self) -> dict[str, Any]:
+        client = self._require_client()
         try:
-            count_sql = f"SELECT COUNT(*) FROM `{self.collection_name}`"
-            result = self.client.perform_raw_text_sql(count_sql)
+            count_sql = f"SELECT COUNT(*) FROM {_sql_table(self.collection_name)}"
+            result = client.perform_raw_text_sql(count_sql)
             row = result.fetchone()
             count = row[0] if row else 0
-
-            return {
-                "name": self.collection_name,
-                "count": count,
-            }
+            return {"name": self.collection_name, "count": count}
         except Exception as e:
             logger.error("Failed to get collection info: {}", e)
             return {"name": self.collection_name, "count": 0}
@@ -531,8 +502,8 @@ class ObVecVectorStore(BaseVectorStore):
         )
 
         await super().start()
-        logger.info("OceanBase collection {} initialized", self.collection_name)
+        logger.info("seekdb / OceanBase vector table {} ready", self.collection_name)
 
     async def close(self):
         self.client = None
-        logger.info("OceanBase client connection closed")
+        logger.info("ObVec client connection closed")

tests/test_vector_store.py

@@ -2,8 +2,8 @@
 """Unified test suite for vector store implementations.
 
 This module provides comprehensive test coverage for LocalVectorStore, ESVectorStore,
-PGVectorStore, QdrantVectorStore, and ChromaVectorStore implementations. Tests can be
-run for specific vector stores or all implementations.
+PGVectorStore, QdrantVectorStore, ChromaVectorStore, and ObVecVectorStore implementations.
+Tests can be run for specific vector stores or all implementations.
 
 Usage:
     python test_vector_store.py --local      # Test LocalVectorStore only
@@ -11,9 +11,8 @@ Usage:
     python test_vector_store.py --pgvector   # Test PGVectorStore only
     python test_vector_store.py --qdrant     # Test QdrantVectorStore only
     python test_vector_store.py --chroma     # Test ChromaVectorStore only
-    python test_vector_store.py --obvec      # Test ObVecVectorStore only
+    python test_vector_store.py --obvec      # Test ObVecVectorStore only (needs seekdb / OceanBase)
     python test_vector_store.py --all        # Test all vector stores
-
 """
 
 import argparse
@@ -41,6 +40,12 @@ from reme.core.vector_store import (
 
 load_env()
 
+
+def _search_score_for_log(metadata: dict) -> object:
+    """Similarity score for log lines (implementations use ``metadata['score']``)."""
+    return metadata.get("score", metadata.get("_score", "N/A"))
+
+
 # ==================== Configuration ====================
 
 
@@ -76,8 +81,10 @@ class TestConfig:
     CHROMA_TENANT = None  # Set for ChromaDB Cloud tenant
     CHROMA_DATABASE = None  # Set for ChromaDB Cloud database
 
-    # ObVecVectorStore: seekdb uses user `root`; OceanBase multi-tenant often uses `root@test`.
-    # Defaults match docker-compose.obvec.yml (ROOT_PASSWORD=root).
+    # ObVecVectorStore: seekdb docker often uses user `root` + ROOT_PASSWORD; OceanBase
+    # multi-tenant commonly uses `root@<tenant>` (see pyobvector defaults).
+    # OBVEC_PASSWORD default `root` matches docker-compose.obvec.yml only—override if your
+    # seekdb uses another ROOT_PASSWORD (e.g. another compose stack on the same port).
     OBVEC_URI = os.environ.get("OBVEC_URI", "127.0.0.1:2881")
     OBVEC_USER = os.environ.get("OBVEC_USER", "root")
     OBVEC_PASSWORD = os.environ.get("OBVEC_PASSWORD", "root")
@@ -351,7 +358,7 @@ async def test_search(store: BaseVectorStore, _store_name: str):
 
     logger.info(f"Search returned {len(results)} results")
     for i, r in enumerate(results, 1):
-        score = r.metadata.get("_score", "N/A")
+        score = _search_score_for_log(r.metadata)
         logger.info(f"  Result {i}: {r.content[:60]}... (score: {score})")
 
     assert len(results) > 0, "Search should return results"
@@ -1034,7 +1041,7 @@ async def test_search_relevance_ranking(store: BaseVectorStore, _store_name: str
 
     logger.info(f"Search results for: '{query}'")
     for i, result in enumerate(results, 1):
-        score = result.metadata.get("_score", "N/A")
+        score = _search_score_for_log(result.metadata)
         relevance = result.metadata.get("relevance", "unknown")
         logger.info(f"  {i}. [{relevance}] score={score}: {result.content[:60]}...")
 
@@ -1052,7 +1059,7 @@ async def test_search_relevance_ranking(store: BaseVectorStore, _store_name: str
     results2 = await store.search(query=query2, limit=5)
     logger.info(f"\nSearch results for: '{query2}'")
     for i, result in enumerate(results2, 1):
-        score = result.metadata.get("_score", "N/A")
+        score = _search_score_for_log(result.metadata)
         logger.info(f"  {i}. score={score}: {result.content[:60]}...")
 
     logger.info("✓ Search relevance ranking test passed")
@@ -1742,7 +1749,7 @@ async def cleanup_store(store: BaseVectorStore, store_type: str):
 
     Args:
         store: Vector store instance
-        store_type: Type of vector store ("local" or "es")
+        store_type: Backend key (e.g. ``"local"``, ``"obvec"``)
     """
     logger.info("=" * 20 + " CLEANUP " + "=" * 20)
 
@@ -1776,6 +1783,13 @@ async def cleanup_store(store: BaseVectorStore, store_type: str):
                 shutil.rmtree(test_dir)
                 logger.info(f"Cleaned up chroma directory: {config.CHROMA_PATH}")
 
+        # ObVecVectorStore uses a temp db_path per run (reserved for local sidecar files).
+        if store_type == "obvec":
+            obvec_dir = getattr(store, "db_path", None)
+            if obvec_dir and Path(obvec_dir).exists():
+                shutil.rmtree(obvec_dir, ignore_errors=True)
+                logger.info(f"Cleaned up obvec temp directory: {obvec_dir}")
+
         logger.info("✓ Cleanup completed")
     except Exception as e:
         logger.error(f"Cleanup error: {e}")
@@ -1796,6 +1810,7 @@ Examples:
   python test_vector_store.py --pgvector   # Test PGVectorStore only
   python test_vector_store.py --qdrant     # Test QdrantVectorStore only
   python test_vector_store.py --chroma     # Test ChromaVectorStore only
+  python test_vector_store.py --obvec      # Test ObVecVectorStore (seekdb / OceanBase)
   python test_vector_store.py --all        # Test all vector stores
         """,
     )