symfonic.memory.backends.mongodb

mongodb

MongoGraphBackend + MongoVectorBackend -- async MongoDB backends using motor.

Requires the mongodb optional dependency group: pip install symfonic-core[mongodb]

Both backends are natively async via motor's AsyncIOMotorClient. Tenant isolation is enforced via per-tenant collection namespacing (graph) and a tenant_id field filter (vectors).

MongoGraphBackend

MongoGraphBackend(
    uri: str,
    database: str = "symfonic",
    collection_prefix: str = "graph",
    server_selection_timeout_ms: int = 5000,
    connect_timeout_ms: int = 5000,
    socket_timeout_ms: int = 10000,
)

GraphBackend implementation backed by MongoDB via motor.

Tenant isolation: each tenant gets a dedicated collection pair. Collections: {prefix}_{tenant_id}_nodes, {prefix}_{tenant_id}_edges

The constructor is intentionally lazy -- no network I/O occurs at init time. Motor's AsyncIOMotorClient defers all server-selection to the first awaited operation, so placing it here is safe. Use ping() to verify connectivity from an async context before trusting the backend.

Source code in src/symfonic/memory/backends/mongodb.py

def __init__(
    self,
    uri: str,
    database: str = "symfonic",
    collection_prefix: str = "graph",
    server_selection_timeout_ms: int = 5000,
    connect_timeout_ms: int = 5000,
    socket_timeout_ms: int = 10000,
) -> None:
    self._client = motor.motor_asyncio.AsyncIOMotorClient(
        uri,
        serverSelectionTimeoutMS=server_selection_timeout_ms,
        connectTimeoutMS=connect_timeout_ms,
        socketTimeoutMS=socket_timeout_ms,
    )
    # v8.12.0: retain the connection URI so the engine can build a matching
    # MongoCheckpointerFactory (langgraph's MongoDBSaver needs its own sync
    # pymongo client from the same URI — see MongoCheckpointerFactory).
    self._uri = uri
    self._db = self._client[database]
    self._prefix = collection_prefix

add_edge async

add_edge(
    scope: TenantScope, edge: MemoryEdge
) -> MemoryEdge

Persist an edge document. Sets tenant_id from scope.

Source code in src/symfonic/memory/backends/mongodb.py

async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Persist an edge document. Sets tenant_id from scope."""
    edge.tenant_id = scope.tenant_id
    doc = _edge_to_doc(edge)
    await self._edges_col(scope).insert_one(doc)
    return edge

add_node async

add_node(
    scope: TenantScope, node: MemoryNode
) -> MemoryNode

Persist a node document. Sets tenant_id + materialises scope_path.

Source code in src/symfonic/memory/backends/mongodb.py

async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Persist a node document. Sets tenant_id + materialises scope_path."""
    node.tenant_id = scope.tenant_id
    node.properties[SCOPE_PATH_KEY] = materialise_scope_path(scope)
    doc = _node_to_doc(node)
    await self._nodes_col(scope).insert_one(doc)
    return node

delete_edge async

delete_edge(scope: TenantScope, edge_id: EdgeId) -> None

Delete a single edge by its ID.

Source code in src/symfonic/memory/backends/mongodb.py

async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    # v8.7.1: scope-gate the delete so a sibling sub-scope cannot delete
    # another sub-scope's edge (mirrors the node paths).
    await self._edges_col(scope).delete_one(
        {"_id": str(edge_id), **mongo_prefix_filter(scope)}
    )

delete_node async

delete_node(
    scope: TenantScope,
    node_id: NodeId,
    cascade: bool = False,
) -> None

Delete a node, optionally cascading to connected edges.

Source code in src/symfonic/memory/backends/mongodb.py

