327 lines
14 KiB
Python
327 lines
14 KiB
Python
"""Jev question definitions for the rectification turn-intent classifier.
|
||
|
||
Offline research only. Does not change production `turn-intent-classifier.ts`.
|
||
Question JSON matches TypeSafe `primitives/advanced` (Choice / Noul with
|
||
structured instructions). Combination legality is enforced in code.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Mapping, Sequence
|
||
|
||
INTENT_VALUES = (
|
||
"answer_current_focus",
|
||
"provide_new_evidence",
|
||
"stop_rectification",
|
||
"ask_about_result",
|
||
"unclear",
|
||
)
|
||
ANSWER_CLASS_VALUES = ("yes", "weak_yes", "no", "unsure")
|
||
COLLECT_ANSWER_CLASSES = ("yes", "weak_yes", "no", "unsure")
|
||
JEV_MODEL = "jev-1.13.0"
|
||
NOUL_TRUE_THRESHOLD = 0.5
|
||
|
||
# Literal criteria. Production hedges ("通常", "不要猜", "不要按关键词") are dropped.
|
||
INTENT_INSTRUCTIONS = {
|
||
"task": "Classify what the user_message is doing in this rectification turn.",
|
||
"inputs": ["current_question", "options", "user_message", "case_status"],
|
||
"rule": "Pick exactly one option. Use only the literal conditions in criteria.",
|
||
}
|
||
|
||
INTENT_CRITERIA: dict[str, str] = {
|
||
"answer_current_focus": (
|
||
"user_message is answering current_question. "
|
||
"If options is non-empty, the message matches the meaning of one option's answer_class. "
|
||
"If options is empty (collect question), the message says the topic happened, did not happen, "
|
||
"or is not remembered, or answers the collect question with a dated experience."
|
||
),
|
||
"provide_new_evidence": (
|
||
"user_message only adds a new experience that includes an approximate year or month, "
|
||
"and does not answer current_question. "
|
||
"If current_question is empty, adding a dated experience also maps here."
|
||
),
|
||
"stop_rectification": (
|
||
"user_message explicitly asks to stop the entire birth-time rectification. "
|
||
"Saying the current topic did not happen is not this option."
|
||
),
|
||
"ask_about_result": (
|
||
"user_message asks for the rectification result, the current time range, "
|
||
"candidate minutes, or whether a result is ready. It is not answering current_question."
|
||
),
|
||
"unclear": (
|
||
"None of the other four conditions hold, or the message mixes them so no single option applies."
|
||
),
|
||
}
|
||
|
||
ANSWER_CLASS_INSTRUCTIONS = {
|
||
"task": "If user_message is answering current_question, pick the matching answer_class.",
|
||
"rule": "Use only the option list in criteria. If the message is not answering current_question, pick none of the content classes; the caller will null this field.",
|
||
}
|
||
|
||
HAS_NEW_DATED_EVENT_INSTRUCTIONS = (
|
||
"这句话里是否出现了一件新的、带大概年或月的经历，且它不是对当前问题的直接回答。"
|
||
)
|
||
HAS_NEW_DATED_EVENT_CRITERIA = {
|
||
"true": (
|
||
"除了回答当前问题之外，还另说了一件带大概年或月的经历。"
|
||
),
|
||
"false": (
|
||
"没有另说新的带年月经历。"
|
||
"单纯否定、单纯记不清、或只用带年月经历来直接回答当前采集题，都不是。"
|
||
),
|
||
}
|
||
|
||
# One row per production prompt sentence that was rewritten or dropped.
|
||
CRITERION_MAP: tuple[dict[str, str], ...] = (
|
||
{
|
||
"production": "结合当前问题和动态选项判断用户是在回答当前问题、提供新的带时间经历、要求停止整个校正、询问结果，还是语义不清。",
|
||
"jev": "intent Choice with five literal criteria, including unclear as the residual option.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "若是在回答当前问题，answer_class 必须使用某个选项提供的 answer_class；否则 answer_class 必须为 null。",
|
||
"jev": "answer_class Choice is built only from the live options; code nulls it when intent != answer_current_focus.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "has_new_dated_event 仅在用户同一句里除了回答当前问题之外，还提供了新的、带大概时间的经历时为 true。",
|
||
"jev": "has_new_dated_event Noul true/false criteria, same split.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "单纯的否定或单纯的选项回答必须为 false。",
|
||
"jev": "Noul false: 单纯否定、单纯记不清不算新经历。",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "若同一句话既回答了当前问题又补充了新的带时间经历，intent 仍为 answer_current_focus，has_new_dated_event 为 true。",
|
||
"jev": "intent and Noul are separate questions; code keeps answer_current_focus when both fire.",
|
||
"lost": "Jev does not itself enforce this pair; code does.",
|
||
},
|
||
{
|
||
"production": "“当前方面没有、那段时间没有变化”通常是回答当前问题，不是停止整个流程。",
|
||
"jev": "stop_rectification criterion: only an explicit stop of the whole flow. 当前方面没有 is answer_current_focus.",
|
||
"lost": "The hedge 通常 is dropped; the stop option is written as an exclusive literal.",
|
||
},
|
||
{
|
||
"production": "只有用户明确要求停止整个校正时才分类为 stop_rectification。",
|
||
"jev": "stop_rectification criterion, same literal.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "不要按 A/B/C/D 的位置猜语义，只按选项 label 与 answer_class 判断。",
|
||
"jev": "answer_class criteria keyed by answer_class, not by A/B/C/D.",
|
||
"lost": "The prohibition itself is not sent; Jev is not given A/B/C/D as the choice keys.",
|
||
},
|
||
{
|
||
"production": "「没有、没发生过、这方面没什么」→ intent 为 answer_current_focus，answer_class 为 no。",
|
||
"jev": "Collect answer_class criterion for no.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "「记不清、不记得、忘了、想不起来、以后再说」→ answer_class 为 unsure。",
|
||
"jev": "Collect answer_class criterion for unsure.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "用户用带大概年月的经历直接回答当前采集题 → intent 为 answer_current_focus，answer_class 为 yes（程度较弱时为 weak_yes），has_new_dated_event 为 false。",
|
||
"jev": "Collect yes/weak_yes criteria + Noul false for a dated answer that is the collect answer.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "不要把「没有」或「记不清」标成 yes。",
|
||
"jev": "Separate no and unsure criteria; yes requires a dated or affirmative answer.",
|
||
"lost": "The 'do not' wording is dropped.",
|
||
},
|
||
{
|
||
"production": "若既没有否定、也没有说记不清、也没有给出带年月经历，intent 为 unclear，answer_class 必须为 null。",
|
||
"jev": "unclear residual on intent Choice; code nulls answer_class.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "若用户只在补充带时间的经历、并没有回答当前采集题，intent 为 provide_new_evidence。",
|
||
"jev": "provide_new_evidence criterion.",
|
||
"lost": "",
|
||
},
|
||
{
|
||
"production": "若同一句话既明确否定当前采集题又补充了新的带时间经历，intent 仍为 answer_current_focus 且 answer_class 为 no，不要改成 provide_new_evidence。",
|
||
"jev": "Code keeps no + Noul true. Jev intent/Noul are independent.",
|
||
"lost": "Jev may split this pair; code does not re-vote intent from the Noul.",
|
||
},
|
||
{
|
||
"production": "不要按关键词表或正则猜测，只根据当前问题与用户这句话的语义分类。",
|
||
"jev": "Not sent. Jev has no keyword table in the question.",
|
||
"lost": "The meta-instruction is dropped (Jev answers the written question, not the intended one).",
|
||
},
|
||
)
|
||
|
||
LOST_SEMANTICS: tuple[str, ...] = tuple(
|
||
row["lost"] for row in CRITERION_MAP if row["lost"]
|
||
)
|
||
|
||
|
||
def _option_rows(options: Sequence[Mapping[str, Any]] | None) -> list[dict[str, str]]:
|
||
rows: list[dict[str, str]] = []
|
||
seen: set[str] = set()
|
||
for item in options or ():
|
||
answer_class = str(item.get("answer_class") or "")
|
||
label = str(item.get("label") or "").strip()
|
||
if answer_class not in ANSWER_CLASS_VALUES or answer_class in seen:
|
||
continue
|
||
seen.add(answer_class)
|
||
rows.append({"answer_class": answer_class, "label": label})
|
||
return rows
|
||
|
||
|
||
def answer_class_criteria(
|
||
*,
|
||
layer: str,
|
||
options: Sequence[Mapping[str, Any]] | None,
|
||
) -> dict[str, str]:
|
||
"""Choice criteria keyed by answer_class. Collect uses the four fixed classes."""
|
||
if layer == "collect" and not _option_rows(options):
|
||
return {
|
||
"yes": "The message answers the collect question with a dated experience that did happen, stated firmly.",
|
||
"weak_yes": "The message answers the collect question with a dated or tentative experience, weakly.",
|
||
"no": "The message says this topic did not happen / there was no change. Close this aspect.",
|
||
"unsure": "The message says the user does not remember, cannot recall, or wants to skip for now.",
|
||
}
|
||
criteria: dict[str, str] = {}
|
||
for row in _option_rows(options):
|
||
criteria[row["answer_class"]] = (
|
||
f"user_message matches this option's meaning. label={row['label']}"
|
||
)
|
||
if not criteria:
|
||
criteria = {
|
||
"yes": "Affirmative answer to current_question.",
|
||
"weak_yes": "Weak affirmative answer to current_question.",
|
||
"no": "Negative answer to current_question.",
|
||
"unsure": "User does not remember the answer to current_question.",
|
||
}
|
||
return criteria
|
||
|
||
|
||
def build_state(sample: Mapping[str, Any]) -> dict[str, Any]:
|
||
focus = sample.get("focus") if isinstance(sample.get("focus"), dict) else {}
|
||
options = list(focus.get("options") or [])
|
||
return {
|
||
"current_question": str(focus.get("current_question") or ""),
|
||
"options": options,
|
||
"user_message": str(sample.get("user_message") or ""),
|
||
"case_status": str(focus.get("case_status") or sample.get("case_status") or "collecting_evidence"),
|
||
}
|
||
|
||
|
||
def build_questions(sample: Mapping[str, Any]) -> dict[str, Any]:
|
||
"""Three TypeSafe questions, same request, official fan-out."""
|
||
layer = str(sample.get("layer") or "none")
|
||
focus = sample.get("focus") if isinstance(sample.get("focus"), dict) else {}
|
||
options = list(focus.get("options") or [])
|
||
return {
|
||
"intent": {
|
||
"type": "choice",
|
||
"instructions": INTENT_INSTRUCTIONS,
|
||
"criteria": dict(INTENT_CRITERIA),
|
||
},
|
||
"answer_class": {
|
||
"type": "choice",
|
||
"instructions": ANSWER_CLASS_INSTRUCTIONS,
|
||
"criteria": answer_class_criteria(layer=layer, options=options),
|
||
},
|
||
"has_new_dated_event": {
|
||
"type": "noul",
|
||
"instructions": HAS_NEW_DATED_EVENT_INSTRUCTIONS,
|
||
"criteria": dict(HAS_NEW_DATED_EVENT_CRITERIA),
|
||
},
|
||
}
|
||
|
||
|
||
def sdk_questions(sample: Mapping[str, Any]) -> dict[str, Any]:
|
||
"""SDK objects when typesafe_sdk is installed; JSON dicts otherwise."""
|
||
payload = build_questions(sample)
|
||
try:
|
||
from typesafe_sdk import Choice, Noul
|
||
except ImportError:
|
||
return payload
|
||
return {
|
||
"intent": Choice(
|
||
instructions=payload["intent"]["instructions"],
|
||
criteria=payload["intent"]["criteria"],
|
||
),
|
||
"answer_class": Choice(
|
||
instructions=payload["answer_class"]["instructions"],
|
||
criteria=payload["answer_class"]["criteria"],
|
||
),
|
||
"has_new_dated_event": Noul(
|
||
instructions=payload["has_new_dated_event"]["instructions"],
|
||
criteria=payload["has_new_dated_event"]["criteria"],
|
||
),
|
||
}
|
||
|
||
|
||
def enforce_combo(
|
||
intent: str | None,
|
||
answer_class: str | None,
|
||
has_new_dated_event: bool | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Production invariant: answer_class is non-null iff intent is answer_current_focus."""
|
||
intent_value = intent if intent in INTENT_VALUES else "unclear"
|
||
dated = bool(has_new_dated_event)
|
||
if intent_value != "answer_current_focus":
|
||
return {
|
||
"intent": intent_value,
|
||
"answer_class": None,
|
||
"has_new_dated_event": dated,
|
||
}
|
||
class_value = answer_class if answer_class in ANSWER_CLASS_VALUES else None
|
||
if class_value is None:
|
||
return {
|
||
"intent": "unclear",
|
||
"answer_class": None,
|
||
"has_new_dated_event": dated,
|
||
}
|
||
return {
|
||
"intent": intent_value,
|
||
"answer_class": class_value,
|
||
"has_new_dated_event": dated,
|
||
}
|
||
|
||
|
||
def parse_jev_answers(answers: Mapping[str, Any]) -> dict[str, Any]:
|
||
intent_row = answers.get("intent") or {}
|
||
class_row = answers.get("answer_class") or {}
|
||
noul_row = answers.get("has_new_dated_event") or {}
|
||
noul_value = noul_row.get("noul")
|
||
dated = None
|
||
if isinstance(noul_value, (int, float)):
|
||
dated = float(noul_value) >= NOUL_TRUE_THRESHOLD
|
||
parsed = enforce_combo(
|
||
str(intent_row.get("choice") or "") or None,
|
||
str(class_row.get("choice") or "") or None,
|
||
dated,
|
||
)
|
||
parsed["raw"] = {
|
||
"intent": {
|
||
"choice": intent_row.get("choice"),
|
||
"probabilities": intent_row.get("probabilities"),
|
||
"confidence": intent_row.get("confidence"),
|
||
},
|
||
"answer_class": {
|
||
"choice": class_row.get("choice"),
|
||
"probabilities": class_row.get("probabilities"),
|
||
"confidence": class_row.get("confidence"),
|
||
},
|
||
"has_new_dated_event": {
|
||
"noul": noul_value,
|
||
},
|
||
}
|
||
intent_conf = intent_row.get("confidence")
|
||
class_conf = class_row.get("confidence")
|
||
if parsed["intent"] == "answer_current_focus" and isinstance(intent_conf, (int, float)) and isinstance(class_conf, (int, float)):
|
||
parsed["confidence"] = min(float(intent_conf), float(class_conf))
|
||
elif isinstance(intent_conf, (int, float)):
|
||
parsed["confidence"] = float(intent_conf)
|
||
else:
|
||
parsed["confidence"] = None
|
||
return parsed
|