Function direction_add

scripts/drift_engine.py:1082–1121

鏂板鏍稿績鏂瑰悜锛氬湪鍘熸湁鏍稿績鏂瑰悜鍩虹涓婃坊鍔犳柊鐨勭爺绌舵柟鍚?

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

def direction_add(
    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", {})

    old_directions = list(core_directions.keys())
    new_directions_pool = params.get("new_directions_pool", [])
    add_count = params.get("add_count", 1)
    initial_weight = params.get("initial_weight", 0.4)

    # 杩囨护鎺夊凡鏈夌殑鏂瑰悜
    available_directions = [d for d in new_directions_pool if d not in core_directions]

    # 闅忔満閫夋嫨鏂版柟鍚?
    new_directions = random.sample(available_directions, min(add_count, len(available_directions)))

    # 娣诲姞鏂版柟鍚?
    for direction in new_directions:
        core_directions[direction] = initial_weight

    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_add",
        "directions_before": old_directions,
        "directions_added": new_directions,
        "directions_after": list(core_directions.keys()),
    }

    return updated, drift_event


def direction_remove(
    profile: Dict[str, Any],