Function topic_weight_decay

scripts/drift_engine.py:923–983

涓婚鏉冮噸婕傜Щ锛氭棫鏍稿績鏂瑰悜鏉冮噸琛板噺锛屾柊鍏存柟鍚戞潈閲嶆彁鍗? Args: profile: 鐢ㄦ埛鐢诲儚 params: checkfile 涓殑鍙傛暟 selected_papers: 鐢ㄦ埛閫変腑鐨勮鏂囷紙鐢ㄤ簬妫€娴嬫柊鍏翠富棰橈級 Returns: (鏇存柊鍚庣殑 profile, drift_event 鍏冩暟鎹?

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

def topic_weight_decay(
    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]]:
    """
    涓婚鏉冮噸婕傜Щ锛氭棫鏍稿績鏂瑰悜鏉冮噸琛板噺锛屾柊鍏存柟鍚戞潈閲嶆彁鍗?
    Args:
        profile: 鐢ㄦ埛鐢诲儚
        params: checkfile 涓殑鍙傛暟
        selected_papers: 鐢ㄦ埛閫変腑鐨勮鏂囷紙鐢ㄤ簬妫€娴嬫柊鍏翠富棰橈級

    Returns:
        (鏇存柊鍚庣殑 profile, drift_event 鍏冩暟鎹?
    """
    updated = copy.deepcopy(profile)
    core_directions = updated.get("core_directions", {})

    # 1. 鏃ф柟鍚戣“鍑?
    for direction in list(core_directions.keys()):
        core_directions[direction] *= params.get("decay_factor", 0.7)

    # 2. 浠庤鏂囦腑鎻愬彇鏂板叴涓婚
    emerging_topics = _extract_emerging_topics(selected_papers, params)

    # 3. 鏂版柟鍚戞彁鍗?娣诲姞
    for topic in emerging_topics:
        if topic in core_directions:
            core_directions[topic] *= params.get("boost_factor", 1.5)
        else:
            core_directions[topic] = params.get("min_emerging_score", 0.4)

    # 4. 褰掍竴鍖栧埌 [0, 1]
    max_score = max(core_directions.values()) if core_directions else 1.0
    if max_score > 0:
        for k in core_directions:
            core_directions[k] = min(1.0, core_directions[k] / max_score)

    # 5. 闄愬埗鏂瑰悜鏁伴噺
    if len(core_directions) > params.get("max_topics", 5):
        sorted_dirs = sorted(core_directions.items(), key=lambda x: x[1], reverse=True)
        core_directions = dict(sorted_dirs[: params["max_topics"]])

    updated["core_directions"] = core_directions

    # 6. 鍚屾鏇存柊 topic_weights
    updated["topic_weights"] = copy.deepcopy(core_directions)

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

    drift_event = {
        "method": "topic_weight_decay",
        "emerging_topics": emerging_topics,
        "old_directions_count": len(profile.get("core_directions", {})),
        "new_directions_count": len(core_directions),
        "directions_added": [t for t in emerging_topics if t not in profile.get("core_directions", {})],
    }