Zareian
33d8f43d74
The model ends a shape question with <diagram>{json}</diagram>; rag/services/diagram.py
splits it off before mathtext, validates it against a whitelist and stores it on
Message.diagram (ai_tutor.0006). It reaches the app in the SSE done event and in
conversation history. Prompt tutor-v6-diagrams; a hint next to the question (shape
words only) raised gpt-4.1-mini from 3/12 to 12/12 diagrams, 0/8 on arithmetic.
progress.0005 adds tedi (600 xp) to the companion catalog.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
172 lines
8.6 KiB
Python
172 lines
8.6 KiB
Python
"""شکلِ پاسخِ دستیار: جدا کردن از متن، اعتبارسنجی، و رسیدنش تا SSE و تاریخچه."""
|
||
import json
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
from django.core.cache import cache
|
||
from django.test import SimpleTestCase, TestCase, override_settings
|
||
from rest_framework.test import APIClient
|
||
|
||
from accounts.authentication import issue_access_token
|
||
from accounts.models import User
|
||
|
||
from .services.clients import reset_clients_cache
|
||
from .services.diagram import StreamSplitter, extract, validate
|
||
from .tests import RAG_ON, _parse_sse
|
||
|
||
CIRCLE = {
|
||
'kind': 'plane', 'title': 'زاویهی ۳۰ درجه', 'axes': True,
|
||
'points': {'O': [0, 0], 'A': [1, 0], 'P': {'polar': [1, 30]}},
|
||
'items': [
|
||
{'type': 'circle', 'center': 'O', 'r': 1},
|
||
{'type': 'segment', 'from': 'O', 'to': 'P'},
|
||
{'type': 'angle', 'vertex': 'O', 'from': 'A', 'to': 'P', 'label': '30°'},
|
||
],
|
||
}
|
||
ANSWER = 'در شکلِ پایین نقطهی P را میبینی.\n✅ جواب: 1/2\n<diagram>\n' + json.dumps(CIRCLE, ensure_ascii=False) + '\n</diagram>'
|
||
|
||
|
||
class ValidateTests(SimpleTestCase):
|
||
def test_real_model_output_passes_unchanged_in_meaning(self):
|
||
spec = validate(CIRCLE)
|
||
self.assertEqual(spec['points']['P'], {'polar': [1.0, 30.0]})
|
||
self.assertEqual(len(spec['items']), 3)
|
||
|
||
def test_unknown_item_type_is_dropped_not_the_whole_diagram(self):
|
||
spec = validate({**CIRCLE, 'items': CIRCLE['items'] + [{'type': 'html', 'html': '<script>x</script>'}]})
|
||
self.assertEqual([i['type'] for i in spec['items']], ['circle', 'segment', 'angle'])
|
||
|
||
def test_function_expression_only_allows_math(self):
|
||
ok = validate({'kind': 'plane', 'items': [{'type': 'function', 'expr': 'sin(x)+x^2', 'from': -3, 'to': 3}]})
|
||
self.assertEqual(ok['items'][0]['expr'], 'sin(x)+x^2')
|
||
for bad in ('alert(1)', 'x; fetch()', 'constructor', '__proto__'):
|
||
self.assertIsNone(validate({'kind': 'plane', 'items': [{'type': 'function', 'expr': bad}]}), bad)
|
||
|
||
def test_numbers_must_be_finite_and_bounded(self):
|
||
self.assertIsNone(validate({'kind': 'numberline', 'min': 0, 'max': 1e9}))
|
||
spec = validate({'kind': 'plane', 'points': {'A': [float('inf'), 0], 'B': [1, 1]}, 'items': []})
|
||
self.assertEqual(list(spec['points']), ['B'])
|
||
|
||
def test_labels_are_truncated_and_fraction_shaded_is_capped(self):
|
||
spec = validate({'kind': 'fraction', 'parts': 4, 'shaded': 9, 'title': 'x' * 500})
|
||
self.assertIsNone(spec)
|
||
spec = validate({'kind': 'fraction', 'parts': 4, 'shaded': 3, 'title': 'x' * 500})
|
||
self.assertEqual((spec['parts'], spec['shaded'], len(spec['title'])), (4, 3, 40))
|
||
|
||
def test_garbage_is_none(self):
|
||
for bad in (None, [], 'svg', {'kind': 'svg'}, {'kind': 'plane', 'items': 'x'}):
|
||
self.assertIsNone(validate(bad), bad)
|
||
|
||
|
||
class ExtractTests(SimpleTestCase):
|
||
def test_block_is_removed_from_text(self):
|
||
text, spec = extract(ANSWER)
|
||
self.assertNotIn('diagram', text)
|
||
self.assertNotIn('{', text)
|
||
self.assertTrue(text.endswith('✅ جواب: 1/2'))
|
||
self.assertEqual(spec['kind'], 'plane')
|
||
|
||
def test_broken_json_drops_only_the_diagram(self):
|
||
text, spec = extract('جواب ۵ است.\n<diagram>{"kind": "plane", </diagram>')
|
||
self.assertEqual((text, spec), ('جواب ۵ است.', None))
|
||
|
||
def test_unclosed_block_never_leaks_json(self):
|
||
text, _ = extract('جواب.\n<diagram>{"kind":"fraction","parts":4,"shaded":3}')
|
||
self.assertEqual(text, 'جواب.')
|
||
|
||
def test_splitter_matches_extract_however_the_stream_is_cut(self):
|
||
expected_text, expected_spec = extract(ANSWER)
|
||
for step in (1, 2, 3, 5, 9, 40, 1000):
|
||
splitter = StreamSplitter()
|
||
out = ''.join(splitter.feed(ANSWER[i:i + step]) for i in range(0, len(ANSWER), step))
|
||
out += splitter.flush()
|
||
self.assertEqual(out.rstrip(), expected_text, f'step={step}')
|
||
self.assertEqual(splitter.diagram, expected_spec, f'step={step}')
|
||
|
||
def test_splitter_passes_text_after_the_block(self):
|
||
splitter = StreamSplitter()
|
||
out = splitter.feed('الف <diagram>{"kind":"fraction","parts":2,"shaded":1}</diagram> ب') + splitter.flush()
|
||
self.assertEqual(out, 'الف ب')
|
||
self.assertEqual(splitter.diagram['parts'], 2)
|
||
|
||
def test_lone_less_than_sign_is_not_swallowed(self):
|
||
splitter = StreamSplitter()
|
||
out = splitter.feed('x <') + splitter.feed(' 3') + splitter.flush()
|
||
self.assertEqual(out, 'x < 3')
|
||
|
||
|
||
def _fake_stream(text: str, step: int = 7):
|
||
events = [
|
||
SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content=text[i:i + step]))])
|
||
for i in range(0, len(text), step)
|
||
]
|
||
create = lambda **kwargs: iter(events) # noqa: E731
|
||
return SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create)))
|
||
|
||
|
||
@override_settings(**RAG_ON)
|
||
class DiagramThroughTheApiTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
reset_clients_cache()
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(device_id='dev-diagram', is_guest=True, name='')
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}')
|
||
|
||
def test_stream_sends_text_without_json_and_diagram_in_done_and_history(self):
|
||
from ai_tutor.models import Message
|
||
|
||
with patch('rag.services.retrieval.retrieve_context', return_value=([], {})), \
|
||
patch('rag.services.clients.get_openai_client', return_value=_fake_stream(ANSWER)):
|
||
resp = self.client.post('/api/ai/ask', {'userQuestion': 'سینوسِ ۳۰؟', 'stream': True}, format='json')
|
||
events = _parse_sse(resp.streaming_content) # استریم همینجا اجرا میشود، داخلِ patch
|
||
text = ''.join(json.loads(d)['delta'] for e, d in events if e == 'chunk')
|
||
self.assertNotIn('<diagram', text)
|
||
self.assertNotIn('"kind"', text)
|
||
self.assertEqual(events[-1][0], 'done', events[-1])
|
||
done = json.loads(events[-1][1])
|
||
self.assertEqual(done['diagram']['points']['P'], {'polar': [1.0, 30.0]})
|
||
|
||
message = Message.objects.get(role='assistant', conversation__user_id=self.user.id)
|
||
self.assertEqual(message.diagram, done['diagram'])
|
||
self.assertEqual(message.prompt_version, 'tutor-v6-diagrams')
|
||
history = self.client.get(f'/api/ai/conversations/{done["conversationId"]}/messages').json()
|
||
self.assertEqual(history['messages'][-1]['diagram']['kind'], 'plane')
|
||
self.assertIsNone(history['messages'][0]['diagram'])
|
||
|
||
def test_answer_without_diagram_has_null(self):
|
||
with patch('rag.services.retrieval.retrieve_context', return_value=([], {})), \
|
||
patch('rag.services.clients.get_openai_client', return_value=_fake_stream('۲۳ × ۴۵ = ۱۰۳۵')):
|
||
resp = self.client.post('/api/ai/ask', {'userQuestion': '۲۳×۴۵', 'stream': True}, format='json')
|
||
done = json.loads(_parse_sse(resp.streaming_content)[-1][1])
|
||
self.assertIsNone(done['diagram'])
|
||
|
||
|
||
class ShapeHintTests(SimpleTestCase):
|
||
def test_hint_goes_next_to_shape_questions_only(self):
|
||
from .services.diagram import DIAGRAM_HINT, hint_for
|
||
from .services.generation import _user_prompt
|
||
|
||
for q in ('وترِ مثلث با ضلعهای ۳ و ۴؟', 'سهچهارم پیتزا', 'نمودار y=x²', 'sin 30°'):
|
||
self.assertEqual(hint_for(q), DIAGRAM_HINT, q)
|
||
self.assertIn(DIAGRAM_HINT, _user_prompt(q, []))
|
||
for q in ('۲۳ × ۴۵ چند میشود؟', 'عبارت (x+2)(x-3) را ساده کن.'):
|
||
self.assertEqual(hint_for(q), '', q)
|
||
self.assertEqual(_user_prompt(q, []), q)
|
||
|
||
def test_every_system_prompt_carries_the_diagram_rule(self):
|
||
from .services.generation import system_prompt
|
||
|
||
for grade in ('سوم', 'هشتم', 'دوازدهم', ''):
|
||
self.assertIn('<diagram>', system_prompt(grade))
|
||
|
||
|
||
class AxesDefaultTests(SimpleTestCase):
|
||
def test_geometry_without_axes_key_gets_no_axes_but_graphs_do(self):
|
||
tri = validate({'kind': 'plane', 'points': {'A': [0, 0], 'B': [3, 0], 'C': [0, 4]},
|
||
'items': [{'type': 'polygon', 'points': ['A', 'B', 'C']}]})
|
||
self.assertFalse(tri['axes'])
|
||
graph = validate({'kind': 'plane', 'items': [{'type': 'function', 'expr': 'x^2', 'from': -2, 'to': 2}]})
|
||
self.assertTrue(graph['axes'])
|
||
self.assertTrue(validate({**CIRCLE})['axes']) # صریحاً true خواسته شده
|