Normalize one or more free-form topic texts through the shared direction layer.
(topic_texts: List[str], user_id: str)
| 638 | |
| 639 | |
| 640 | def normalize_profile_update_topics(topic_texts: List[str], user_id: str) -> Dict[str, Any]: |
| 641 | """Normalize one or more free-form topic texts through the shared direction layer.""" |
| 642 | llm_parser = importlib.import_module("agents.master-coordinator.scripts.llm_parser") |
| 643 | |
| 644 | canonical_directions: List[Dict[str, Any]] = [] |
| 645 | temporary_matches: List[Dict[str, Any]] = [] |
| 646 | pending_candidates: List[Dict[str, Any]] = [] |
| 647 | explanations: List[str] = [] |
| 648 | seen_canonical: set[str] = set() |
| 649 | seen_temporary: set[tuple[str, str]] = set() |
| 650 | seen_pending: set[str] = set() |
| 651 | profile = get_profile(user_id) or {} |
| 652 | |
| 653 | for topic_text in topic_texts: |
| 654 | cleaned = clean_profile_topic_text(topic_text) |
| 655 | if not cleaned: |
| 656 | continue |
| 657 | |
| 658 | profile_key = find_profile_topic_key(profile, cleaned) |
| 659 | if profile_key and profile_key not in seen_canonical: |
| 660 | seen_canonical.add(profile_key) |
| 661 | canonical_directions.append( |
| 662 | { |
| 663 | "name": profile_key, |
| 664 | "name_cn": format_direction_label(profile_key), |
| 665 | "confidence": 1.0, |
| 666 | "source_text": cleaned, |
| 667 | "is_known": True, |
| 668 | } |
| 669 | ) |
| 670 | continue |
| 671 | |
| 672 | resolved = resolve_canonical_direction(cleaned, include_paper_terms=True) |
| 673 | if resolved: |
| 674 | direction_name = str(resolved.get("canonical_name") or "").strip() |
| 675 | if direction_name and direction_name not in seen_canonical: |
| 676 | entry = resolved.get("entry") or {} |
| 677 | seen_canonical.add(direction_name) |
| 678 | canonical_directions.append( |
| 679 | { |
| 680 | "name": direction_name, |
| 681 | "name_cn": str(entry.get("name_cn") or entry.get("name") or direction_name), |
| 682 | "confidence": 1.0, |
| 683 | "source_text": cleaned, |
| 684 | "is_known": True, |
| 685 | } |
| 686 | ) |
| 687 | continue |
| 688 | |
| 689 | normalized = llm_parser.normalize_research_directions( |
| 690 | cleaned, |
| 691 | auto_persist_known_aliases=True, |
| 692 | user_id=user_id, |
| 693 | ) |
| 694 | for direction in normalized.get("canonical_directions", []): |
| 695 | direction_name = str(direction.get("name") or "").strip() |
| 696 | if not direction_name or direction_name in seen_canonical: |
| 697 | continue |
no test coverage detected