async def delete_node(
    self, scope: TenantScope, node_id: NodeId, cascade: bool = False
) -> None:
    """Delete a node, optionally cascading to connected edges."""
    # v8.7.1: carry the scope prefix filter (mirrors get_node/update_node)
    # so a sibling sub-scope cannot delete another sub-scope's node/edges.
    await self._nodes_col(scope).delete_one(
        {"_id": str(node_id), **mongo_prefix_filter(scope)}
    )
    if cascade:
        await self._edges_col(scope).delete_many(
            {"$and": [
                {"$or": [{"source": str(node_id)}, {"target": str(node_id)}]},
                mongo_prefix_filter(scope),
            ]}
        )

get_neighbors async

get_neighbors(
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]

Return nodes connected to node_id by edges, filtered by relationship.

Source code in src/symfonic/memory/backends/mongodb.py

async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]:
    """Return nodes connected to node_id by edges, filtered by relationship."""
    query: dict[str, Any] = {
        "$or": [{"source": str(node_id)}, {"target": str(node_id)}]
    }
    if relationship is not None:
        query["relationship"] = relationship

    neighbor_ids: set[str] = set()
    async for edge_doc in self._edges_col(scope).find(query):
        src, tgt = edge_doc["source"], edge_doc["target"]
        neighbor_ids.add(tgt if src == str(node_id) else src)

    if not neighbor_ids:
        return []

    nodes: list[MemoryNode] = []
    node_query = {
        "_id": {"$in": list(neighbor_ids)},
        **mongo_prefix_filter(scope),
    }
    async for doc in self._nodes_col(scope).find(node_query):
        nodes.append(_doc_to_node(doc))
    return nodes

get_node async

get_node(
    scope: TenantScope, node_id: NodeId
) -> MemoryNode | None

Fetch a node by ID within the scope's visible prefix chain.

Source code in src/symfonic/memory/backends/mongodb.py

async def get_node(self, scope: TenantScope, node_id: NodeId) -> MemoryNode | None:
    """Fetch a node by ID within the scope's visible prefix chain."""
    query = {"_id": str(node_id), **mongo_prefix_filter(scope)}
    doc = await self._nodes_col(scope).find_one(query)
    return _doc_to_node(doc) if doc else None

ping async

ping() -> bool

Return True if the MongoDB server is reachable, False otherwise.

Call this from an async context (e.g. FastAPI startup) to verify connectivity rather than discovering failures on the first data operation.

Source code in src/symfonic/memory/backends/mongodb.py

async def ping(self) -> bool:
    """Return True if the MongoDB server is reachable, False otherwise.

    Call this from an async context (e.g. FastAPI startup) to verify
    connectivity rather than discovering failures on the first data operation.
    """
    try:
        await self._db.command("ping")
        return True
    except Exception:  # noqa: BLE001
        return False

query_edges async

query_edges(
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]

Query edges from the tenant edge collection with pagination.

Mongo collections are already tenant-scoped, so no tenant_id filter is needed in the query document itself.

Source code in src/symfonic/memory/backends/mongodb.py

