813 lines
30 KiB
Python
813 lines
30 KiB
Python
"""Application-owned conversational workflow around deterministic RollCalc APIs."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from copy import deepcopy
|
||
from dataclasses import dataclass, field
|
||
import math
|
||
import secrets
|
||
from threading import Lock
|
||
from typing import Any
|
||
|
||
from core_presets import CorePreset, core_family, matching_core_presets
|
||
from ollama_nlu import (
|
||
ClarificationInterpretation,
|
||
NLUError,
|
||
NLUInterpretation,
|
||
NLUResponse,
|
||
OllamaNLUClient,
|
||
parse_clarification_reply,
|
||
)
|
||
from roll_calculation import calculate_roll, modify_calculation
|
||
|
||
|
||
class ConversationNotFoundError(KeyError):
|
||
"""Raised when an in-memory conversation ID is unknown."""
|
||
|
||
|
||
class ConversationReportNotFoundError(KeyError):
|
||
"""Raised when an in-memory report ID is unknown."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ArticleCandidate:
|
||
article_number: str
|
||
name: str
|
||
|
||
def to_dict(self) -> dict[str, str]:
|
||
return {
|
||
"article_number": self.article_number,
|
||
"name": self.name,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PendingClarification:
|
||
field: str
|
||
reason: str
|
||
core_family: str | None = None
|
||
candidates: tuple[CorePreset, ...] = ()
|
||
article_candidates: tuple[ArticleCandidate, ...] = ()
|
||
requested_changes: dict[str, Any] = field(default_factory=dict)
|
||
|
||
@property
|
||
def allowed_values(self) -> tuple[float, ...]:
|
||
return tuple(preset.diameter_mm for preset in self.candidates)
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
result = {
|
||
"field": self.field,
|
||
"reason": self.reason,
|
||
"core_family": self.core_family,
|
||
"allowed_values": list(self.allowed_values),
|
||
"candidates": [
|
||
{
|
||
"key": preset.key,
|
||
"label": preset.label,
|
||
"diameter_mm": preset.diameter_mm,
|
||
"material": preset.material,
|
||
}
|
||
for preset in self.candidates
|
||
],
|
||
}
|
||
if self.requested_changes:
|
||
result["requested_changes"] = deepcopy(self.requested_changes)
|
||
if self.reason == "ambiguous_article":
|
||
result.pop("core_family")
|
||
result.pop("allowed_values")
|
||
result["candidates"] = [
|
||
candidate.to_dict() for candidate in self.article_candidates
|
||
]
|
||
return result
|
||
|
||
|
||
@dataclass
|
||
class ConversationRecord:
|
||
calculation_state: dict[str, Any] | None = None
|
||
last_result: dict[str, Any] | None = None
|
||
pending_clarification: PendingClarification | None = None
|
||
|
||
@property
|
||
def pending_fields(self) -> tuple[str, ...]:
|
||
if self.pending_clarification is None:
|
||
return ()
|
||
return (self.pending_clarification.field,)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StoredReport:
|
||
request: dict[str, Any]
|
||
conversation_id: str
|
||
|
||
|
||
class InMemoryConversationStore:
|
||
"""Process-local MVP storage; all values are copied across its boundary."""
|
||
|
||
def __init__(self) -> None:
|
||
self._conversations: dict[str, ConversationRecord] = {}
|
||
self._reports: dict[str, StoredReport] = {}
|
||
self._lock = Lock()
|
||
|
||
def create(self) -> str:
|
||
conversation_id = secrets.token_urlsafe(18)
|
||
with self._lock:
|
||
self._conversations[conversation_id] = ConversationRecord()
|
||
return conversation_id
|
||
|
||
def get(self, conversation_id: str) -> ConversationRecord:
|
||
with self._lock:
|
||
record = self._conversations.get(conversation_id)
|
||
if record is None:
|
||
raise ConversationNotFoundError(conversation_id)
|
||
return deepcopy(record)
|
||
|
||
def save(self, conversation_id: str, record: ConversationRecord) -> None:
|
||
with self._lock:
|
||
if conversation_id not in self._conversations:
|
||
raise ConversationNotFoundError(conversation_id)
|
||
self._conversations[conversation_id] = deepcopy(record)
|
||
|
||
def add_report(self, conversation_id: str, request: dict[str, Any]) -> str:
|
||
report_id = secrets.token_urlsafe(18)
|
||
with self._lock:
|
||
if conversation_id not in self._conversations:
|
||
raise ConversationNotFoundError(conversation_id)
|
||
self._reports[report_id] = StoredReport(
|
||
request=deepcopy(request),
|
||
conversation_id=conversation_id,
|
||
)
|
||
return report_id
|
||
|
||
def get_report(self, report_id: str) -> StoredReport:
|
||
with self._lock:
|
||
report = self._reports.get(report_id)
|
||
if report is None:
|
||
raise ConversationReportNotFoundError(report_id)
|
||
return deepcopy(report)
|
||
|
||
|
||
def _calculation_request(interpretation: NLUInterpretation) -> dict[str, Any]:
|
||
payload = interpretation.to_dict()
|
||
payload["intent"] = "calculate_roll_diameter"
|
||
return payload
|
||
|
||
|
||
def _has_valid_calculation(record: ConversationRecord) -> bool:
|
||
return (
|
||
record.calculation_state is not None
|
||
and isinstance(record.last_result, dict)
|
||
and record.last_result.get("status") == "success"
|
||
)
|
||
|
||
|
||
def _current_nlu_context(record: ConversationRecord) -> dict[str, Any] | None:
|
||
if not _has_valid_calculation(record):
|
||
return None
|
||
request = record.calculation_state.get("request", {})
|
||
if not isinstance(request, dict):
|
||
return None
|
||
fields = (
|
||
"article_number",
|
||
"article_name_hint",
|
||
"roll_length_m",
|
||
"width_m",
|
||
"core_type",
|
||
"core_diameter_mm",
|
||
"include_roll_weight",
|
||
)
|
||
resolved_article = record.last_result.get("resolved_article", {})
|
||
resolved_name = (
|
||
resolved_article.get("name")
|
||
if isinstance(resolved_article, dict)
|
||
else None
|
||
)
|
||
return {
|
||
"has_successful_calculation": True,
|
||
"resolved_article_name": resolved_name,
|
||
**{field: request.get(field) for field in fields},
|
||
}
|
||
|
||
|
||
def _new_calculation_followup_changes(
|
||
interpretation: NLUInterpretation,
|
||
record: ConversationRecord,
|
||
message: str,
|
||
) -> dict[str, Any] | None:
|
||
if interpretation.intent != "new_calculation" or not _has_valid_calculation(record):
|
||
return None
|
||
if interpretation.article_number is not None:
|
||
return None
|
||
if interpretation.article_name_hint is not None:
|
||
return None
|
||
normalized_message = " ".join(message.casefold().split())
|
||
if any(
|
||
marker in normalized_message
|
||
for marker in (
|
||
"neue berechnung",
|
||
"neue rollenberechnung",
|
||
"new calculation",
|
||
"start over",
|
||
)
|
||
):
|
||
return None
|
||
|
||
changes = {}
|
||
for field in (
|
||
"roll_length_m",
|
||
"width_m",
|
||
"core_type",
|
||
"core_diameter_mm",
|
||
):
|
||
value = getattr(interpretation, field)
|
||
if value is not None:
|
||
changes[field] = value
|
||
if interpretation.include_roll_weight:
|
||
changes["include_roll_weight"] = True
|
||
return changes or None
|
||
|
||
|
||
def _pending_clarification(
|
||
result: dict[str, Any],
|
||
*,
|
||
requested_changes: dict[str, Any] | None = None,
|
||
) -> PendingClarification | None:
|
||
missing = result.get("missing", [])
|
||
ambiguous = result.get("ambiguous", [])
|
||
for field in (
|
||
"width_m",
|
||
"core_diameter_mm",
|
||
"roll_length_m",
|
||
"thickness_mm",
|
||
"area_weight_g_m2",
|
||
):
|
||
if field in missing:
|
||
return PendingClarification(
|
||
field,
|
||
"missing_parameter",
|
||
requested_changes=deepcopy(requested_changes or {}),
|
||
)
|
||
if "core_type" in ambiguous:
|
||
request = result.get("request", {})
|
||
core_type = request.get("core_type") if isinstance(request, dict) else None
|
||
if isinstance(core_type, str):
|
||
candidates = matching_core_presets(core_type)
|
||
family = core_family(core_type)
|
||
if len(candidates) > 1 and family is not None:
|
||
return PendingClarification(
|
||
field="core_diameter_mm",
|
||
reason="ambiguous_core_preset",
|
||
core_family=family,
|
||
candidates=candidates,
|
||
requested_changes=deepcopy(requested_changes or {}),
|
||
)
|
||
return PendingClarification(
|
||
"core_diameter_mm",
|
||
"ambiguous_core_type",
|
||
requested_changes=deepcopy(requested_changes or {}),
|
||
)
|
||
if "article" in ambiguous or any(
|
||
field in ambiguous for field in ("article_name_hint", "article_number")
|
||
):
|
||
resolution = result.get("article_resolution", {})
|
||
raw_candidates = (
|
||
resolution.get("candidates", [])
|
||
if isinstance(resolution, dict)
|
||
else []
|
||
)
|
||
candidates = tuple(
|
||
ArticleCandidate(
|
||
article_number=item["article_number"],
|
||
name=item["name"],
|
||
)
|
||
for item in raw_candidates
|
||
if isinstance(item, dict)
|
||
and isinstance(item.get("article_number"), str)
|
||
and isinstance(item.get("name"), str)
|
||
)
|
||
return PendingClarification(
|
||
"article",
|
||
"ambiguous_article",
|
||
article_candidates=candidates,
|
||
requested_changes=deepcopy(requested_changes or {}),
|
||
)
|
||
return None
|
||
|
||
|
||
def _format_diameter_choices(values: tuple[float, ...]) -> str:
|
||
labels = [f"{value:g} mm" for value in values]
|
||
if len(labels) < 2:
|
||
return labels[0] if labels else ""
|
||
if len(labels) == 2:
|
||
return f"{labels[0]} oder {labels[1]}"
|
||
return f"{', '.join(labels[:-1])} und {labels[-1]}"
|
||
|
||
|
||
def _core_family_names(family: str | None) -> tuple[str, str]:
|
||
if family == "steel":
|
||
return "Stahlkern", "Stahlkerne"
|
||
if family == "pvc":
|
||
return "PVC-Kern", "PVC-Kerne"
|
||
return "Kern", "Kerne"
|
||
|
||
|
||
def ambiguous_core_message(pending: PendingClarification) -> str:
|
||
_, plural = _core_family_names(pending.core_family)
|
||
choices = _format_diameter_choices(pending.allowed_values)
|
||
return (
|
||
f"Es sind mehrere {plural} hinterlegt. "
|
||
f"Welchen Kerndurchmesser soll ich verwenden: {choices}?"
|
||
)
|
||
|
||
|
||
def unavailable_core_diameter_message(
|
||
pending: PendingClarification,
|
||
value: float,
|
||
) -> str:
|
||
singular, _ = _core_family_names(pending.core_family)
|
||
choices = _format_diameter_choices(pending.allowed_values).replace(
|
||
" oder ",
|
||
" und ",
|
||
)
|
||
return (
|
||
f"{value:g} mm ist für einen {singular} nicht hinterlegt. "
|
||
f"Verfügbar sind {choices}."
|
||
)
|
||
|
||
|
||
def ambiguous_article_message(pending: PendingClarification) -> str:
|
||
choices = "\noder\n".join(
|
||
f"{candidate.article_number} – {' '.join(candidate.name.split())}"
|
||
for candidate in pending.article_candidates
|
||
)
|
||
return (
|
||
"Ich habe mehrere passende Artikel gefunden. Welchen meinst Du:\n"
|
||
f"{choices}?"
|
||
)
|
||
|
||
|
||
def incompatible_clarification_message(field: str) -> str:
|
||
messages = {
|
||
"core_diameter_mm": (
|
||
"Ich benötige den Kerndurchmesser in Millimetern, z. B. 150 mm."
|
||
),
|
||
"roll_length_m": "Ich benötige die Rollenlänge in Metern, z. B. 80 m.",
|
||
"width_m": "Ich benötige die Breite in Metern, z. B. 4,90 m.",
|
||
"thickness_mm": "Ich benötige die Materialdicke in Millimetern.",
|
||
"area_weight_g_m2": "Ich benötige das Flächengewicht in g/m².",
|
||
"article_number": "Bitte gib die genaue Artikelnummer an.",
|
||
"article": "Bitte wähle eine der genannten Artikelnummern aus.",
|
||
}
|
||
return messages.get(field, "Die Antwort passt nicht zur angeforderten Angabe.")
|
||
|
||
|
||
def clarification_message(
|
||
result: dict[str, Any],
|
||
pending: PendingClarification | None = None,
|
||
) -> str:
|
||
"""Map deterministic result statuses to deterministic German messages."""
|
||
status = result.get("status")
|
||
if status == "article_not_found":
|
||
return "Der angegebene Artikel wurde nicht gefunden."
|
||
if status == "article_conflict":
|
||
return (
|
||
"Artikelnummer und Produktbezeichnung passen nicht zusammen. "
|
||
"Bitte prüfe die Angaben."
|
||
)
|
||
if status == "invalid_parameter":
|
||
invalid = result.get("invalid", [])
|
||
if invalid and isinstance(invalid[0], dict):
|
||
field = invalid[0].get("field", "Eingabe")
|
||
detail = invalid[0].get("message", "ungültiger Wert")
|
||
return f"Die Eingabe für {field} ist ungültig: {detail}."
|
||
return "Die Eingaben sind ungültig. Bitte prüfe die Angaben."
|
||
|
||
missing = result.get("missing", [])
|
||
ambiguous = result.get("ambiguous", [])
|
||
if "width_m" in missing:
|
||
return (
|
||
"Für diesen Artikel ist keine eindeutige Breite verfügbar. "
|
||
"Welche Breite soll ich verwenden?"
|
||
)
|
||
if "core_diameter_mm" in missing:
|
||
return "Welchen Kerndurchmesser soll ich verwenden?"
|
||
if "roll_length_m" in missing:
|
||
return "Welche Rollenlänge soll ich verwenden?"
|
||
if "thickness_mm" in missing:
|
||
return (
|
||
"Für diese Berechnung ist keine Materialdicke verfügbar. "
|
||
"Bitte gib die Materialdicke an."
|
||
)
|
||
if "area_weight_g_m2" in missing:
|
||
return (
|
||
"Für die Gewichtsberechnung ist kein Flächengewicht verfügbar. "
|
||
"Bitte gib das Flächengewicht an."
|
||
)
|
||
if "core_type" in ambiguous:
|
||
if pending is not None and pending.reason == "ambiguous_core_preset":
|
||
return ambiguous_core_message(pending)
|
||
request = result.get("request", {})
|
||
core_type = request.get("core_type") if isinstance(request, dict) else None
|
||
family = core_family(core_type) if isinstance(core_type, str) else None
|
||
if family == "steel":
|
||
return (
|
||
"Es sind mehrere Stahlkern-Presets vorhanden. "
|
||
"Bitte gib den Kerndurchmesser an."
|
||
)
|
||
if family == "pvc":
|
||
return (
|
||
"Es sind mehrere PVC-Kern-Presets vorhanden. "
|
||
"Bitte gib den Kerndurchmesser an."
|
||
)
|
||
return "Der Kerntyp ist nicht eindeutig. Bitte gib den Kerndurchmesser an."
|
||
if pending is not None and pending.reason == "ambiguous_article":
|
||
return ambiguous_article_message(pending)
|
||
if any(
|
||
field in ambiguous
|
||
for field in ("article", "article_name_hint", "article_number")
|
||
):
|
||
return "Der Artikel ist nicht eindeutig. Bitte gib die genaue Artikelnummer an."
|
||
return "Für die Berechnung fehlen eindeutige Angaben. Bitte ergänze die Eingaben."
|
||
|
||
|
||
def success_message(result: dict[str, Any], pdf_url: str) -> str:
|
||
"""Render only authoritative result values using a fixed response template."""
|
||
article = result.get("resolved_article") or {}
|
||
calculation = result["calculation"]
|
||
number = article.get("number") or "ohne Artikelnummer"
|
||
name = article.get("name") or "ohne Produktbezeichnung"
|
||
lines = [
|
||
"Berechnung abgeschlossen.",
|
||
"",
|
||
f"Artikel: {name} ({number})",
|
||
f"Rollenlänge: {calculation['effective_roll_length_m']:.2f} m",
|
||
]
|
||
effective_width = result.get("effective_inputs", {}).get("width_m")
|
||
if isinstance(effective_width, dict) and effective_width.get("value") is not None:
|
||
lines.append(f"Breite: {effective_width['value']:.2f} m")
|
||
lines.extend((
|
||
f"Minimum: {calculation['minimum_diameter_mm']:.1f} mm",
|
||
f"Mittelwert: {calculation['average_diameter_mm']:.1f} mm",
|
||
f"Maximum: {calculation['maximum_diameter_mm']:.1f} mm",
|
||
))
|
||
if calculation.get("roll_weight_kg") is not None:
|
||
lines.append(f"Rollengewicht: {calculation['roll_weight_kg']:.1f} kg")
|
||
for warning in result.get("warnings", []):
|
||
lines.append(f"Warnung: {warning['title']} – {warning['message']}")
|
||
for note in result.get("notes", []):
|
||
lines.append(f"Hinweis: {note}")
|
||
lines.extend(("", f"PDF: {pdf_url}"))
|
||
return "\n".join(lines)
|
||
|
||
|
||
class ConversationService:
|
||
def __init__(
|
||
self,
|
||
nlu_client: OllamaNLUClient,
|
||
*,
|
||
store: InMemoryConversationStore | None = None,
|
||
build_info: dict[str, str] | None = None,
|
||
) -> None:
|
||
self.nlu_client = nlu_client
|
||
self.store = store or InMemoryConversationStore()
|
||
self.build_info = build_info or {}
|
||
|
||
def create_conversation(self) -> str:
|
||
return self.store.create()
|
||
|
||
def conversation_state(self, conversation_id: str) -> ConversationRecord:
|
||
return self.store.get(conversation_id)
|
||
|
||
def report_request(self, report_id: str) -> dict[str, Any]:
|
||
return self.store.get_report(report_id).request
|
||
|
||
def handle_message(
|
||
self,
|
||
conversation_id: str,
|
||
message: str,
|
||
*,
|
||
include_diagnostics: bool = False,
|
||
) -> dict[str, Any]:
|
||
record = self.store.get(conversation_id)
|
||
clarification: ClarificationInterpretation | None = None
|
||
nlu: NLUResponse | None = None
|
||
applied_as_modification = False
|
||
pending = record.pending_clarification
|
||
had_valid_result = (
|
||
isinstance(record.last_result, dict)
|
||
and record.last_result.get("status") == "success"
|
||
)
|
||
if pending is not None:
|
||
expected_field = pending.field
|
||
selected_article: ArticleCandidate | None = None
|
||
if pending.reason == "ambiguous_article":
|
||
article_number = message.strip()
|
||
selected_article = next(
|
||
(
|
||
candidate
|
||
for candidate in pending.article_candidates
|
||
if candidate.article_number == article_number
|
||
),
|
||
None,
|
||
)
|
||
clarification = ClarificationInterpretation(
|
||
"parsed" if selected_article is not None else "incompatible",
|
||
expected_field,
|
||
selected_article.article_number if selected_article else None,
|
||
)
|
||
else:
|
||
clarification = parse_clarification_reply(message, expected_field)
|
||
if clarification.status != "parsed":
|
||
response = {
|
||
"status": "needs_clarification",
|
||
"message": (
|
||
ambiguous_article_message(pending)
|
||
if pending.reason == "ambiguous_article"
|
||
else incompatible_clarification_message(expected_field)
|
||
),
|
||
"clarification": clarification.to_dict(),
|
||
"pending_clarification": pending.to_dict(),
|
||
"result": record.last_result,
|
||
}
|
||
return self._with_clarification(
|
||
response,
|
||
clarification,
|
||
include_diagnostics,
|
||
)
|
||
|
||
if record.calculation_state is None:
|
||
return {
|
||
"status": "needs_calculation",
|
||
"message": (
|
||
"Es gibt noch keine Berechnung, die ergänzt werden kann. "
|
||
"Bitte starte zuerst eine neue Berechnung."
|
||
),
|
||
}
|
||
if pending.reason == "ambiguous_article":
|
||
changes = dict(pending.requested_changes)
|
||
changes["article_number"] = clarification.value
|
||
else:
|
||
changes = dict(pending.requested_changes)
|
||
changes[expected_field] = clarification.value
|
||
if pending.reason == "ambiguous_core_preset":
|
||
preset = next(
|
||
(
|
||
candidate
|
||
for candidate in pending.candidates
|
||
if math.isclose(
|
||
candidate.diameter_mm,
|
||
clarification.value,
|
||
)
|
||
),
|
||
None,
|
||
)
|
||
if preset is None:
|
||
response = {
|
||
"status": "needs_clarification",
|
||
"message": unavailable_core_diameter_message(
|
||
pending,
|
||
clarification.value,
|
||
),
|
||
"clarification": {
|
||
"status": "invalid_value",
|
||
"field": pending.field,
|
||
"value": clarification.value,
|
||
},
|
||
"pending_clarification": pending.to_dict(),
|
||
"result": record.last_result,
|
||
}
|
||
return self._with_clarification(
|
||
response,
|
||
clarification,
|
||
include_diagnostics,
|
||
)
|
||
changes.update({
|
||
"core_type": preset.label,
|
||
"core_diameter_mm": preset.diameter_mm,
|
||
})
|
||
elif expected_field == "core_diameter_mm":
|
||
current_request = record.calculation_state.get("request", {})
|
||
current_core_type = current_request.get("core_type")
|
||
if current_core_type is not None:
|
||
changes["core_type"] = current_core_type
|
||
result = modify_calculation(
|
||
record.calculation_state,
|
||
changes,
|
||
build_info=self.build_info,
|
||
)
|
||
applied_as_modification = True
|
||
else:
|
||
try:
|
||
nlu = self.nlu_client.interpret(
|
||
message,
|
||
has_state=record.calculation_state is not None,
|
||
expected_fields=(),
|
||
current_state=_current_nlu_context(record),
|
||
)
|
||
except NLUError as error:
|
||
return {
|
||
"status": "nlu_error",
|
||
"error": error.__class__.__name__,
|
||
"message": (
|
||
"Die Spracheingabe konnte nicht sicher verarbeitet werden. "
|
||
"Bitte versuche es erneut oder formuliere die Angaben expliziter."
|
||
),
|
||
}
|
||
|
||
interpretation = nlu.interpretation
|
||
if interpretation.intent == "unsupported":
|
||
response = {
|
||
"status": "unsupported",
|
||
"message": (
|
||
"Diese Anfrage wird derzeit nicht unterstützt. Möglich sind neue "
|
||
"Rollendurchmesser-Berechnungen und Änderungen einer bestehenden "
|
||
"Berechnung."
|
||
),
|
||
}
|
||
return self._with_nlu(response, nlu, include_diagnostics)
|
||
|
||
followup_changes = _new_calculation_followup_changes(
|
||
interpretation,
|
||
record,
|
||
message,
|
||
)
|
||
if followup_changes is not None:
|
||
result = modify_calculation(
|
||
record.calculation_state,
|
||
followup_changes,
|
||
build_info=self.build_info,
|
||
)
|
||
applied_as_modification = True
|
||
elif interpretation.intent == "new_calculation":
|
||
result = calculate_roll(
|
||
_calculation_request(interpretation),
|
||
build_info=self.build_info,
|
||
)
|
||
else:
|
||
if record.calculation_state is None:
|
||
response = {
|
||
"status": "needs_calculation",
|
||
"message": (
|
||
"Es gibt noch keine Berechnung, die geändert werden kann. "
|
||
"Bitte starte zuerst eine neue Berechnung."
|
||
),
|
||
}
|
||
return self._with_nlu(response, nlu, include_diagnostics)
|
||
changes = interpretation.changes or {}
|
||
current_request = record.calculation_state.get("request", {})
|
||
current_article = current_request.get("article_number")
|
||
different_article = (
|
||
"article_number" in changes
|
||
and changes["article_number"] != current_article
|
||
) or "article_name_hint" in changes
|
||
if different_article:
|
||
new_request = {
|
||
"intent": "calculate_roll_diameter",
|
||
**{
|
||
field: changes[field]
|
||
for field in ("article_number", "article_name_hint")
|
||
if field in changes
|
||
},
|
||
}
|
||
result = calculate_roll(
|
||
new_request,
|
||
build_info=self.build_info,
|
||
)
|
||
else:
|
||
result = modify_calculation(
|
||
record.calculation_state,
|
||
changes,
|
||
build_info=self.build_info,
|
||
)
|
||
applied_as_modification = True
|
||
|
||
if clarification is not None and result.get("status") == "invalid_parameter":
|
||
response = {
|
||
"status": "needs_clarification",
|
||
"message": incompatible_clarification_message(clarification.field),
|
||
"clarification": clarification.to_dict(),
|
||
"result": result,
|
||
}
|
||
return self._with_clarification(
|
||
response,
|
||
clarification,
|
||
include_diagnostics,
|
||
)
|
||
|
||
pending_changes = None
|
||
if applied_as_modification:
|
||
if clarification is not None:
|
||
pending_changes = changes
|
||
elif nlu is not None and nlu.interpretation.intent == "modify_calculation":
|
||
pending_changes = nlu.interpretation.changes
|
||
else:
|
||
pending_changes = followup_changes
|
||
new_pending = _pending_clarification(
|
||
result,
|
||
requested_changes=pending_changes,
|
||
)
|
||
preserve_valid_state = (
|
||
had_valid_result
|
||
and (
|
||
(
|
||
new_pending is not None
|
||
and new_pending.reason == "ambiguous_article"
|
||
)
|
||
or (
|
||
applied_as_modification
|
||
and (
|
||
(
|
||
new_pending is not None
|
||
and new_pending.reason
|
||
in {
|
||
"ambiguous_core_preset",
|
||
"ambiguous_core_type",
|
||
"ambiguous_article",
|
||
}
|
||
)
|
||
or result.get("status")
|
||
in {
|
||
"invalid_parameter",
|
||
"article_not_found",
|
||
"article_conflict",
|
||
}
|
||
)
|
||
)
|
||
)
|
||
)
|
||
if not preserve_valid_state:
|
||
state = result.get("state")
|
||
if isinstance(state, dict):
|
||
record.calculation_state = state
|
||
record.last_result = result
|
||
record.pending_clarification = new_pending
|
||
self.store.save(conversation_id, record)
|
||
|
||
if result.get("status") != "success":
|
||
response = {
|
||
"status": result.get("status", "invalid_parameter"),
|
||
"message": clarification_message(result, new_pending),
|
||
"result": result,
|
||
}
|
||
if new_pending is not None:
|
||
pending_context = new_pending.to_dict()
|
||
response["pending_clarification"] = pending_context
|
||
if new_pending.reason == "ambiguous_article":
|
||
response.update({
|
||
"field": pending_context["field"],
|
||
"reason": pending_context["reason"],
|
||
"candidates": pending_context["candidates"],
|
||
})
|
||
if clarification is not None:
|
||
return self._with_clarification(
|
||
response,
|
||
clarification,
|
||
include_diagnostics,
|
||
)
|
||
return self._with_nlu(response, nlu, include_diagnostics)
|
||
|
||
report_id = self.store.add_report(conversation_id, result["request"])
|
||
pdf_url = f"/api/conversations/reports/{report_id}.pdf"
|
||
response = {
|
||
"status": "success",
|
||
"message": success_message(result, pdf_url),
|
||
"result": result,
|
||
"pdf": {"report_id": report_id, "url": pdf_url},
|
||
}
|
||
if clarification is not None:
|
||
return self._with_clarification(
|
||
response,
|
||
clarification,
|
||
include_diagnostics,
|
||
)
|
||
return self._with_nlu(response, nlu, include_diagnostics)
|
||
|
||
@staticmethod
|
||
def _with_nlu(
|
||
response: dict[str, Any],
|
||
nlu: NLUResponse,
|
||
include_diagnostics: bool,
|
||
) -> dict[str, Any]:
|
||
response["interpretation"] = nlu.interpretation.to_dict()
|
||
response["nlu"] = {
|
||
"model": nlu.model,
|
||
"latency_ms": round(nlu.latency_ms, 1),
|
||
}
|
||
if include_diagnostics:
|
||
response["nlu"]["raw_model_json"] = nlu.raw_model_json
|
||
return response
|
||
|
||
@staticmethod
|
||
def _with_clarification(
|
||
response: dict[str, Any],
|
||
clarification: ClarificationInterpretation,
|
||
include_diagnostics: bool,
|
||
) -> dict[str, Any]:
|
||
response["interpretation"] = clarification.to_dict()
|
||
response["nlu"] = {
|
||
"model": "deterministic_clarification_parser",
|
||
"latency_ms": 0.0,
|
||
}
|
||
if include_diagnostics:
|
||
response["nlu"]["raw_model_json"] = None
|
||
return response
|