| from __future__ import annotations |
|
|
| import re |
| from typing import Any, Iterable, Mapping, Sequence |
|
|
|
|
| PROFILE_CATEGORIES = {"profile", "preference", "goal", "constraint", "stage_state", "status"} |
| PROFILE_TYPES = {"setup", "preference", "constraint", "goal", "avoid", "usage_context"} |
| PROFILE_SEMANTIC_SLOTS = {"identity", "research_topic", "education", "occupation"} |
| PROFILE_AGGREGATE_SOURCE_KIND = "public_dialog_profile" |
| PROFILE_AGGREGATE_CATEGORY = "profile" |
| PROFILE_CLUSTER_SOURCE_KIND = "public_dialog_profile_cluster" |
| PROFILE_CLUSTER_CATEGORY = "profile" |
| PROFILE_CONSOLIDATOR_VERSION = "profile_consolidator_v1_structured_summary" |
|
|
| _PROFILE_QUERY_MARKERS = ( |
| "preference", |
| "prefer", |
| "like", |
| "dislike", |
| "recommend", |
| "suggest", |
| "advice", |
| "advise", |
| "any advice", |
| "any suggestions", |
| "any tips", |
| "tips", |
| "trouble with", |
| "struggling with", |
| "what do you think", |
| "learn more", |
| "resources", |
| "suited", |
| "fit me", |
| "for me", |
| "based on me", |
| "based on my", |
| "my setup", |
| "my profile", |
| "my goal", |
| "my constraint", |
| "my occupation", |
| "my previous occupation", |
| "my role", |
| "my previous role", |
| "my job", |
| "my previous job", |
| "what was my", |
| "where did i work", |
| "worked as", |
| "occupation", |
| "previous occupation", |
| "role", |
| "previous role", |
| "job", |
| "career", |
| "background", |
| "identity", |
| "experience", |
| "avoid", |
| "what should i", |
| "should i", |
| "what should", |
| "serve", |
| "dinner", |
| "homegrown", |
| "ingredients", |
| "battery life", |
| "getting around", |
| "偏好", |
| "喜欢", |
| "不喜欢", |
| "推荐", |
| "建议", |
| "适合我", |
| "根据我", |
| "我的情况", |
| "我的配置", |
| "我的目标", |
| "我的约束", |
| "画像", |
| "避免", |
| ) |
|
|
|
|
| def _clean_text(value: Any) -> str: |
| return re.sub(r"\s+", " ", str(value or "").strip()) |
|
|
|
|
| def _normalize(value: Any) -> str: |
| return _clean_text(value).lower() |
|
|
|
|
| def _tokens(value: Any) -> list[str]: |
| text = _normalize(value) |
| english = re.findall(r"[a-z0-9_]+", text) |
| cjk = [char for char in text if "\u4e00" <= char <= "\u9fff"] |
| return _dedupe([*english, *cjk]) |
|
|
|
|
| def _slug(value: Any, *, fallback: str = "general") -> str: |
| text = _normalize(value) |
| parts = re.findall(r"[a-z0-9]+|[\u4e00-\u9fff]", text) |
| slug = "_".join(parts[:10]).strip("_") |
| return slug or fallback |
|
|
|
|
| def _dedupe(items: Iterable[Any], *, max_items: int | None = None) -> list[str]: |
| values: list[str] = [] |
| seen: set[str] = set() |
| for item in items: |
| text = _clean_text(item) |
| if not text: |
| continue |
| key = _normalize(text) |
| if key in seen: |
| continue |
| seen.add(key) |
| values.append(text) |
| if max_items is not None and len(values) >= max_items: |
| break |
| return values |
|
|
|
|
| def _first_nonempty(*values: Any) -> str: |
| for value in values: |
| text = _clean_text(value) |
| if text: |
| return text |
| return "" |
|
|
|
|
| def _bounded_union(*groups: Iterable[Any], max_items: int) -> list[str]: |
| return _dedupe([item for group in groups for item in group], max_items=max_items) |
|
|
|
|
| def _bounded_int_union(*groups: Iterable[Any], max_items: int) -> list[int]: |
| values: list[int] = [] |
| seen: set[int] = set() |
| for group in groups: |
| for item in group: |
| try: |
| value = int(item) |
| except Exception: |
| continue |
| if value in seen: |
| continue |
| seen.add(value) |
| values.append(value) |
| if len(values) >= max_items: |
| return sorted(values) |
| return sorted(values) |
|
|
|
|
| def is_profile_layer_record(*, category: Any, source_kind: Any = "", semantic_slot: Any = "", metadata: Mapping[str, Any] | None = None) -> bool: |
| data = dict(metadata or {}) |
| if bool(data.get("profile_layer")): |
| return True |
| category_text = _normalize(category) |
| slot_text = _normalize(semantic_slot or data.get("semantic_slot", "")) |
| source_text = _normalize(source_kind) |
| if category_text in PROFILE_CATEGORIES: |
| return True |
| if slot_text in PROFILE_SEMANTIC_SLOTS or slot_text.startswith("profile_"): |
| return True |
| return source_text in { |
| "public_dialog_profile", |
| "public_dialog_profile_cluster", |
| "public_dialog_preference", |
| "public_dialog_goal", |
| "public_dialog_constraint", |
| } |
|
|
|
|
| def infer_profile_type(*, category: Any, semantic_slot: Any = "", relation: Any = "", value: Any = "", metadata: Mapping[str, Any] | None = None) -> str: |
| data = dict(metadata or {}) |
| explicit = _normalize(data.get("profile_type", "")) |
| if explicit in PROFILE_TYPES: |
| return explicit |
| category_text = _normalize(category) |
| combined = _normalize(f"{semantic_slot} {relation} {value}") |
| if category_text == "preference" or any(marker in combined for marker in ("prefer", "like", "default", "偏好", "喜欢", "默认")): |
| return "preference" |
| if category_text == "constraint" or any(marker in combined for marker in ("must", "cannot", "forbid", "constraint", "必须", "不能", "约束")): |
| return "constraint" |
| if category_text == "goal" or any(marker in combined for marker in ("goal", "target", "objective", "目标")): |
| return "goal" |
| if any(marker in combined for marker in ("avoid", "dislike", "do not", "don't", "避免", "不喜欢", "不要")): |
| return "avoid" |
| if any(marker in combined for marker in ("setup", "environment", "workflow", "current", "配置", "环境", "流程", "当前")): |
| return "setup" |
| return "usage_context" |
|
|
|
|
| def infer_profile_domain( |
| *, |
| category: Any, |
| semantic_slot: Any = "", |
| slot_key: Any = "", |
| anchors: Sequence[Any] = (), |
| value: Any = "", |
| metadata: Mapping[str, Any] | None = None, |
| ) -> tuple[str, str]: |
| data = dict(metadata or {}) |
| explicit = _first_nonempty(data.get("profile_domain", ""), data.get("profile_domain_label", ""), data.get("domain", "")) |
| if explicit: |
| return _slug(explicit), explicit |
| subject = _first_nonempty(data.get("subject", ""), data.get("extracted_subject", "")) |
| if subject: |
| return _slug(subject), subject |
| slot = _normalize(semantic_slot) |
| if slot and slot not in {"profile", "preference", "goal", "constraint", "event", "fact", "status"}: |
| return _slug(slot), slot.replace("_", " ") |
| for anchor in anchors: |
| anchor_text = _clean_text(anchor) |
| if not anchor_text: |
| continue |
| if re.search(r"^\d{4}|\d{1,2}\s+[A-Za-z]+|^[A-Z][a-z]+$", anchor_text): |
| continue |
| return _slug(anchor_text), anchor_text |
| slot_key_text = _clean_text(slot_key) |
| if slot_key_text: |
| tail = slot_key_text.split(".")[-1].replace("_", " ") |
| if tail: |
| return _slug(tail), tail |
| value_tokens = _tokens(value) |
| if value_tokens: |
| label = " ".join(value_tokens[:4]) |
| return _slug(label), label |
| return _slug(category), _normalize(category) or "general" |
|
|
|
|
| def profile_candidate_metadata( |
| *, |
| category: Any, |
| semantic_slot: Any = "", |
| relation: Any = "", |
| value: Any = "", |
| source_span: Any = "", |
| slot_key: Any = "", |
| anchors: Sequence[Any] = (), |
| subject: Any = "", |
| subject_signature: Any = "", |
| proposal: Mapping[str, Any] | None = None, |
| ) -> dict[str, Any]: |
| proposal_data = dict(proposal or {}) |
| base_metadata = { |
| "profile_type": proposal_data.get("profile_type", ""), |
| "profile_domain": proposal_data.get("profile_domain", ""), |
| "profile_domain_label": proposal_data.get("profile_domain_label", ""), |
| "subject": subject, |
| "extracted_subject": proposal_data.get("extracted_subject", ""), |
| } |
| if not is_profile_layer_record(category=category, semantic_slot=semantic_slot, metadata=base_metadata): |
| return {} |
| profile_type = infer_profile_type( |
| category=category, |
| semantic_slot=semantic_slot, |
| relation=relation, |
| value=value or source_span, |
| metadata=proposal_data, |
| ) |
| domain, domain_label = infer_profile_domain( |
| category=category, |
| semantic_slot=semantic_slot, |
| slot_key=slot_key, |
| anchors=anchors, |
| value=value or source_span, |
| metadata={**proposal_data, "subject": subject}, |
| ) |
| normalized_subject_signature = _slug(subject_signature or subject or domain, fallback=domain) |
| route_terms = _dedupe( |
| [ |
| profile_type, |
| domain_label, |
| semantic_slot, |
| subject, |
| *list(anchors or []), |
| ], |
| max_items=12, |
| ) |
| return { |
| "profile_layer": True, |
| "profile_candidate_status": _clean_text(proposal_data.get("profile_candidate_status", "")) or "writer_candidate", |
| "profile_consolidation_stage": _clean_text(proposal_data.get("profile_consolidation_stage", "")) or "pre_consolidation", |
| "profile_type": profile_type, |
| "profile_domain": domain, |
| "profile_domain_label": domain_label, |
| "profile_subject_signature": normalized_subject_signature, |
| "profile_support_key": f"{profile_type}:{domain}:{normalized_subject_signature}", |
| "profile_route_terms": route_terms, |
| } |
|
|
|
|
| def profile_aggregate_slot_key(metadata: Mapping[str, Any]) -> str: |
| data = dict(metadata or {}) |
| support_key = _clean_text(data.get("profile_support_key", "")) |
| if support_key: |
| return f"tmcra.profile.aggregate.{_slug(support_key)}" |
| profile_type = _normalize(data.get("profile_type", "")) or "usage_context" |
| domain = _normalize(data.get("profile_domain", "")) or _slug(data.get("profile_domain_label", "general")) |
| subject = _normalize(data.get("profile_subject_signature", "")) or domain |
| return f"tmcra.profile.aggregate.{_slug(f'{profile_type}:{domain}:{subject}')}" |
|
|
|
|
| def profile_aggregate_value( |
| *, |
| profile_type: Any, |
| domain_label: Any, |
| support_values: Sequence[Any], |
| ) -> str: |
| typed = _clean_text(profile_type) or "usage_context" |
| domain = _clean_text(domain_label) or "general" |
| values = _dedupe(support_values, max_items=5) |
| if not values: |
| return f"User {typed} profile for {domain}." |
| return f"User {typed} profile for {domain}: " + "; ".join(values) |
|
|
|
|
| def _profile_output_kind(profile_types: Sequence[Any]) -> str: |
| normalized = {_normalize(item) for item in profile_types if _clean_text(item)} |
| if "constraint" in normalized: |
| return "constraint_profile" |
| if "goal" in normalized: |
| return "goal_profile" |
| if normalized.intersection({"preference", "avoid"}): |
| return "preference_profile" |
| if "setup" in normalized: |
| return "setup_profile" |
| return "usage_context_profile" |
|
|
|
|
| def _profile_update_policy(profile_types: Sequence[Any]) -> str: |
| normalized = {_normalize(item) for item in profile_types if _clean_text(item)} |
| if "constraint" in normalized: |
| return "preserve_until_explicitly_changed" |
| if normalized.intersection({"preference", "avoid", "goal", "setup"}): |
| return "update_on_newer_user_evidence" |
| return "background_context" |
|
|
|
|
| def _profile_memory_type(profile_types: Sequence[Any]) -> str: |
| normalized = {_normalize(item) for item in profile_types if _clean_text(item)} |
| if "constraint" in normalized: |
| return "hard_constraint" |
| if normalized.intersection({"preference", "avoid"}): |
| return "durable_preference" |
| return "profile_context" |
|
|
|
|
| def profile_summary( |
| *, |
| profile_types: Sequence[Any], |
| domain_label: Any, |
| support_values: Sequence[Any], |
| stage: str, |
| ) -> str: |
| kind = _profile_output_kind(profile_types) |
| domain = _clean_text(domain_label) or "general" |
| values = _dedupe(support_values, max_items=6 if stage == "cluster" else 4) |
| prefix = { |
| "constraint_profile": "User constraint profile", |
| "goal_profile": "User goal profile", |
| "preference_profile": "User preference profile", |
| "setup_profile": "User setup profile", |
| "usage_context_profile": "User usage-context profile", |
| }.get(kind, "User profile") |
| if not values: |
| return f"{prefix} for {domain}." |
| return f"{prefix} for {domain}: " + "; ".join(values) |
|
|
|
|
| def build_profile_aggregate_metadata( |
| *, |
| support_record_id: Any, |
| support_turn_index: int, |
| support_value: Any, |
| support_anchors: Sequence[Any], |
| support_metadata: Mapping[str, Any], |
| existing_metadata: Mapping[str, Any] | None = None, |
| ) -> dict[str, Any]: |
| support = dict(support_metadata or {}) |
| existing = dict(existing_metadata or {}) |
| profile_type = _normalize(support.get("profile_type", "")) or "usage_context" |
| domain = _normalize(support.get("profile_domain", "")) or _slug(support.get("profile_domain_label", "general")) |
| domain_label = _clean_text(support.get("profile_domain_label", "")) or domain.replace("_", " ") |
| subject_signature = _clean_text(support.get("profile_subject_signature", "")) or domain |
| support_ids = _bounded_union(existing.get("profile_support_ids", []) or [], [support_record_id], max_items=64) |
| support_turns = _bounded_int_union(existing.get("profile_support_turns", []) or [], [support_turn_index], max_items=64) |
| support_values = _bounded_union(existing.get("profile_support_values", []) or [], [support_value], max_items=12) |
| support_route_terms = _bounded_union( |
| existing.get("profile_route_terms", []) or [], |
| support.get("profile_route_terms", []) or [], |
| support_anchors, |
| max_items=24, |
| ) |
| profile_types = [profile_type] |
| output_kind = _profile_output_kind(profile_types) |
| memory_type = _profile_memory_type(profile_types) |
| summary = profile_summary( |
| profile_types=profile_types, |
| domain_label=domain_label, |
| support_values=support_values, |
| stage="aggregate", |
| ) |
| value = profile_aggregate_value( |
| profile_type=profile_type, |
| domain_label=domain_label, |
| support_values=support_values, |
| ) |
| return { |
| **existing, |
| "profile_layer": True, |
| "profile_candidate_status": "consolidated", |
| "profile_consolidation_stage": "aggregate", |
| "profile_consolidator_version": PROFILE_CONSOLIDATOR_VERSION, |
| "profile_aggregate_node": True, |
| "profile_type": profile_type, |
| "profile_domain": domain, |
| "profile_domain_label": domain_label, |
| "profile_subject_signature": subject_signature, |
| "profile_support_key": f"{profile_type}:{domain}:{subject_signature}", |
| "profile_support_ids": support_ids, |
| "profile_support_turns": support_turns, |
| "profile_support_values": support_values, |
| "profile_support_count": len(support_ids), |
| "profile_route_terms": support_route_terms, |
| "profile_value": value, |
| "profile_summary": summary, |
| "profile_output_kind": output_kind, |
| "profile_update_policy": _profile_update_policy(profile_types), |
| "profile_conflict_policy": "latest_active_support_only", |
| "profile_evidence_count": len(support_ids), |
| "memory_type": memory_type, |
| "durable_memory_type": memory_type, |
| "memory_chain_depth_layer": "profile", |
| "depth_layer": "profile", |
| } |
|
|
|
|
| _PROFILE_CLUSTER_STOPWORDS = { |
| "user", |
| "profile", |
| "preference", |
| "preferences", |
| "constraint", |
| "constraints", |
| "goal", |
| "goals", |
| "avoid", |
| "usage", |
| "context", |
| "general", |
| "default", |
| "should", |
| "would", |
| "could", |
| "want", |
| "wants", |
| "need", |
| "needs", |
| "when", |
| "then", |
| "than", |
| "with", |
| "that", |
| "this", |
| "from", |
| "into", |
| "instead", |
| } |
|
|
|
|
| def profile_cluster_tokens(*values: Any) -> list[str]: |
| tokens: list[str] = [] |
| for value in values: |
| if isinstance(value, (list, tuple, set)): |
| for item in value: |
| tokens.extend(_tokens(item)) |
| else: |
| tokens.extend(_tokens(value)) |
| filtered = [ |
| token |
| for token in tokens |
| if token |
| and token not in _PROFILE_CLUSTER_STOPWORDS |
| and not token.isdigit() |
| and (len(token) > 2 or any("\u4e00" <= char <= "\u9fff" for char in token)) |
| ] |
| return _dedupe(filtered, max_items=32) |
|
|
|
|
| def profile_cluster_similarity(source_metadata: Mapping[str, Any], target_metadata: Mapping[str, Any]) -> float: |
| source = dict(source_metadata or {}) |
| target = dict(target_metadata or {}) |
| source_tokens = set( |
| profile_cluster_tokens( |
| source.get("profile_domain_label", ""), |
| source.get("profile_domain", ""), |
| source.get("profile_route_terms", []) or [], |
| source.get("profile_support_values", []) or [], |
| source.get("profile_value", ""), |
| ) |
| ) |
| target_tokens = set( |
| profile_cluster_tokens( |
| target.get("profile_domain_label", ""), |
| target.get("profile_domain", ""), |
| target.get("profile_route_terms", []) or [], |
| target.get("profile_support_values", []) or [], |
| target.get("profile_value", ""), |
| ) |
| ) |
| if not source_tokens or not target_tokens: |
| return 0.0 |
| overlap = len(source_tokens & target_tokens) / max(1, len(source_tokens | target_tokens)) |
| containment = len(source_tokens & target_tokens) / max(1, min(len(source_tokens), len(target_tokens))) |
| type_bonus = 0.06 if _normalize(source.get("profile_type", "")) == _normalize(target.get("profile_type", "")) else 0.0 |
| return round(min(1.0, (0.62 * overlap) + (0.38 * containment) + type_bonus), 6) |
|
|
|
|
| def profile_cluster_slot_key(metadata: Mapping[str, Any]) -> str: |
| data = dict(metadata or {}) |
| tokens = profile_cluster_tokens( |
| data.get("profile_domain_label", ""), |
| data.get("profile_domain", ""), |
| data.get("profile_route_terms", []) or [], |
| data.get("profile_support_values", []) or [], |
| data.get("profile_value", ""), |
| ) |
| if tokens: |
| return f"tmcra.profile.cluster.{_slug('_'.join(tokens[:5]))}" |
| support_profiles = data.get("profile_support_profile_ids", []) or [] |
| support_seed = _clean_text(support_profiles[0] if support_profiles else "") |
| seed = support_seed or _clean_text(data.get("profile_support_key", "general")) |
| return f"tmcra.profile.cluster.{_slug(seed)}" |
|
|
|
|
| def profile_cluster_value(*, support_values: Sequence[Any]) -> str: |
| values = _dedupe(support_values, max_items=8) |
| if not values: |
| return "User profile cluster." |
| return "User profile cluster: " + "; ".join(values) |
|
|
|
|
| def build_profile_cluster_metadata( |
| *, |
| support_profile_id: Any, |
| support_metadata: Mapping[str, Any], |
| existing_metadata: Mapping[str, Any] | None = None, |
| ) -> dict[str, Any]: |
| support = dict(support_metadata or {}) |
| existing = dict(existing_metadata or {}) |
| support_profile_ids = _bounded_union( |
| existing.get("profile_support_profile_ids", []) or [], |
| [support_profile_id], |
| max_items=32, |
| ) |
| support_ids = _bounded_union( |
| existing.get("profile_support_ids", []) or [], |
| support.get("profile_support_ids", []) or [], |
| max_items=96, |
| ) |
| support_turns = _bounded_int_union( |
| existing.get("profile_support_turns", []) or [], |
| support.get("profile_support_turns", []) or [], |
| max_items=96, |
| ) |
| raw_support_values = support.get("profile_support_values", []) or [] |
| support_values = _bounded_union( |
| existing.get("profile_support_values", []) or [], |
| raw_support_values, |
| [] if raw_support_values else [support.get("profile_value", "")], |
| max_items=16, |
| ) |
| route_terms = _bounded_union( |
| existing.get("profile_route_terms", []) or [], |
| existing.get("profile_cluster_route_terms", []) or [], |
| support.get("profile_route_terms", []) or [], |
| profile_cluster_tokens( |
| support.get("profile_domain_label", ""), |
| support.get("profile_domain", ""), |
| support.get("profile_value", ""), |
| support.get("profile_support_values", []) or [], |
| ), |
| max_items=32, |
| ) |
| profile_types = _bounded_union( |
| existing.get("profile_cluster_types", []) or [], |
| [support.get("profile_type", "")], |
| max_items=8, |
| ) |
| domains = _bounded_union( |
| existing.get("profile_cluster_domains", []) or [], |
| [support.get("profile_domain", ""), support.get("profile_domain_label", "")], |
| max_items=16, |
| ) |
| primary_type = _clean_text(profile_types[0] if profile_types else support.get("profile_type", "")) or "usage_context" |
| primary_domain = _clean_text(domains[0] if domains else support.get("profile_domain", "")) or "general" |
| output_kind = _profile_output_kind(profile_types) |
| memory_type = _profile_memory_type(profile_types) |
| summary = profile_summary( |
| profile_types=profile_types, |
| domain_label=primary_domain, |
| support_values=support_values, |
| stage="cluster", |
| ) |
| value = profile_cluster_value(support_values=support_values) |
| return { |
| **existing, |
| "profile_layer": True, |
| "profile_candidate_status": "consolidated", |
| "profile_consolidation_stage": "cluster", |
| "profile_consolidator_version": PROFILE_CONSOLIDATOR_VERSION, |
| "profile_cluster_node": True, |
| "profile_type": primary_type, |
| "profile_domain": _slug(primary_domain), |
| "profile_domain_label": primary_domain.replace("_", " "), |
| "profile_subject_signature": _clean_text(existing.get("profile_subject_signature", "")) or _slug(primary_domain), |
| "profile_support_key": _clean_text(existing.get("profile_support_key", "")) or f"cluster:{_slug(primary_domain)}", |
| "profile_support_profile_ids": support_profile_ids, |
| "profile_support_ids": support_ids, |
| "profile_support_turns": support_turns, |
| "profile_support_values": support_values, |
| "profile_support_count": len(support_ids), |
| "profile_cluster_profile_count": len(support_profile_ids), |
| "profile_cluster_types": profile_types, |
| "profile_cluster_domains": domains, |
| "profile_cluster_route_terms": route_terms, |
| "profile_route_terms": route_terms, |
| "profile_value": value, |
| "profile_summary": summary, |
| "profile_output_kind": output_kind, |
| "profile_update_policy": _profile_update_policy(profile_types), |
| "profile_conflict_policy": "latest_active_support_only", |
| "profile_evidence_count": len(support_ids), |
| "memory_type": memory_type, |
| "durable_memory_type": memory_type, |
| "memory_chain_depth_layer": "profile", |
| "depth_layer": "profile", |
| } |
|
|
|
|
| def infer_profile_query_intent(query: Any) -> dict[str, Any]: |
| query_text = _clean_text(query) |
| lowered = _normalize(query_text) |
| enabled = any(marker in lowered for marker in _PROFILE_QUERY_MARKERS) |
| types: list[str] = [] |
| if any(marker in lowered for marker in ("prefer", "preference", "like", "dislike", "偏好", "喜欢", "不喜欢")): |
| types.append("preference") |
| if any(marker in lowered for marker in ("constraint", "must", "cannot", "policy", "约束", "限制", "必须", "不能")): |
| types.append("constraint") |
| if any(marker in lowered for marker in ("goal", "target", "objective", "目标")): |
| types.append("goal") |
| if any(marker in lowered for marker in ("avoid", "dislike", "避免", "不喜欢")): |
| types.append("avoid") |
| if any(marker in lowered for marker in ("setup", "profile", "current", "occupation", "role", "job", "career", "background", "identity", "experience", "配置", "画像", "当前")): |
| types.append("setup") |
| if enabled and any( |
| marker in lowered |
| for marker in ( |
| "recommend", |
| "suggest", |
| "advice", |
| "tips", |
| "trouble with", |
| "struggling with", |
| "what do you think", |
| "should i", |
| "what should", |
| "serve", |
| "dinner", |
| "battery life", |
| "getting around", |
| "resources", |
| "learn more", |
| "推荐", |
| "建议", |
| ) |
| ): |
| types.append("usage_context") |
| if not types and enabled: |
| types.append("usage_context") |
| return { |
| "enabled": enabled, |
| "types": _dedupe(types), |
| "tokens": _tokens(query_text), |
| } |
|
|
|
|
| def profile_query_score_delta( |
| *, |
| query: Any, |
| query_tokens: set[str], |
| category: Any, |
| source_kind: Any, |
| semantic_slot: Any, |
| value: Any, |
| anchors: Sequence[Any], |
| metadata: Mapping[str, Any] | None = None, |
| ) -> tuple[float, str]: |
| data = dict(metadata or {}) |
| intent = infer_profile_query_intent(query) |
| if not intent.get("enabled"): |
| return 0.0, "" |
| if not is_profile_layer_record(category=category, source_kind=source_kind, semantic_slot=semantic_slot, metadata=data): |
| return 0.0, "" |
| profile_type = _normalize(data.get("profile_type", "")) or infer_profile_type( |
| category=category, |
| semantic_slot=semantic_slot, |
| value=value, |
| metadata=data, |
| ) |
| domain = _normalize(data.get("profile_domain_label", "") or data.get("profile_domain", "")) |
| route_terms = " ".join(str(item) for item in data.get("profile_route_terms", []) or []) |
| record_tokens = set(_tokens(f"{profile_type} {domain} {semantic_slot} {value} {' '.join(str(item) for item in anchors)} {route_terms}")) |
| overlap = len(set(query_tokens) & record_tokens) / max(1, len(set(query_tokens) | record_tokens)) if query_tokens or record_tokens else 0.0 |
| type_match = profile_type in set(intent.get("types", []) or []) |
| delta = 0.20 + (0.16 if type_match else 0.0) + (0.18 * overlap) |
| if data.get("profile_candidate_status") == "consolidated": |
| delta += 0.06 |
| if data.get("profile_cluster_node") or data.get("profile_consolidation_stage") == "cluster": |
| delta += 0.08 |
| return round(min(delta, 0.54), 6), "profile_route" |
|
|
|
|
| def profile_edge_score(source_metadata: Mapping[str, Any], target_metadata: Mapping[str, Any], *, source_value: Any = "", target_value: Any = "") -> tuple[float, str]: |
| source = dict(source_metadata or {}) |
| target = dict(target_metadata or {}) |
| if not source.get("profile_layer") or not target.get("profile_layer"): |
| return 0.0, "" |
| source_key = _normalize(source.get("profile_support_key", "")) |
| target_key = _normalize(target.get("profile_support_key", "")) |
| source_domain = _normalize(source.get("profile_domain", "")) |
| target_domain = _normalize(target.get("profile_domain", "")) |
| source_type = _normalize(source.get("profile_type", "")) |
| target_type = _normalize(target.get("profile_type", "")) |
| source_tokens = set(_tokens(f"{source_value} {source.get('profile_domain_label', '')} {' '.join(source.get('profile_route_terms', []) or [])}")) |
| target_tokens = set(_tokens(f"{target_value} {target.get('profile_domain_label', '')} {' '.join(target.get('profile_route_terms', []) or [])}")) |
| overlap = len(source_tokens & target_tokens) / max(1, len(source_tokens | target_tokens)) if source_tokens or target_tokens else 0.0 |
| if source_key and source_key == target_key: |
| return round(0.72 + (0.14 * overlap), 6), "profile_support" |
| if source_domain and source_domain == target_domain: |
| return round(0.58 + (0.12 if source_type == target_type else 0.04) + (0.12 * overlap), 6), "profile_tunnel" |
| if source_type and source_type == target_type and overlap >= 0.22: |
| return round(0.44 + (0.18 * overlap), 6), "profile_soft_tunnel" |
| return 0.0, "" |
|
|