async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges from the tenant edge collection with pagination.

    Mongo collections are already tenant-scoped, so no tenant_id filter
    is needed in the query document itself.
    """
    query: dict[str, Any] = {}
    relationship = (filters or {}).get("relationship")
    if relationship is not None:
        query["relationship"] = relationship

    edges: list[MemoryEdge] = []
    cursor = (
        self._edges_col(scope)
        .find(query)
        .sort("created_at", -1)
        .skip(offset)
        .limit(limit)
    )
    async for doc in cursor:
        edges.append(_doc_to_edge(doc))
    return edges

query_nodes async

query_nodes(
    scope: TenantScope,
    filters: dict[str, Any],
    limit: int | None = 50,
) -> list[MemoryNode]

Filter nodes by arbitrary field criteria within the scope.

v8.0: ANDs the prefix-isolation filter (always-on) onto the field query. Combined under $and so a $or in either part never clobbers the other.

v8.7.2: limit=None returns all matching nodes (no truncation).

Source code in src/symfonic/memory/backends/mongodb.py

async def query_nodes(
    self,
    scope: TenantScope,
    filters: dict[str, Any],
    limit: int | None = 50,
) -> list[MemoryNode]:
    """Filter nodes by arbitrary field criteria within the scope.

    v8.0: ANDs the prefix-isolation filter (always-on) onto the field
    query.  Combined under ``$and`` so a ``$or`` in either part never
    clobbers the other.

    v8.7.2: ``limit=None`` returns all matching nodes (no truncation).
    """
    # v8.7.1: Mongo ``.limit(0)`` means "no limit" (the OPPOSITE of an
    # empty result), so limit==0 must short-circuit rather than reach
    # ``.limit()``.  None → 0 = unlimited.
    if limit == 0:
        return []
    field_query = _build_node_query(filters)
    prefix_query = mongo_prefix_filter(scope)
    query: dict[str, Any] = {"$and": [field_query, prefix_query]}
    cursor = self._nodes_col(scope).find(query).limit(
        limit if limit is not None else 0
    )
    nodes: list[MemoryNode] = []
    async for doc in cursor:
        nodes.append(_doc_to_node(doc))
    return nodes

traverse async

traverse(
    scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]

BFS traversal from start node up to max_depth hops.

Source code in src/symfonic/memory/backends/mongodb.py

async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS traversal from start node up to max_depth hops."""
    visited: set[str] = set()
    queue: deque[tuple[str, int]] = deque([(str(start), 0)])
    result: list[MemoryNode] = []

    while queue:
        current_id, depth = queue.popleft()
        if current_id in visited:
            continue
        visited.add(current_id)
        doc = await self._nodes_col(scope).find_one(
            {"_id": current_id, **mongo_prefix_filter(scope)}
        )
        if doc is not None:
            result.append(_doc_to_node(doc))
        if depth < max_depth:
            neighbors = await self.get_neighbors(scope, NodeId(current_id))
            for nbr in neighbors:
                nid = str(nbr.id)
                if nid not in visited:
                    queue.append((nid, depth + 1))
    return result

update_node async

update_node(
    scope: TenantScope,
    node_id: NodeId,
    updates: dict[str, Any],
) -> MemoryNode

Partial update of node fields. Raises KeyError if node not found.

Source code in src/symfonic/memory/backends/mongodb.py

async def update_node(
    self,
    scope: TenantScope,
    node_id: NodeId,
    updates: dict[str, Any],
) -> MemoryNode:
    """Partial update of node fields. Raises KeyError if node not found."""
    mongo_updates: dict[str, Any] = {}
    for key, value in updates.items():
        if key == "properties" and isinstance(value, dict):
            # v7.26.2 T0/Contract E: a top-level ``properties`` dict
            # full-replaces the properties subdocument, matching the
            # InMemory + Postgres backends (whose update_node loop sets
            # ``node.properties`` wholesale for this key).  This keeps
            # the read-modify-write path in
            # MemoryOrchestrator.upgrade_durability uniform across all
            # three backends.  Without this branch the key would route
            # to ``properties.properties`` (a nested clobber).
            mongo_updates["properties"] = value
        elif key in {
            "label", "layer", "importance", "access_count",
            "spreading_access_count",  # v6.2 T02
            "embedding",
        }:
            if key == "layer" and hasattr(value, "value"):
                value = value.value
            mongo_updates[key] = value
        else:
            mongo_updates[f"properties.{key}"] = value

    mongo_updates["updated_at"] = datetime.now(UTC).isoformat()

    result = await self._nodes_col(scope).find_one_and_update(
        # v8.7.1 H4: carry the scope prefix filter (mirrors get_node) so a
        # sibling sub-scope cannot overwrite another sub-scope's node.
        {"_id": str(node_id), **mongo_prefix_filter(scope)},
        {"$set": mongo_updates},
        return_document=True,
    )
    if result is None:
        raise KeyError(f"Node {node_id} not found for tenant {scope.tenant_id}")
    return _doc_to_node(result)

upsert_edge async

upsert_edge(
    scope: TenantScope, edge: MemoryEdge
) -> MemoryEdge

Insert edge or increment weight by 1 if (source, target, relationship) matches.

