Benjamin Admin
807a7002b2
The advisor was structurally correct but unusable: every question showed a snake_case capability id plus a
single generic fallback reason ("Keine Anhaltspunkte im Unternehmensprofil — klären"). The expert text
already EXISTED in the transition patterns (why_asked / reviewable_claim) — the pipeline just dropped it.
- transition_reasoning: TargetRequirement gains `rationale`; assess_transition uses it as the request
reason when present, else the generic fallback (additive, backward-compatible for all consumers).
- onboarding_service._target carries the pattern's why_asked (delta) and reviewable_claim (likely_covered)
into the requirement rationale -> the question's `why`.
- knowledge/onboarding/capability_labels.yaml: curated DE labels (id -> human), reusable across targets;
labels_for() + response.capability_labels expose them; the frontend renders label || prettified id.
Now ISO27001->TISAX reads "Auftragsverarbeitung (Art. 28 DSGVO) — If a TISAX data label is in scope, you
must show Art. 28 GDPR processing-on-behalf controls; ISO 27001 does not establish these." instead of
"data_protection_processing_on_behalf — klären". why_asked text is still EN (existing knowledge; translation
is curation). 34 onboarding+transition tests pass, mypy --strict clean (13 modules), check-loc 0.
90 lines
4.1 KiB
Python
90 lines
4.1 KiB
Python
"""Onboarding Advisor service — the app-caller that loads knowledge and runs the pure orchestration.
|
|
|
|
This is the SERVICE layer that makes the Smart Onboarding Advisor runtime-usable: it loads the curated
|
|
knowledge (certification hypotheses, signal vocabulary + map, the target's required capabilities) once
|
|
and calls the already-built, pure orchestration (normalize_signals -> silent_intake -> advisor_start).
|
|
It adds NO new reasoning logic — it only exposes what exists. No DB, no persistence (by scope).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any, Dict, Iterable, List, Sequence, Tuple
|
|
|
|
import yaml
|
|
|
|
from compliance.onboarding import (
|
|
AdvisorResult,
|
|
CapabilityHypothesis,
|
|
OnboardingInput,
|
|
ProducedSignal,
|
|
SignalMapping,
|
|
SignalVocabularyEntry,
|
|
advisor_start,
|
|
normalize_signals,
|
|
resolve_for_certifications,
|
|
silent_intake,
|
|
)
|
|
from compliance.transition_reasoning import TargetRequirement
|
|
|
|
_K = os.path.join(os.path.dirname(__file__), "..", "..", "knowledge")
|
|
|
|
|
|
def _load(*parts: str) -> Any:
|
|
return yaml.safe_load(open(os.path.join(_K, *parts), encoding="utf-8"))
|
|
|
|
|
|
_HYP_LIB = [CapabilityHypothesis(**h) for h in _load("certification_hypotheses", "hypotheses.yaml")["hypotheses"]]
|
|
_VOCAB = [SignalVocabularyEntry(**v) for v in _load("onboarding", "signal_vocabulary.yaml")["signals"]]
|
|
_SIGNAL_MAP = [SignalMapping(**m) for m in _load("onboarding", "intake_signal_map.yaml")["mappings"]]
|
|
_LABELS: Dict[str, str] = _load("onboarding", "capability_labels.yaml")["labels"]
|
|
|
|
|
|
def labels_for(capability_ids: Iterable[str]) -> Dict[str, str]:
|
|
"""Human labels (DE) for the given capability ids — presentation only. Ids without a curated label
|
|
are omitted (the frontend falls back to a prettified id). Deduped, deterministic."""
|
|
return {c: _LABELS[c] for c in dict.fromkeys(capability_ids) if c in _LABELS}
|
|
|
|
# target id -> transition pattern that defines its required capabilities (curated registry)
|
|
_TARGET_PATTERNS = {
|
|
"CRA": "transition_pattern_iso27001_to_cra_maschinenvo_v1.yaml",
|
|
"TISAX": "transition_pattern_isms_to_tisax_v1.yaml",
|
|
"MDR": "transition_pattern_iso13485_to_medical_v1.yaml",
|
|
"Environmental": "transition_pattern_iso14001_to_environmental_v1.yaml",
|
|
}
|
|
|
|
|
|
def supported_targets() -> List[str]:
|
|
return sorted(_TARGET_PATTERNS)
|
|
|
|
|
|
def _target(target_id: str) -> Tuple[List[TargetRequirement], Dict[str, List[str]]]:
|
|
pat = _load("transition_patterns", _TARGET_PATTERNS[target_id])
|
|
reqs = [TargetRequirement(capability_id=a["capability"], rationale=a.get("reviewable_claim", "")) for a in pat["likely_covered"]]
|
|
reqs += [TargetRequirement(capability_id=d["capability"], question_intent=d.get("needed_information", "verify_existence"),
|
|
rationale=d.get("why_asked", ""), expected_evidence=d.get("expected_evidence", []))
|
|
for d in pat["delta_requirements"]]
|
|
covers = {d["capability"]: d.get("covers_targets", []) for d in pat["delta_requirements"]}
|
|
return reqs, covers
|
|
|
|
|
|
def run_advisor(
|
|
company: str, certifications: Sequence[str], target: str,
|
|
signals: Sequence[ProducedSignal], known_evidence: Sequence[str],
|
|
products: Sequence[str], markets: Sequence[str], industry: str = "",
|
|
) -> Tuple[AdvisorResult, str]:
|
|
"""Producers (ProducedSignal) -> Normalizer -> Silent Pass -> Advisor. Returns an AdvisorResult.
|
|
|
|
`target` must be a supported target id. Raises KeyError otherwise (the handler maps it to 400/404).
|
|
"""
|
|
reqs, covers = _target(target)
|
|
si = silent_intake(normalize_signals(signals, _VOCAB), _SIGNAL_MAP)
|
|
inp = OnboardingInput(company=company, industry=industry or None, products=list(products),
|
|
markets=list(markets), certifications=list(certifications),
|
|
known_evidence=list(known_evidence), target=[target])
|
|
result = advisor_start(
|
|
inp, resolve_for_certifications(certifications, _HYP_LIB), reqs, target_id=target,
|
|
covers_targets=covers, corpus_status={target: "validated"},
|
|
detected_capabilities=si.capability_ids(), indicative_capabilities=si.indicative_capability_ids())
|
|
return result, si.summary
|