core/kernel/router.py

"""路由 Router：依据输入类型与用户意图选择合适的 Skill（需求 1.5）。

合规取向（对应 design.md「3. 路由 Router」）：**用规则 + LLM 判断该调哪个 Skill，
但 Skill 选定后即走确定性流水线**，LLM 不参与「要不要算 / 怎么算」（需求 1.5、2）。

三级路由策略：

1. **规则匹配（确定性）**：根据 ``RawInput`` 实际携带的输入类型集合，匹配各
   ``SkillMeta.input_kinds``。复用现有 ``detect_analysis_type`` 的判定取向
   （SMILES+辅料 → 相容性，文件+目标 → 稳定性）。为消解「文件+目标」同时命中
   稳定性（``{FILE, TEXT}``）与通用问答（``{TEXT}``）的常见重叠，规则匹配只保留
   **最具体（maximal）** 的候选：若候选 A 的 ``input_kinds`` 是候选 B 的真子集且
   B 也被命中，则 A 被 B「支配」而剔除。如此「文件+目标」唯一命中稳定性。
2. **LLM 意图兜底**：当规则匹配仍得到多个候选（真正的歧义，互不支配）时，调用
   ``svc.llm`` 做意图分类，从候选中择一。LLM 不可用或失败时优雅降级到第三级。
3. **can_handle 打分兜底**：仍不确定时取候选中 ``can_handle`` 最高分者；最高分为 0
   则返回「需用户手动选择」。

``Router.route`` 返回结构化的 :class:`RouteDecision`，记录所选 Skill、决策方式与
候选列表，便于审计与前端提示。
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Optional

from .skill_base import InputKind, PharmaSkill, RawInput

if TYPE_CHECKING:  # pragma: no cover - 仅类型检查
    from .registry import SkillRegistry
    from .services import Services
    from .task_sheet import AnalysisTaskSheet

logger = logging.getLogger(__name__)

#: 意图 → Skill id 映射（与 understanding._INTENT_TO_SKILL 保持一致，需求 7.1）。
INTENT_TO_SKILL = {
    "descriptive_summary": "descriptive_summary",
    "shelf_life_extrapolation": "stability",
    "compatibility": "compatibility",
}


@dataclass
class RouteDecision:
    """路由决策结果。

    - ``skill``：选定的 Skill；``None`` 表示无法自动判定，需用户手动选择。
    - ``method``：决策方式，``"rule"`` | ``"llm"`` | ``"score"`` | ``"none"``。
    - ``candidates``：本次参与决策的候选 Skill id 列表（供审计 / 前端展示）。
    - ``needs_user_choice``：是否需要用户手动选择（``skill is None`` 时为真）。
    - ``reason``：人类可读的决策理由。
    """

    skill: Optional[PharmaSkill]
    method: str
    candidates: list[str] = field(default_factory=list)
    needs_user_choice: bool = False
    reason: str = ""


class Router:
    """技能路由器：规则匹配 → LLM 意图兜底 → can_handle 打分兜底。"""

    def __init__(self, registry: "SkillRegistry", svc: "Services" | None = None) -> None:
        self.registry = registry
        self.svc = svc

    # ------------------------------------------------------------------
    # 公共入口
    # ------------------------------------------------------------------
    def route(self, raw: RawInput) -> RouteDecision:
        """对给定 ``RawInput`` 选择最合适的 Skill。"""
        present = self.detect_input_kinds(raw)

        # 一级：规则匹配（仅保留最具体候选）。
        rule_candidates = self._rule_match(present)
        if len(rule_candidates) == 1:
            chosen = rule_candidates[0]
            return RouteDecision(
                skill=chosen,
                method="rule",
                candidates=[chosen.meta.id],
                reason=f"输入类型 {self._kinds_str(present)} 唯一匹配规则。",
            )

        # 二级：规则不唯一时，用 LLM 从候选中择一。
        if len(rule_candidates) > 1:
            chosen = self._llm_pick(raw, rule_candidates)
            if chosen is not None:
                return RouteDecision(
                    skill=chosen,
                    method="llm",
                    candidates=[s.meta.id for s in rule_candidates],
                    reason="规则匹配存在多个候选，由 LLM 意图分类择一。",
                )
            pool = rule_candidates  # LLM 不可用 / 失败 → 在候选中打分
        else:
            pool = self.registry.all()  # 规则无候选 → 全量打分兜底

        # 三级：can_handle 打分兜底。
        best, score = self._best_by_score(raw, pool)
        if best is not None and score > 0:
            return RouteDecision(
                skill=best,
                method="score",
                candidates=[s.meta.id for s in pool],
                reason=f"can_handle 打分择优（最高分 {score:.2f}）。",
            )

        return RouteDecision(
            skill=None,
            method="none",
            candidates=[s.meta.id for s in pool],
            needs_user_choice=True,
            reason="无法自动判定合适的 Skill，请用户手动选择。",
        )

    # ------------------------------------------------------------------
    # 意图路由（intent-understanding-layer 任务 8）
    # ------------------------------------------------------------------
    def route_from_sheet(self, sheet: "AnalysisTaskSheet") -> RouteDecision:
        """依据已确认的《分析任务单》路由到 Skill（需求 3.5 / 7.1）。

        优先级：
        1. 用户编辑过的 ``proposed_skill_id`` 若能解析为已注册 Skill → 采用。
        2. 否则按 ``sheet.intent`` 经 :data:`INTENT_TO_SKILL` 映射。
        3. 意图为 UNKNOWN / 无映射 / 解析不到 → ``needs_user_choice=True``。

        澄清态任务单（``needs_clarification()``）不应进入路由——调用方应先处理澄清；
        此处仍稳健地返回 ``needs_user_choice``。
        """
        # 澄清态：交回用户选择。
        if getattr(sheet, "needs_clarification", lambda: False)():
            return RouteDecision(
                skill=None, method="none",
                needs_user_choice=True,
                reason="任务单处于澄清状态，需用户先确认数据类型 / 意图。",
            )

        # 1) 用户编辑的 proposed_skill_id 优先。
        proposed = getattr(sheet, "proposed_skill_id", "") or ""
        if proposed:
            skill = self.registry.get(proposed)
            if skill is not None:
                return RouteDecision(
                    skill=skill, method="sheet",
                    candidates=[proposed],
                    reason=f"采用任务单建议 / 用户指定的 Skill：{proposed}。",
                )

        # 2) 意图映射。
        intent_value = getattr(getattr(sheet, "intent", None), "value", None)
        mapped_id = INTENT_TO_SKILL.get(intent_value)
        if mapped_id:
            skill = self.registry.get(mapped_id)
            if skill is not None:
                return RouteDecision(
                    skill=skill, method="intent",
                    candidates=[mapped_id],
                    reason=f"按意图 {intent_value} 路由到 {mapped_id}。",
                )

        # 3) 无法判定。
        return RouteDecision(
            skill=None, method="none",
            needs_user_choice=True,
            reason="无法依据任务单意图路由，请用户手动选择。",
        )

    # ------------------------------------------------------------------
    # 一级：规则匹配
    # ------------------------------------------------------------------
    @staticmethod
    def detect_input_kinds(raw: RawInput) -> set[InputKind]:
        """从 ``RawInput`` 推断其实际携带的输入类型集合。"""
        kinds: set[InputKind] = set()
        if raw.files:
            kinds.add(InputKind.FILE)
        if raw.smiles and raw.smiles.strip():
            kinds.add(InputKind.SMILES)
        if raw.excipient and raw.excipient.strip():
            kinds.add(InputKind.EXCIPIENT)
        if raw.goal and raw.goal.strip():
            kinds.add(InputKind.TEXT)
        return kinds

    def _rule_match(self, present: set[InputKind]) -> list[PharmaSkill]:
        """返回声明的 ``input_kinds`` 被 ``present`` 完全覆盖且最具体的候选 Skill。"""
        matched = [
            s
            for s in self.registry.all()
            if s.meta.input_kinds and s.meta.input_kinds <= present
        ]
        # 仅保留「最具体」候选：剔除被其它候选真超集支配者，消解输入类型重叠歧义。
        maximal: list[PharmaSkill] = []
        for skill in matched:
            dominated = any(
                other is not skill and skill.meta.input_kinds < other.meta.input_kinds
                for other in matched
            )
            if not dominated:
                maximal.append(skill)
        return maximal

    # ------------------------------------------------------------------
    # 二级：LLM 意图兜底
    # ------------------------------------------------------------------
    def _llm_pick(
        self, raw: RawInput, candidates: list[PharmaSkill]
    ) -> Optional[PharmaSkill]:
        """调用 ``svc.llm`` 在候选中做意图分类；不可用 / 失败时返回 ``None``。"""
        llm = getattr(self.svc, "llm", None)
        if llm is None:
            return None

        goal = (raw.goal or "").strip()
        try:
            # 优先使用 LLM 服务的专用意图分类钩子（若提供）。
            classify = getattr(llm, "classify_intent", None)
            if callable(classify):
                chosen_id = classify(goal, [self._meta_brief(s) for s in candidates])
                return self._match_id(chosen_id, candidates)

            # 退一步：使用通用 complete 接口并解析返回的 id。
            complete = getattr(llm, "complete", None)
            if callable(complete):
                chosen_id = self._classify_via_complete(complete, goal, candidates)
                return self._match_id(chosen_id, candidates)
        except Exception:  # noqa: BLE001 - LLM 兜底失败不得中断路由
            logger.warning("LLM 意图兜底失败，降级到 can_handle 打分。", exc_info=True)
            return None

        return None

    @staticmethod
    def _classify_via_complete(complete, goal: str, candidates: list[PharmaSkill]):
        """用通用 ``complete(system, user)`` 接口做意图分类，返回所选 id 文本。"""
        options = "\n".join(
            f"- {s.meta.id}: {s.meta.display_name} —— {s.meta.description}"
            for s in candidates
        )
        system = (
            "你是药学分析平台的意图路由器。根据用户目标，从候选技能中选出最合适的一个，"
            "只输出该技能的 id，不要任何解释或多余字符。"
        )
        user = f"用户目标：{goal or '（未提供文字目标）'}\n\n候选技能：\n{options}"
        result = complete(system, user, temperature=0.0)
        # complete 可能返回字符串或带 .text 字段的结果对象。
        text = getattr(result, "text", result)
        return str(text).strip() if text is not None else None

    @staticmethod
    def _match_id(
        chosen_id, candidates: list[PharmaSkill]
    ) -> Optional[PharmaSkill]:
        """把 LLM 返回的 id 文本匹配回候选 Skill；匹配失败返回 ``None``。"""
        if not chosen_id:
            return None
        needle = str(chosen_id).strip()
        # 精确匹配优先。
        for skill in candidates:
            if skill.meta.id == needle:
                return skill
        # 容错：LLM 输出可能包含额外文字，做包含匹配。
        for skill in candidates:
            if skill.meta.id and skill.meta.id in needle:
                return skill
        return None

    @staticmethod
    def _meta_brief(skill: PharmaSkill) -> dict:
        """构造供 LLM 分类用的候选元数据摘要。"""
        meta = skill.meta
        return {
            "id": meta.id,
            "display_name": meta.display_name,
            "description": meta.description,
            "input_kinds": sorted(k.value for k in meta.input_kinds),
        }

    # ------------------------------------------------------------------
    # 三级：can_handle 打分兜底
    # ------------------------------------------------------------------
    @staticmethod
    def _best_by_score(
        raw: RawInput, pool: list[PharmaSkill]
    ) -> tuple[Optional[PharmaSkill], float]:
        """返回 ``pool`` 中 ``can_handle`` 最高分的 Skill 与其分值。"""
        best: Optional[PharmaSkill] = None
        best_score = 0.0
        for skill in pool:
            try:
                score = float(skill.can_handle(raw))
            except Exception:  # noqa: BLE001 - 单个打分异常不得影响整体
                logger.debug("Skill %s 的 can_handle 打分异常。", skill.meta.id, exc_info=True)
                score = 0.0
            if score > best_score:
                best, best_score = skill, score
        return best, best_score

    # ------------------------------------------------------------------
    @staticmethod
    def _kinds_str(kinds: set[InputKind]) -> str:
        return "{" + ", ".join(sorted(k.value for k in kinds)) + "}"


__all__ = ["Router", "RouteDecision"]