Uses MongoDB $inc with upsert=True so the operation is atomic. On insert the full document is written; on match only weight changes.

Source code in src/symfonic/memory/backends/mongodb.py

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if (source, target, relationship) matches.

    Uses MongoDB ``$inc`` with ``upsert=True`` so the operation is atomic.
    On insert the full document is written; on match only ``weight`` changes.
    """
    edge.tenant_id = scope.tenant_id
    filter_q = {
        "source": str(edge.source),
        "target": str(edge.target),
        "relationship": edge.relationship,
    }
    # '$inc' owns the 'weight' field, so it must NOT also appear in
    # '$setOnInsert' -- Mongo rejects updating the same path in two
    # operators ("conflict at 'weight'"). On insert, $inc from an absent
    # field yields weight=1; on match it increments. Round-2 fix.
    insert_doc = _edge_to_doc(edge)
    insert_doc.pop("weight", None)
    update_q = {
        "$inc": {"weight": 1},
        "$setOnInsert": insert_doc,
    }
    result = await self._edges_col(scope).find_one_and_update(
        filter_q,
        update_q,
        upsert=True,
        return_document=True,
    )
    return _doc_to_edge(result) if result else edge

MongoVectorBackend

MongoVectorBackend(
    uri: str,
    database: str = "symfonic",
    collection: str = "vectors",
    server_selection_timeout_ms: int = 5000,
    connect_timeout_ms: int = 5000,
    socket_timeout_ms: int = 10000,
    *,
    vector_search_mode: str = "python",
    atlas_index_name: str = "vector_index",
    atlas_num_candidates_multiplier: int = 10,
    atlas_num_candidates_min: int = 100,
)

VectorBackend backed by MongoDB.

Stores document vectors with tenant-scoped metadata. Similarity search loads all tenant vectors and computes cosine similarity in Python.

For production workloads, migrate to MongoDB Atlas Vector Search ($vectorSearch) which runs ANN search server-side.

The constructor is intentionally lazy -- no network I/O occurs at init time. Use ping() to verify connectivity from an async context.

Construct the MongoDB vector backend.

T-7.21.C1 -- adds Atlas $vectorSearch mode. The default vector_search_mode='python' keeps the v7.20 behaviour byte- identical: Python cosine similarity over all tenant vectors -- known-bad for large corpora, documented for prod replacement. vector_search_mode='atlas' switches the search method to issue a server-side ANN query via MongoDB Atlas's $vectorSearch aggregation stage.

Atlas-mode parameters:

• atlas_index_name: name of the Atlas vector-search index. Must be pre-created on the Atlas collection (or created via :meth:ensure_vector_search_index).
• atlas_num_candidates_multiplier: numCandidates is computed as max(atlas_num_candidates_min, multiplier * top_k). Atlas's docs recommend numCandidates >= 10*limit for good recall on production-sized corpora.
• atlas_num_candidates_min: floor for the numCandidates computation so small top_k values still query enough candidates to surface relevant hits.

Raises ValueError for unknown vector_search_mode values so misconfiguration fails fast at boot rather than silently falling through to the Python path.

Source code in src/symfonic/memory/backends/mongodb.py

def __init__(
    self,
    uri: str,
    database: str = "symfonic",
    collection: str = "vectors",
    server_selection_timeout_ms: int = 5000,
    connect_timeout_ms: int = 5000,
    socket_timeout_ms: int = 10000,
    *,
    vector_search_mode: str = "python",
    atlas_index_name: str = "vector_index",
    atlas_num_candidates_multiplier: int = 10,
    atlas_num_candidates_min: int = 100,
) -> None:
    """Construct the MongoDB vector backend.

    T-7.21.C1 -- adds Atlas $vectorSearch mode.  The default
    ``vector_search_mode='python'`` keeps the v7.20 behaviour byte-
    identical: Python cosine similarity over all tenant vectors --
    known-bad for large corpora, documented for prod replacement.
    ``vector_search_mode='atlas'`` switches the ``search`` method
    to issue a server-side ANN query via MongoDB Atlas's
    ``$vectorSearch`` aggregation stage.

    Atlas-mode parameters:

    - ``atlas_index_name``: name of the Atlas vector-search index.
      Must be pre-created on the Atlas collection (or created via
      :meth:`ensure_vector_search_index`).
    - ``atlas_num_candidates_multiplier``: numCandidates is computed
      as ``max(atlas_num_candidates_min, multiplier * top_k)``.
      Atlas's docs recommend numCandidates >= 10*limit for good
      recall on production-sized corpora.
    - ``atlas_num_candidates_min``: floor for the numCandidates
      computation so small top_k values still query enough
      candidates to surface relevant hits.

    Raises ``ValueError`` for unknown ``vector_search_mode`` values
    so misconfiguration fails fast at boot rather than silently
    falling through to the Python path.
    """
    if vector_search_mode not in ("python", "atlas"):
        raise ValueError(
            "vector_search_mode must be 'python' or 'atlas', "
            f"got {vector_search_mode!r}"
        )
    self._client = motor.motor_asyncio.AsyncIOMotorClient(
        uri,
        serverSelectionTimeoutMS=server_selection_timeout_ms,
        connectTimeoutMS=connect_timeout_ms,
        socketTimeoutMS=socket_timeout_ms,
    )
    self._db = self._client[database]
    self._col = self._db[collection]
    self._vector_search_mode = vector_search_mode
    self._atlas_index_name = atlas_index_name
    self._atlas_num_candidates_multiplier = atlas_num_candidates_multiplier
    self._atlas_num_candidates_min = atlas_num_candidates_min

add async

add(
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None

Upsert vector documents scoped to the tenant.

Source code in src/symfonic/memory/backends/mongodb.py

async def add(
    self,
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Upsert vector documents scoped to the tenant."""
    if not ids:
        return
    scope_path = materialise_scope_path(scope)
    docs = []
    for i, doc_id in enumerate(ids):
        meta = dict(metadatas[i]) if i < len(metadatas) else {}
        meta.setdefault(SCOPE_PATH_KEY, scope_path)
        docs.append({
            "_id": doc_id,
            "tenant_id": scope.tenant_id,
            "embedding": embeddings[i],
            "metadata": meta,
            "document": documents[i] if i < len(documents) else "",
            # v8.0: top-level scope_path for the $in prefix-isolation filter.
            "scope_path": scope_path,
        })
    await self._col.insert_many(docs)

