Function ensure_profile_shape

agents/coldstart-agent/main.py:151–186

Normalize a stored profile into the current schema.

(profile: Optional[Dict[str, Any]], user_id: str)

def ensure_profile_shape(profile: Optional[Dict[str, Any]], user_id: str) -> Dict[str, Any]:
    """Normalize a stored profile into the current schema."""
    normalized = copy.deepcopy(profile) if profile else build_empty_profile(user_id)
    normalized["user_id"] = user_id
    normalized.setdefault("version", "0.1")
    normalized.setdefault("created_at", normalized.get("updated_at", datetime.now().isoformat()))
    normalized["updated_at"] = datetime.now().isoformat()

    for key in ("core_directions", "methodology_preferences", "topic_weights", "author_heat", "institution_heat", "taste_profile"):
        value = normalized.get(key)
        normalized[key] = value if isinstance(value, dict) else {}

    must_read = normalized.get("must_read")
    normalized["must_read"] = must_read if isinstance(must_read, dict) else {}
    for key in ("authors", "institutions", "keywords"):
        value = normalized["must_read"].get(key)
        normalized["must_read"][key] = value if isinstance(value, list) else []

    for key in ("interest_vector", "reading_history", "behavior_logs"):
        value = normalized.get(key)
        normalized[key] = value if isinstance(value, list) else []

    drift_state = normalized.get("drift_state")
    normalized["drift_state"] = drift_state if isinstance(drift_state, dict) else {}
    for key, default_value in build_default_drift_state(normalized["updated_at"]).items():
        normalized["drift_state"].setdefault(key, copy.deepcopy(default_value))

    raw_core_directions = dict(normalized.get("core_directions") or {})
    raw_topic_weights = dict(normalized.get("topic_weights") or {})
    normalized["core_directions"] = _normalize_direction_weight_map(raw_core_directions)
    normalized["topic_weights"] = _normalize_direction_weight_map(raw_topic_weights)

    if normalized["core_directions"] and normalized["core_directions"] != raw_core_directions:
        normalized["interest_vector"] = generate_interest_vector(normalized["core_directions"])

    return normalized


def _normalize_direction_weight_map(raw_weights: Any) -> Dict[str, float]: