"""路由 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"]