count async

count(scope: TenantScope) -> int

Return the number of stored vectors visible to the tenant scope.

Without this method EpisodicLayer.count_events() falls back to 0, which silently disables every consolidation phase gated on episodic volume for Mongo-backed deployments. Round-2 fix.

Source code in src/symfonic/memory/backends/mongodb.py

async def count(self, scope: TenantScope) -> int:
    """Return the number of stored vectors visible to the tenant scope.

    Without this method EpisodicLayer.count_events() falls back to 0,
    which silently disables every consolidation phase gated on episodic
    volume for Mongo-backed deployments. Round-2 fix.
    """
    find_query = {
        "$and": [{"tenant_id": scope.tenant_id}, mongo_prefix_filter(scope)]
    }
    return int(await self._col.count_documents(find_query))

delete async

delete(scope: TenantScope, ids: list[str]) -> None

Delete vectors by ID, scoped to the tenant.

Source code in src/symfonic/memory/backends/mongodb.py

async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by ID, scoped to the tenant."""
    if not ids:
        return
    await self._col.delete_many(
        {"_id": {"$in": ids}, "tenant_id": scope.tenant_id}
    )

ensure_vector_search_index async

ensure_vector_search_index(
    *, dimensions: int, similarity: str = "cosine"
) -> None

Idempotent helper: create the Atlas vector-search index if missing.

T-7.21.C4 -- so adopters can bootstrap a fresh Atlas collection from code instead of hand-rolling the index in the Atlas UI. Reads the existing search-indexes via list_search_indexes() and short-circuits if the index named self._atlas_index_name already exists (idempotent semantics: safe to call on every boot).

The created index definition includes tenant_id as a filter field -- this is the index-side counterpart to the $vectorSearch.filter clause used by _search_atlas. Without registering tenant_id as a filter field, the Atlas filter clause silently turns into a no-op (the same class as the load-bearing $match defence-in-depth at _search_atlas); the helper here closes the index-side leg by stamping the filter dimension at create-time.

Parameters:

Name	Type	Description	Default
dimensions	int	embedding vector dimension (e.g. 1536 for OpenAI text-embedding-3-small). Atlas requires this at index-create time.	required
similarity	str	similarity function -- "cosine" (default), "euclidean", or "dotProduct". Match the embedding model's training objective for best recall.	'cosine'

Source code in src/symfonic/memory/backends/mongodb.py

async def ensure_vector_search_index(
    self,
    *,
    dimensions: int,
    similarity: str = "cosine",
) -> None:
    """Idempotent helper: create the Atlas vector-search index if missing.

    T-7.21.C4 -- so adopters can bootstrap a fresh Atlas collection
    from code instead of hand-rolling the index in the Atlas UI.
    Reads the existing search-indexes via ``list_search_indexes()``
    and short-circuits if the index named
    ``self._atlas_index_name`` already exists (idempotent semantics:
    safe to call on every boot).

    The created index definition includes ``tenant_id`` as a filter
    field -- this is the index-side counterpart to the
    ``$vectorSearch.filter`` clause used by ``_search_atlas``.
    Without registering tenant_id as a filter field, the Atlas
    ``filter`` clause silently turns into a no-op (the same class
    as the load-bearing $match defence-in-depth at
    ``_search_atlas``); the helper here closes the index-side leg
    by stamping the filter dimension at create-time.

    Args:
        dimensions: embedding vector dimension (e.g. 1536 for
            OpenAI text-embedding-3-small).  Atlas requires this
            at index-create time.
        similarity: similarity function -- ``"cosine"`` (default),
            ``"euclidean"``, or ``"dotProduct"``.  Match the
            embedding model's training objective for best recall.
    """
    async for entry in self._col.list_search_indexes():
        if entry.get("name") == self._atlas_index_name:
            return
    await self._col.create_search_index(
        {
            "name": self._atlas_index_name,
            "type": "vectorSearch",
            "definition": {
                "fields": [
                    {
                        "type": "vector",
                        "path": "embedding",
                        "numDimensions": dimensions,
                        "similarity": similarity,
                    },
                    # tenant_id as a filter field -- required for
                    # the $vectorSearch.filter clause in
                    # _search_atlas to actually do anything.
                    # Without it, the filter is a silent no-op
                    # (see Risk R-V7.21-C in plan §7).
                    {"type": "filter", "path": "tenant_id"},
                ],
            },
        },
    )

ping async

ping() -> bool

Return True if the MongoDB server is reachable, False otherwise.

Source code in src/symfonic/memory/backends/mongodb.py

async def ping(self) -> bool:
    """Return True if the MongoDB server is reachable, False otherwise."""
    try:
        await self._db.command("ping")
        return True
    except Exception:  # noqa: BLE001
        return False

search async

search(
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]

Dispatch to the configured vector-search backend.

T-7.21.C2 -- the body is split into mode-specific helpers so adopters can opt in to Atlas $vectorSearch by passing vector_search_mode='atlas' at construction time without touching this method. Default mode is byte-identical to v7.20.

Source code in src/symfonic/memory/backends/mongodb.py

async def search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]:
    """Dispatch to the configured vector-search backend.

    T-7.21.C2 -- the body is split into mode-specific helpers so
    adopters can opt in to Atlas $vectorSearch by passing
    ``vector_search_mode='atlas'`` at construction time without
    touching this method.  Default mode is byte-identical to v7.20.
    """
    if self._vector_search_mode == "atlas":
        return await self._search_atlas(scope, query_embedding, top_k)
    return await self._search_python(scope, query_embedding, top_k)