Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 67 additions & 1 deletion apps/api/app/services/tools/ncs_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,11 @@
from __future__ import annotations

import json
import re
from copy import deepcopy
from typing import Any

from app.services.calculation_policy import _mask_quoted_intent
from app.services.ncs_submission import submitted_choice_from_request
from app.services.tools import arithmetic
from app.services.tools.base import Tool, ToolContext, ToolResult
Expand All @@ -29,6 +31,22 @@
"계산기에 입력된 식과 선지에서 유일한 정답을 확인하지 못했습니다. "
"채점을 보류하고 문제 조건·계산식·단위·선지를 다시 확인해야 합니다."
)
_REVIEW_REQUEST = re.compile(
r"(?:문항|문제|선지|선택지)[^.!?\n]{0,24}(?:검토|검증|점검)|"
r"정답[^.!?\n]{0,24}(?:하나|유일|복수)|(?:중복|동치)[^.!?\n]{0,12}(?:선지|선택지)"
)
_CREATE_QUESTION = re.compile(
r"출제해|출제하|(?:문항|문제|퀴즈)(?:을|를)?\s*"
r"(?:(?:새로|다시|하나|한\s*개)\s*)?(?:만들|생성|작성)"
)
_WITHHOLD_ANSWER = re.compile(
r"(?:정답|답|해설)[^.!?\n]{0,20}(?:"
r"(?:알려(?:주)?|말하|공개하|보여주|제시하|표시하|쓰)지(?:는|도)?\s*(?:마|말)|"
r"나중|숨겨|없이)"
)
_CHOICE_MARKER = re.compile(r"선택지|선지")
_CHOICE_LABEL = re.compile(r"(?<!\d)(10|[1-9])\s*번")
_CHOICE_NUMBER = re.compile(r"(?<![\d.])[+-]?(?:\d+(?:\.\d+)?|\.\d+)")
_ERROR = json.dumps(
{
"error": "invalid_ncs_check",
Expand Down Expand Up @@ -59,6 +77,53 @@ def _invalid() -> ToolResult:
return ToolResult(content=_ERROR, detail="문항 검산 미완료", failed=True)


def _reviews_supplied_choices(ctx: ToolContext | None) -> bool:
"""Enable details only for a bounded, explicit review of supplied numeric options."""
request = ctx.request if ctx is not None else None
if not isinstance(request, str) or len(request) > 8192:
return False
intent = _mask_quoted_intent(request)
if (
not _REVIEW_REQUEST.search(intent)
or _CREATE_QUESTION.search(intent) or _WITHHOLD_ANSWER.search(intent)
):
return False
marker = _CHOICE_MARKER.search(request)
if marker is None:
return False
choices = request[marker.end():]
labels = list(_CHOICE_LABEL.finditer(choices))
supplied = set()
for index, label in enumerate(labels):
end = labels[index + 1].start() if index + 1 < len(labels) else len(choices)
# Labels alone can be a request to generate options, not supplied numeric data.
if _CHOICE_NUMBER.search(choices[label.end():min(end, label.end() + 80)]):
supplied.add(label.group(1))
return len(supplied) >= 2


def _non_unique_text(data: dict[str, Any], ctx: ToolContext | None) -> str:
if not _reviews_supplied_choices(ctx):
return _NON_UNIQUE
if data["decimal_places"] is None:
value = f"계산기에 입력된 식의 값은 {data['value']}입니다. "
else:
value = (
f"계산기에 입력된 식의 정확한 값은 {data['exact']}이며, "
f"소수점 {data['decimal_places']}자리 반올림값은 {data['value']}입니다. "
)
if data["choice_status"] == "ambiguous":
indices = ", ".join(f"{index}번" for index in data["matched_choices"])
mismatch = f"입력된 선지 {indices}이 이 값과 일치해 "
else:
mismatch = "입력된 선지 중 이 값과 일치하는 것이 없어 "
return (
value + mismatch + "유일한 정답을 고를 수 없습니다. 채점은 보류합니다. "
"이는 입력된 식의 산술 결과이며, 식·단위·선지가 실제 문항을 정확히 반영했는지는 "
"별도로 확인해야 합니다."
)


def _normalize_choices(raw: object) -> list[str] | None:
if isinstance(raw, str):
try:
Expand Down Expand Up @@ -127,7 +192,8 @@ async def check_ncs_answer(arguments: dict[str, Any], ctx: ToolContext | None =
detail="문항 검산 미완료" if result.failed else "문항 검산 완료",
failed=result.failed,
final_text=(
_NON_UNIQUE if data.get("choice_status") in {"no_match", "ambiguous"} else None
_non_unique_text(data, ctx)
if data.get("choice_status") in {"no_match", "ambiguous"} else None
),
)

Expand Down
134 changes: 134 additions & 0 deletions apps/api/tests/test_ncs_review_diagnostics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
"""A supplied-question review can explain ambiguity without grading a learner."""

import json

import pytest
from test_ncs_preflight import _call, _stream, _text

from app.services import agent
from app.services.tools.base import ToolContext
from app.services.tools.ncs_check import CHECK_NCS_ANSWER, check_ncs_answer

REVIEW = (
"NCS 문항을 검토해줘. '20% 증가한 뒤 20% 감소한 값은 원래 값과 비교하면?' "
"선택지: 1번 4% 감소, 2번 원래의 96%, 3번 변화 없음, 4번 4% 증가. "
"정답이 하나로 정해지는지도 설명해줘."
)
ARGUMENTS = {
"decision": "calculate",
"expression": "(1 + 0.2) * (1 - 0.2)",
"choices": ["0.96", "0.96", "1", "1.04"],
}


def _context(prompt):
return ToolContext(user_id="learner", session_id="review", request=prompt)


async def test_supplied_equivalent_choices_keep_value_and_matching_indices():
output = await check_ncs_answer(ARGUMENTS, _context(REVIEW))
data = json.loads(output.content)
assert data["exact"] == "24/25" and data["value"] == "0.96"
assert data["matched_choices"] == [1, 2] and data["answer"] is None
assert data["grading"] == "not_requested"
assert "0.96" in output.final_text
assert "1번, 2번" in output.final_text
assert "유일한 정답" in output.final_text and "채점은 보류" in output.final_text
assert "실제 문항" in output.final_text and "별도로 확인" in output.final_text
assert "0.96" not in output.detail


async def test_no_match_review_distinguishes_a_missing_matching_choice():
output = await check_ncs_answer(
{"decision": "calculate", "expression": "1600/20", "choices": ["79", "82.5"]},
_context("다음 문항을 검토해줘. 선택지: 1번 79, 2번 82.5. 정답이 유일한가요?"),
)
assert "80" in output.final_text
assert "일치하는 것이 없어" in output.final_text
assert "채점은 보류" in output.final_text
assert json.loads(output.content)["answer"] is None


async def test_review_explains_when_duplicate_matches_use_requested_rounding():
output = await check_ncs_answer(
{"decision": "calculate", "expression": "1/3", "choices": ["0.33", "33/100"],
"decimal_places": 2},
_context("선지 검토해줘. 선택지: 1번 0.33, 2번 33/100. 소수점 2자리로 반올림."),
)
assert "1/3" in output.final_text
assert "소수점 2자리 반올림값은 0.33" in output.final_text
assert "1번, 2번" in output.final_text


@pytest.mark.parametrize(
"prompt",
[
None,
"문항을 만들어줘. 정답은 아직 알려주지 마.",
"선택지 1번, 2번이 있는 새 문항을 만들어줘. 정답이 유일한지도 검토해줘.",
"선지 1번, 2번인 퀴즈를 작성해줘. 정답이 하나인지 확인해줘.",
"문항을 검토해줘. 선택지는 모델이 채워줘.",
"문항을 검토해줘. 선택지: 1번 4% 감소.",
"문항을 검토해줘. 선택지: 1번 4% 감소, 1번 원래의 96%.",
"선택지: 1번 4% 감소, 2번 원래의 96%. 문제만 그대로 보여줘.",
"선택지 1번과 2번을 검토해줘.",
REVIEW + " 같은 유형의 새 문항을 만들어줘.",
REVIEW + " 새로 출제해줘.",
REVIEW + " 정답은 공개하지 마.",
REVIEW + " 정답을 말하지는 마.",
REVIEW + " 정답은 나중에 알려줘.",
'다음 문장을 그대로 번역해줘: "문항 검토해줘. 선택지 1번 0.96, 2번 0.96."',
],
)
async def test_practice_or_incomplete_review_never_discloses_computed_values(prompt):
output = await check_ncs_answer(ARGUMENTS, _context(prompt))
assert "0.96" not in output.final_text
assert "1번" not in output.final_text and "2번" not in output.final_text
assert not any(character.isdigit() for character in output.final_text)
assert json.loads(output.content)["answer"] is None


async def test_model_supplied_review_claim_cannot_enable_visible_diagnostics():
output = await check_ncs_answer({**ARGUMENTS, "userRequested": True}, _context(REVIEW))
assert output.failed and output.final_text is None
assert "0.96" not in output.content


async def test_review_does_not_assign_correct_or_incorrect_to_an_explicit_submission():
prompt = REVIEW.replace("'", "") + " 제 답은 1번입니다."
output = await check_ncs_answer(ARGUMENTS, _context(prompt))
data = json.loads(output.content)
assert data["submission_status"] == "explicit"
assert data["grading"] == "ambiguous" and data["answer"] is None
assert "0.96" in output.final_text
assert "정답입니다" not in output.final_text and "오답입니다" not in output.final_text
assert "채점은 보류" in output.final_text


@pytest.mark.parametrize("mask", [False, True])
async def test_real_agent_returns_diagnostics_once_after_sanitizing_without_another_hop(
monkeypatch, mask,
):
snapshots = []
_stream(monkeypatch, [{"text": ["UNVERIFIED_DRAFT"], "calls": [
_call(arguments=json.dumps(ARGUMENTS)),
]}], snapshots)

def sanitize(text):
return text.replace("0.96", "[MASKED_VALUE]"), text.count("0.96")

events = [event async for event in agent.run_turn(
"synthetic/model", [{"role": "user", "content": REVIEW}], [CHECK_NCS_ANSWER],
_context(REVIEW), preflight_tool="check_ncs_answer", calculation_required=True,
**({"sanitize_tool_output": sanitize} if mask else {}),
)]
text = _text(events)
assert len(snapshots) == 1
assert "1번, 2번" in text and "채점은 보류" in text
assert "UNVERIFIED_DRAFT" not in text
assert ("[MASKED_VALUE]" if mask else "0.96") in text
if mask:
assert "0.96" not in text
assert [event for event in events if event["type"] == "usage"] == [
{"type": "usage", "inputTokens": 7, "outputTokens": 11},
]
Loading