Function direction_remove

scripts/drift_engine.py:1124–1175

绉婚櫎鏃ф柟鍚戯細绉婚櫎鏉冮噸浣庝簬闃堝€肩殑琛伴€€鏍稿績鏂瑰悜

(
    profile: Dict[str, Any],
    params: Dict[str, Any],
    selected_papers: Optional[List[Dict]] = None,
    date: Optional[str] = None,
)

def direction_remove(
    profile: Dict[str, Any],
    params: Dict[str, Any],
    selected_papers: Optional[List[Dict]] = None,
    date: Optional[str] = None,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    绉婚櫎鏃ф柟鍚戯細绉婚櫎鏉冮噸浣庝簬闃堝€肩殑琛伴€€鏍稿績鏂瑰悜
    """
    updated = copy.deepcopy(profile)
    core_directions = updated.get("core_directions", {})

    weight_threshold = params.get("weight_threshold", 0.3)
    max_remove_count = params.get("max_remove_count", 2)

    old_directions = list(core_directions.keys())

    # 鎵惧嚭鏉冮噸浣庝簬闃堝€肩殑鏂瑰悜
    low_weight_directions = [
        (k, v) for k, v in core_directions.items() if v < weight_threshold
    ]

    # 鎸夋潈閲嶆帓搴忥紝绉婚櫎鏈€浣庣殑
    low_weight_directions.sort(key=lambda x: x[1])
    to_remove = [k for k, v in low_weight_directions[:max_remove_count]]

    # 绉婚櫎鏂瑰悜
    for direction in to_remove:
        del core_directions[direction]

    # 褰掍竴鍖?
    if params.get("renormalize", True) and core_directions:
        max_score = max(core_directions.values())
        if max_score > 0:
            for k in core_directions:
                core_directions[k] = min(1.0, core_directions[k] / max_score)

    updated["core_directions"] = core_directions
    updated["topic_weights"] = copy.deepcopy(core_directions)

    # 鏇存柊 drift_state
    updated["drift_state"] = _update_drift_state(profile, date, score_delta=0.2)

    drift_event = {
        "method": "direction_remove",
        "directions_before": old_directions,
        "directions_removed": to_remove,
        "directions_after": list(core_directions.keys()),
        "reason": f"weight < {weight_threshold}",
    }

    return updated, drift_event


# ==================== 杈呭姪鍑芥暟 ====================