Function vector_search_nodes

skills/wiki-store/scripts/wiki_db.py:581–628

(
    user_id: str,
    query: str,
    node_type: Optional[str] = None,
    limit: int = 20,
)

def vector_search_nodes(
    user_id: str,
    query: str,
    node_type: Optional[str] = None,
    limit: int = 20,
) -> List[Dict[str, Any]]:
    init_wiki_schema()
    conn = db_ops.get_connection()
    params: List[Any] = [user_id]
    type_filter = ""
    if node_type:
        type_filter = " AND node_type = ?"
        params.append(node_type)
    rows = conn.execute(
        f"""
        SELECT * FROM wiki_nodes
        WHERE user_id = ?
          AND embedding IS NOT NULL
          {type_filter}
        """,
        params,
    ).fetchall()
    conn.close()
    if not rows:
        return []

    from paperflow.providers import build_embedding_provider

    provider = build_embedding_provider()
    query_vector = np.array(provider.embed(query), dtype=np.float32)
    query_norm = float(np.linalg.norm(query_vector))
    if query_norm > 0:
        query_vector = query_vector / query_norm

    scored: List[Dict[str, Any]] = []
    for row in rows:
        node = _row_to_node(row)
        vector = _blob_to_vector(row["embedding"])
        if vector.size == 0:
            continue
        dim = min(vector.size, query_vector.size)
        if dim <= 0:
            continue
        score = float(np.dot(vector[:dim], query_vector[:dim]))
        node["vector_score"] = score
        scored.append(node)
    scored.sort(key=lambda item: float(item.get("vector_score", 0.0)), reverse=True)
    return scored[: max(1, int(limit))]


def search_nodes(