reference_image_gate.py - JoyAI-Echo

reference_image_gate.py

1	"""Deterministic first-frame upload-gate matching (PDF truth table)."""
2
3	from __future__ import annotations
4
5	from dataclasses import dataclass
6	from typing import Any, Literal
7
8	SESSION_GATE_DECLINE_STREAK_KEY = "reference_image_gate_decline_streak"
9	SESSION_GATE_SKIP_NEXT_KEY = "reference_image_gate_skip_next_message"
10	SESSION_GATE_PENDING_QUESTION_KEY = "reference_image_gate_pending_question"
11
12	CONFIRM_LABEL = "需要上传,已上传完毕"
13	DECLINE_LABEL = "不上传"
14	GATE_OPTIONS = (CONFIRM_LABEL, DECLINE_LABEL)
15
16	CONFIRM_ALIASES = frozenset({CONFIRM_LABEL, "已上传完毕", "已上传", "确认上传"})
17	DECLINE_ALIASES = frozenset({DECLINE_LABEL, "无需上传"})
18
19	MISMATCH_MISSING_QUESTION = "识别到未上传参考图，是否还需上传参考图？"
20	MISMATCH_PRESENT_QUESTION = "识别到已上传参考图，是否确认使用此参考图？"
21	INITIAL_GATE_QUESTION = "在开始构思之前，是否要上传首帧参考图？"
22
23	MATCH_INJECT_PREFIX = (
24	"REFERENCE_IMAGE_GATE match=true present={present} intent={intent}. "
25	"The user's upload-gate choice matches persisted first-frame state. "
26	"Continue conceiving the story. Reply 「接下来我们开始构思吧，你想要什么样的故事呢？」 "
27	"when they have not given a story idea yet. Do not re-ask whether to upload "
28	"unless they pick 我想修改/增删参考图."
29	)
30
31	Intent = Literal["confirm", "decline", "other"]
32
33
34	@dataclass(frozen=True)
35	class UploadGateResult:
36	intent: Intent
37	match: bool
38	skip_agent: bool
39	delete_image: bool
40	reset_streak: bool
41	next_streak: int
42	mismatch_question: str \| None
43	inject_note: str \| None
44
45
46	def classify_upload_gate_intent(answer: str) -> Intent:
47	label = (answer or "").strip()
48	if label in CONFIRM_ALIASES:
49	return "confirm"
50	if label in DECLINE_ALIASES:
51	return "decline"
52	return "other"
53
54
55	def is_upload_gate_answer(answer: str) -> bool:
56	return classify_upload_gate_intent(answer) != "other"
57
58
59	def decline_streak(metadata: dict[str, Any] \| None) -> int:
60	if not isinstance(metadata, dict):
61	return 0
62	try:
63	return max(0, int(metadata.get(SESSION_GATE_DECLINE_STREAK_KEY) or 0))
64	except (TypeError, ValueError):
65	return 0
66
67
68	def evaluate_upload_gate(
69	*,
70	present: bool,
71	answer: str,
72	streak: int = 0,
73	) -> UploadGateResult:
74	"""Match option intent against persisted image presence.
75
76	Second consecutive decline while present=true is a match and deletes the image.
77	"""
78	intent = classify_upload_gate_intent(answer)
79	if intent == "other":
80	return UploadGateResult(
81	intent=intent,
82	match=True,
83	skip_agent=False,
84	delete_image=False,
85	reset_streak=False,
86	next_streak=streak,
87	mismatch_question=None,
88	inject_note=None,
89	)
90	if intent == "confirm":
91	if present:
92	return UploadGateResult(
93	intent=intent,
94	match=True,
95	skip_agent=False,
96	delete_image=False,
97	reset_streak=True,
98	next_streak=0,
99	mismatch_question=None,
100	inject_note=MATCH_INJECT_PREFIX.format(present="true", intent="confirm"),
101	)
102	return UploadGateResult(
103	intent=intent,
104	match=False,
105	skip_agent=True,
106	delete_image=False,
107	reset_streak=True,
108	next_streak=0,
109	mismatch_question=MISMATCH_MISSING_QUESTION,
110	inject_note=None,
111	)
112	# decline
113	if not present:
114	return UploadGateResult(
115	intent=intent,
116	match=True,
117	skip_agent=False,
118	delete_image=False,
119	reset_streak=True,
120	next_streak=0,
121	mismatch_question=None,
122	inject_note=MATCH_INJECT_PREFIX.format(present="false", intent="decline"),
123	)
124	if streak >= 1:
125	return UploadGateResult(
126	intent=intent,
127	match=True,
128	skip_agent=False,
129	delete_image=True,
130	reset_streak=True,
131	next_streak=0,
132	mismatch_question=None,
133	inject_note=MATCH_INJECT_PREFIX.format(present="false", intent="decline"),
134	)
135	return UploadGateResult(
136	intent=intent,
137	match=False,
138	skip_agent=True,
139	delete_image=False,
140	reset_streak=False,
141	next_streak=streak + 1,
142	mismatch_question=MISMATCH_PRESENT_QUESTION,
143	inject_note=None,
144	)
145
146
147	def commit_upload_gate(metadata: dict[str, Any], result: UploadGateResult) -> None:
148	"""Write streak / skip-next flags onto session metadata."""
149	if result.reset_streak or result.next_streak == 0:
150	metadata.pop(SESSION_GATE_DECLINE_STREAK_KEY, None)
151	if result.next_streak > 0:
152	metadata[SESSION_GATE_DECLINE_STREAK_KEY] = result.next_streak
153	if result.skip_agent:
154	metadata[SESSION_GATE_SKIP_NEXT_KEY] = True
155	if result.mismatch_question:
156	metadata[SESSION_GATE_PENDING_QUESTION_KEY] = result.mismatch_question
157	else:
158	metadata.pop(SESSION_GATE_SKIP_NEXT_KEY, None)
159	metadata.pop(SESSION_GATE_PENDING_QUESTION_KEY, None)
160
161
162	def consume_skip_next_message(metadata: dict[str, Any] \| None) -> str \| None:
163	"""Pop skip-next and return the pending mismatch question, if any."""
164	if not isinstance(metadata, dict):
165	return None
166	skip = bool(metadata.pop(SESSION_GATE_SKIP_NEXT_KEY, False))
167	question = metadata.pop(SESSION_GATE_PENDING_QUESTION_KEY, None)
168	if skip and isinstance(question, str) and question.strip():
169	return question.strip()
170	return None
171
172
173	def mismatch_card(question: str) -> dict[str, Any]:
174	return {
175	"id": "upload-gate",
176	"question": question,
177	"options": [{"label": CONFIRM_LABEL}, {"label": DECLINE_LABEL}],
178	"allow_custom": False,
179	"status": "pending",
180	"answered": None,
181	}
182

182 lines PYTHON