File size: 5,555 Bytes
0e6887b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """分析编排器 ``AnalysisOrchestrator``（intent-understanding-layer 任务 11）。
协调「理解 → 确认 → 分析」两阶段流程（design.md「4. AnalysisOrchestrator」）：
- Phase A：:meth:`build_task_sheet` 调用理解层产出《分析任务单》（不触达 ``compute``）。
- Phase B：:meth:`run_confirmed` 接收**用户已确认**的任务单，经 :meth:`Router.route_from_sheet`
路由后，把任务单挂到 ``RawInput.extra["task_sheet"]`` 交由固定 ``Pipeline`` 执行。
合规保证（Property 5 / 需求 1.3、5.4、6.2）：``compute`` 只能经 :meth:`run_confirmed`
触达，而后者要求传入一份非澄清态任务单；澄清态 / 路由失败时**短路返回**澄清说明，
绝不进入流水线。审计在确认执行时记录（需求 8.2 / 9.4）。
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Optional
from .pipeline import Pipeline
from .router import Router
from .skill_base import RawInput, ReportSections
from .task_sheet import AnalysisTaskSheet
from .understanding import UnderstandingLayer
if TYPE_CHECKING: # pragma: no cover - 仅类型检查
from .registry import SkillRegistry
from .services import Services
logger = logging.getLogger(__name__)
class AnalysisOrchestrator:
"""两阶段分析编排：理解（Phase A）→ 用户确认 → 分析（Phase B）。"""
def __init__(self, registry: "SkillRegistry", svc: "Services") -> None:
self.registry = registry
self.svc = svc
self.router = Router(registry, svc)
# ------------------------------------------------------------------
# Phase A：理解
# ------------------------------------------------------------------
def build_task_sheet(self, raw: RawInput) -> AnalysisTaskSheet:
"""运行理解层产出《分析任务单》（不进行任何分析数值计算，Property 4）。"""
return UnderstandingLayer(self.svc).understand(raw)
# ------------------------------------------------------------------
# Phase B：分析（仅在用户确认后调用）
# ------------------------------------------------------------------
def run_confirmed(
self,
sheet: AnalysisTaskSheet,
raw: RawInput,
*,
pipeline: Optional[Pipeline] = None,
user: str = "anonymous",
) -> ReportSections:
"""执行已确认任务单对应的分析。
前置：``sheet`` 必须为**非澄清态**（澄清态直接返回澄清说明，Property 5）。
路由失败（需用户选择）同样短路返回，不进入流水线。
"""
# 澄清态：不得进入流水线（需求 6.2 / Property 5）。
if sheet is None or sheet.needs_clarification():
return self._clarification_sections(sheet)
decision = self.router.route_from_sheet(sheet)
if decision.needs_user_choice or decision.skill is None:
return ReportSections(
sections={
"clarification": {
"reason": "需要用户选择分析功能",
"detail": decision.reason,
}
}
)
raw2 = self._attach_confirmed_payload(raw, sheet)
self._audit_confirm(user, decision.skill.meta.id, sheet)
pipeline = pipeline or Pipeline()
return pipeline.run(decision.skill, raw2, self.svc)
# ------------------------------------------------------------------
@staticmethod
def _attach_confirmed_payload(raw: RawInput, sheet: AnalysisTaskSheet) -> RawInput:
"""把已确认任务单写入 ``RawInput.extra["task_sheet"]``（不改契约）。"""
extra = dict(getattr(raw, "extra", None) or {})
extra["task_sheet"] = sheet.to_dict()
return RawInput(
smiles=getattr(raw, "smiles", "") or "",
excipient=getattr(raw, "excipient", "") or "",
goal=getattr(raw, "goal", "") or "",
files=list(getattr(raw, "files", None) or []),
extra=extra,
)
@staticmethod
def _clarification_sections(sheet: Optional[AnalysisTaskSheet]) -> ReportSections:
clar = getattr(sheet, "clarification", None) if sheet else None
return ReportSections(
sections={
"clarification": {
"reason_code": getattr(clar, "reason_code", "") if clar else "",
"message_key": getattr(clar, "message_key", "") if clar else "",
"options": list(getattr(clar, "options", []) or []) if clar else [],
}
}
)
def _audit_confirm(self, user: str, skill_id: str, sheet: AnalysisTaskSheet) -> None:
audit = getattr(self.svc, "audit", None)
recorder = getattr(audit, "record_analysis", None) or getattr(audit, "record_event", None)
if callable(recorder):
try:
recorder(user or "anonymous", skill_id, step="confirmed_task_sheet")
except TypeError:
try:
recorder("confirm", skill_id, sheet.intent.value)
except Exception: # noqa: BLE001
logger.debug("编排器确认审计失败（已忽略）。", exc_info=True)
except Exception: # noqa: BLE001 - 审计失败不影响主流程
logger.debug("编排器确认审计失败（已忽略）。", exc_info=True)
__all__ = ["AnalysisOrchestrator"]
|