Function search_nodes

skills/wiki-store/scripts/wiki_db.py:631–671

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

def search_nodes(
    user_id: str,
    query: str,
    node_type: Optional[str] = None,
    limit: int = 20,
) -> List[Dict[str, Any]]:
    init_wiki_schema()
    query_text = (query or "").strip()
    if not query_text:
        return list_nodes(user_id, node_type=node_type, limit=limit)

    keyword_hits = _keyword_search_nodes(user_id, query_text, node_type, max(limit, 30))
    try:
        vector_hits = vector_search_nodes(user_id, query_text, node_type=node_type, limit=max(limit, 30))
    except Exception:
        vector_hits = []
    merged: Dict[str, Dict[str, Any]] = {}

    for node in keyword_hits:
        merged[node["node_id"]] = node
    vector_scores = [float(node.get("vector_score", 0.0)) for node in vector_hits]
    vector_min = min(vector_scores) if vector_scores else 0.0
    vector_max = max(vector_scores) if vector_scores else 1.0
    vector_range = max(vector_max - vector_min, 1e-6)
    for node in vector_hits:
        existing = merged.setdefault(node["node_id"], node)
        existing["vector_score"] = float(node.get("vector_score", 0.0))
        existing["vector_score_norm"] = (existing["vector_score"] - vector_min) / vector_range

    for node in merged.values():
        keyword_score = float(node.get("keyword_score", 0.0))
        vector_score = float(node.get("vector_score_norm", 0.0))
        interaction_bonus = _interaction_bonus(user_id, node)
        type_weight = {"section": 1.0, "paper": 0.9, "trajectory": 0.8, "topic": 0.75}.get(
            str(node.get("node_type")),
            0.7,
        )
        node["score"] = (0.45 * keyword_score + 0.55 * vector_score + interaction_bonus) * type_weight

    results = sorted(merged.values(), key=lambda item: float(item.get("score", 0.0)), reverse=True)
    return results[: max(1, int(limit))]


def _interaction_bonus(user_id: str, node: Dict[str, Any]) -> float: