"""Advisory LLM reconciliation for collected nutrition onboarding answers.""" from __future__ import annotations from dataclasses import asdict, dataclass import json from typing import cast from checkin_cli.nutrition_onboarding_contract import QUESTION_FIELDS class ReconciliationError(ValueError): """The untrusted reconciliation response violated its contract.""" class ReconciliationUnavailable(RuntimeError): """The configured advisory model could not produce a safe result.""" @dataclass(frozen=True) class NutritionOnboardingClarification: """One advisory request to re-answer a canonical onboarding field.""" field: str kind: str question_ko: str @dataclass(frozen=True) class NutritionOnboardingReconciliation: """Structured advisory output that cannot authorize onboarding.""" summary_ko: str facts_ko: tuple[str, ...] ambiguities_ko: tuple[str, ...] contradictions_ko: tuple[str, ...] safety_observations_ko: tuple[str, ...] clarifications: tuple[NutritionOnboardingClarification, ...] = () @property def requires_correction(self) -> bool: return bool(self.clarifications) def reconciliation_payload( reconciliation: NutritionOnboardingReconciliation, ) -> dict[str, object]: return cast(dict[str, object], asdict(reconciliation)) def reconciliation_from_payload( payload: object, ) -> NutritionOnboardingReconciliation: return parse_reconciliation_json( json.dumps(payload, ensure_ascii=False, separators=(",", ":")) ) def render_authoritative_customer_summary( answers: dict[str, object], ) -> str: def value(field: str, *, default: str = "미입력") -> str: raw = answers.get(field) if raw is None or raw == "" or raw == []: return default if isinstance(raw, list): return ", ".join(str(item) for item in raw) or default return str(raw) health_fields = ( "medical_conditions", "medications", "pregnancy_lactation", "eating_disorder_risk", ) health_review = ( "검토할 입력이 있습니다." if any(value(field, default="없음") not in {"없음", "해당 없음", "아니요"} for field in health_fields) else "별도 검토 입력이 없습니다." ) restrictions = " · ".join( value(field, default="없음") for field in ( "allergies", "intolerances", "dietary_preferences", "religious_restrictions", ) ) return "\n".join( ( f"- 목표: {value('goal')}", ( "- 활동: " f"{value('activity_category')} · {value('training_details')}" ), f"- 식사: 하루 {value('meal_count')}끼", ( "- 조리·예산: " f"{value('cooking_access')} · {value('budget_band')}" ), f"- 식사 시간: {value('schedule_constraints')}", f"- 알레르기·불내증·식품 제한: {restrictions}", f"- 건강 관련 입력: {health_review}", ) ) def render_reconciliation_text( reconciliation: NutritionOnboardingReconciliation, ) -> str: if reconciliation.clarifications: lines = [ "입력 내용을 정확히 반영하려고 추가 확인이 필요합니다.", "", ] lines.extend( f"- {item.question_ko}" for item in reconciliation.clarifications ) lines.extend( ( "", "질문에 답하면 입력 내용을 다시 정리해 드립니다.", "이 단계에서는 확인·승인·활성화가 진행되지 않습니다.", ) ) return "\n".join(lines) lines = [ "입력 내용을 정리했어요", "", reconciliation.summary_ko, ] sections = ( ("확인된 내용", reconciliation.facts_ko), ("확인이 필요한 모호함", reconciliation.ambiguities_ko), ("서로 맞지 않는 내용", reconciliation.contradictions_ko), ("안전 관련 참고사항", reconciliation.safety_observations_ko), ) for title, items in sections: if items: lines.extend(("", title, *(f"- {item}" for item in items))) lines.extend( ( "", "이 요약은 참고용이며 안전 판정·승인·활성화를 대신하지 않습니다.", "내용이 맞으면 아래 확인 버튼을 눌러 주세요.", ) ) return "\n".join(lines) _FIELDS = { "summary_ko", "facts_ko", "ambiguities_ko", "contradictions_ko", "safety_observations_ko", "clarifications", } _QUESTION_FIELDS = frozenset(QUESTION_FIELDS) _CLARIFICATION_KINDS = frozenset({"ambiguity", "contradiction"}) def _text(value: object, *, field: str, limit: int) -> str: if not isinstance(value, str) or not value.strip(): raise ReconciliationError(f"{field} must be non-empty text") normalized = value.strip() if len(normalized) > limit: raise ReconciliationError(f"{field} exceeds its size limit") return normalized def _text_list(value: object, *, field: str) -> tuple[str, ...]: if not isinstance(value, list) or len(value) > 30: raise ReconciliationError(f"{field} must be a bounded list") return tuple( _text(item, field=field, limit=300) for item in value ) def _summary(value: object) -> str: if isinstance(value, str): return _text(value, field="summary_ko", limit=1200) lines = _text_list(value, field="summary_ko") if not lines: raise ReconciliationError("summary_ko must not be empty") return _text( "\n".join(lines), field="summary_ko", limit=1200, ) def _clarifications( value: object, ) -> tuple[NutritionOnboardingClarification, ...]: if not isinstance(value, list) or len(value) > len(_QUESTION_FIELDS): raise ReconciliationError("clarifications must be a bounded list") items: list[NutritionOnboardingClarification] = [] seen_fields: set[str] = set() for item in value: if not isinstance(item, dict) or set(item) != { "field", "kind", "question_ko", }: raise ReconciliationError( "clarification schema is not exact" ) mapping = cast(dict[str, object], item) field = mapping["field"] kind = mapping["kind"] if not isinstance(field, str) or field not in _QUESTION_FIELDS: raise ReconciliationError( "clarification field is not canonical" ) if field in seen_fields: raise ReconciliationError( "clarification fields must be unique" ) if not isinstance(kind, str) or kind not in _CLARIFICATION_KINDS: raise ReconciliationError( "clarification kind is invalid" ) items.append( NutritionOnboardingClarification( field=field, kind=kind, question_ko=_text( mapping["question_ko"], field="clarification question_ko", limit=180, ), ) ) seen_fields.add(field) return tuple(items) def parse_reconciliation_json( raw: str, ) -> NutritionOnboardingReconciliation: try: payload = json.loads(raw) except (TypeError, json.JSONDecodeError) as exc: raise ReconciliationError("response must be plain JSON") from exc if not isinstance(payload, dict) or set(payload) != _FIELDS: raise ReconciliationError("response schema is not exact") reconciliation = NutritionOnboardingReconciliation( summary_ko=_summary(payload["summary_ko"]), facts_ko=_text_list(payload["facts_ko"], field="facts_ko"), ambiguities_ko=_text_list( payload["ambiguities_ko"], field="ambiguities_ko", ), contradictions_ko=_text_list( payload["contradictions_ko"], field="contradictions_ko", ), safety_observations_ko=_text_list( payload["safety_observations_ko"], field="safety_observations_ko", ), clarifications=_clarifications(payload["clarifications"]), ) if ( reconciliation.ambiguities_ko or reconciliation.contradictions_ko ) and not reconciliation.clarifications: raise ReconciliationError( "ambiguous or contradictory output needs clarification" ) return reconciliation from gateway.platforms.nutrition_onboarding_reconciler import ( NutritionOnboardingReconciler, )