""" ساختِ سوال‌های چهارگزینه‌ای از مسئله‌های استخراج‌شده‌ی کتاب. `seed_curriculum_from_textbook` اسکلت (Grade/Unit/Node/Lesson) را از دادهٔ موجود و بدونِ هزینه می‌سازد؛ این فرمان تکه‌ی گران را اضافه می‌کند: برای هر درس یک سوال با سه گزینه‌ی انحرافیِ **باورپذیر**. چرا مدل لازم است و مکانیکی نمی‌شود: گزینه‌ی انحرافیِ خوب باید همان اشتباهی باشد که دانش‌آموز واقعاً می‌کند (علامتِ توان را جابه‌جا کند، مخرج مشترک نگیرد). عددِ تصادفی یا «جوابِ درست ± ۱» در دمو فوراً تابلو می‌شود و کلِ سنجشِ کیفیت را بی‌معنا می‌کند. python manage.py generate_questions_from_chunks --grade نهم --limit 5 --dry-run python manage.py generate_questions_from_chunks --grade نهم --limit 20 python manage.py generate_questions_from_chunks --grade نهم مسیرِ رایگان (و پیش‌فرضِ عملی): `--export DIR` هیچ فراخوانیِ پولی نمی‌کند و فقط جفت‌های (درس، مسئله‌ی تأییدشده) را در چند فایلِ JSON می‌ریزد تا subagentها سوال‌ها را بنویسند و خروجی با `content.question_io` وارد شود. اندازه‌گیریِ ۱۴۰۵/۰۶/۲۹: o4-mini روی این پروکسی ۰٫۷۷ سنت برای هر سوال، subagentِ Sonnet صفر — با بازده‌ی ۹۲٪. متنِ پرامپت برای هر دو مسیر یکی است (همین PROMPT پایین)، پس کیفیت دو شاخه از هم جدا نمی‌افتد. """ import json import os import threading import time from concurrent.futures import ThreadPoolExecutor from django.core.management.base import BaseCommand, CommandError from django.db import transaction from content.models import Choice, Grade, Lesson, Question from rag.models import LessonChunk from rag.services import clients from rag.services.config import build_collection_name, rag_is_configured from rag.services.textbooks import TEXTBOOKS, get_textbook def _is_account_limit(exc) -> bool: """ آیا این خطا یعنی «حسابِ سرویس دیگر جواب نمی‌دهد» یا فقط کندیِ لحظه‌ای؟ دو شکلِ واقعی که دیدیم و هر دو باید اجرا را متوقف کنند: 429 initial_usage_limit_exceeded — سقفِ حسابِ تازه، ساعت‌ها 403 insufficient user quota — اعتبار تمام شده نکته: متنِ دومی «insufficient user quota» است با فاصله، نه زیرخط. محافظِ اولِ ما دنبالِ `insufficient_quota` می‌گشت و نمی‌گرفتش، و اجرا ۷۷۰ فراخوانیِ محکوم‌به‌شکست زد به‌جای اینکه سرِ اولی بایستد. """ text = str(exc).lower() if 'usage_limit' in text or 'quota' in text: return True return '403' in text and 'insufficient' in text PROMPT = """تو یک معلمِ ریاضیِ باتجربه‌ی پایه‌ی {grade} هستی. از روی «مسئله» و «راه‌حل»ِ زیر — که مستقیماً از کتابِ درسی درآمده و درستیِ راه‌حلش جداگانه بررسی و تأیید شده — یک سوالِ چهارگزینه‌ای بساز. ⚠️ نکته‌ی نگارشی: «/» بینِ دو عدد در این متن‌ها جداکننده‌ی اعشار است، نه خطِ کسر. یعنی «۲/۵» عددِ ۲٫۵ است. قواعد — نقضِ هرکدام یعنی سوالِ بی‌ارزش: - **خودکفا:** تمامِ داده‌ی لازم باید داخلِ صورتِ سوال باشد. «مساحتِ شکل چند است؟» بی‌ارزش است چون دانش‌آموز شکل را نمی‌بیند. اگر مسئله بدونِ تصویرِ کتاب قابلِ بازنویسی نیست، دقیقاً {{"skip": true}} برگردان. - **دقیقاً یک گزینه‌ی درست.** هر سه گزینه‌ی دیگر را جداگانه بسنج و مطمئن شو قطعاً غلط‌اند. تله‌ی رایج: در «کدام مجموعه تهی است؟» چند گزینه تهی از آب دربیایند، یا دو گزینه از نظر عددی برابر باشند و فقط شکلِ نوشتنشان فرق کند (مثلاً «√۷/√۶» و «√۴۲/۶» یک عددند). - **گزینه‌های غلط باید خطای واقعیِ دانش‌آموز باشند** — جابه‌جایی علامتِ توان، مخرجِ مشترک نگرفتن، یک مرحله جا انداختن. نه عددِ تصادفی، نه گزینه‌ی مسخره. - هیچ دو گزینه‌ای هم‌معنا نباشند. - **بدونِ LaTeX و بدونِ markdown.** به‌جای \\frac{{a}}{{b}} بنویس (a)/(b). متن عیناً همان‌طور که می‌نویسی به دانش‌آموز نشان داده می‌شود. - فارسی، با ارقامِ فارسی. هر گزینه حداکثر ۴۰ نویسه. نوعِ مسئله: {problem_type} مسئله: {problem} راه‌حل: {solution} فقط این JSON را برگردان، بدونِ markdown و بدونِ توضیح: {{"prompt": "صورتِ سوال", "options": ["الف", "ب", "ج", "د"], "correctIndex": 0}}""" class Command(BaseCommand): help = 'ساختِ سوالِ چهارگزینه‌ای از chunkهای کتاب (هزینه دارد).' def add_arguments(self, parser): parser.add_argument('--subject', default='ریاضی') parser.add_argument('--grade', required=True) parser.add_argument('--chapters', nargs='*', type=int, default=None) parser.add_argument('--limit', type=int, default=None, help='سقفِ تعدادِ سوال در این اجرا (کنترلِ هزینه).') parser.add_argument('--per-lesson', type=int, default=1, help='چند سوال برای هر درس.') parser.add_argument('--dry-run', action='store_true', help='فقط نشان بده چند سوال ساخته می‌شود؛ هیچ فراخوانیِ پولی نکن.') parser.add_argument('--export', metavar='DIR', help='به‌جای فراخوانیِ مدل، جفت‌های (درس، مسئله) را در این ' 'پوشه به JSON بنویس تا subagent سوال‌ها را بسازد. رایگان.') parser.add_argument('--batch-size', type=int, default=30, help='چند جفت در هر فایلِ --export. پیش‌فرض ۳۰.') parser.add_argument('--model', default='o4-mini', help='مدلِ سازنده. پیش‌فرض o4-mini؛ gpt-4o-mini کیفیتِ ' 'قابلِ قبولی نداد (۴ سوال از ۶ خراب).') parser.add_argument('--all-chunks', dest='verified_only', action='store_false', help='از chunkهای بازبینی‌نشده هم بساز (پیش‌فرض: فقط تأییدشده‌ها).') parser.set_defaults(verified_only=True) parser.add_argument('--workers', type=int, default=4, help='فراخوانیِ هم‌زمان. بیش از ۴ روی این پروکسی ۴۲۹ می‌گیرد.') parser.add_argument('--strict-title', dest='by_page', action='store_false', help='فقط chunkهایی که عنوانشان دقیقاً با عنوانِ درس یکی است ' '(رفتارِ قدیم؛ بعد از بازخوانیِ کتاب‌ها تقریباً هیچ‌چیز پیدا نمی‌کند).') parser.set_defaults(by_page=True) def handle(self, *args, **opts): book = get_textbook(opts['subject'], opts['grade']) if book is None: raise CommandError( f'کتابی برای «{opts["subject"]} {opts["grade"]}» ثبت نشده. ' 'موجود: ' + '، '.join(sorted(TEXTBOOKS)) ) grade = Grade.objects.filter(title=f'پایه {book.grade}').first() if grade is None: raise CommandError( f'پایه‌ی «پایه {book.grade}» در دیتابیس نیست. ' f'اول: python manage.py seed_curriculum_from_textbook --grade {book.grade}' ) collection = build_collection_name(book.subject, book.grade) self._book = book targets = self._targets(grade, collection, opts) if not targets: self.stdout.write(self.style.WARNING('هیچ درسی بدونِ سوال پیدا نشد — کاری لازم نیست.')) return self.stdout.write(self.style.MIGRATE_HEADING( f'\n{len(targets)} سوال برای پایه {book.grade}\n' )) for lesson, chunk in targets[:10]: self.stdout.write(f' • {lesson.node.unit.title} / {lesson.title}') if len(targets) > 10: self.stdout.write(f' … و {len(targets) - 10} مورد دیگر') if opts['export']: self._export(targets, opts) return # ~۹۰۰ توکن ورودی و ~۱۵۰ خروجی برای هر سوال با gpt-4o-mini. self.model = opts['model'] cost = len(targets) * 0.0018 # اندازه‌گیریِ واقعیِ o4-mini self.stdout.write(self.style.WARNING( f'\nبرآوردِ هزینه با {self.model}: ≈ ${cost:.2f}' )) if opts['dry_run']: self.stdout.write(self.style.WARNING('--dry-run: چیزی ساخته نشد.')) return if not rag_is_configured(): raise CommandError('RAG پیکربندی نشده (کلیدها خالی‌اند یا RAG_FORCE_MOCK روشن است).') # سقفِ مصرفِ حسابِ سرویس با فشارِ لحظه‌ای فرق دارد: اولی ساعت‌ها باز # نمی‌شود و تلاشِ دوباره فقط پنجره را طولانی‌تر می‌کند. stop = threading.Event() def work(pair): lesson, chunk = pair if stop.is_set(): return lesson, None, None delay = 4.0 last = None for _ in range(4): try: return lesson, self._ask_model(chunk, book.grade), None except Exception as exc: # noqa: BLE001 last = exc text = str(exc).lower() if _is_account_limit(exc): stop.set() return lesson, None, exc if '429' not in text and 'rate' not in text: break time.sleep(delay) delay *= 2 return lesson, None, last created = skipped = failed = 0 with ThreadPoolExecutor(max_workers=opts['workers']) as pool: for lesson, data, err in pool.map(work, targets): if err is not None: failed += 1 self.stdout.write(self.style.ERROR(f' ✗ {lesson.title}: {str(err)[:85]}')) continue if data is None: skipped += 1 continue self._persist(lesson, data) created += 1 self.stdout.write(self.style.SUCCESS( f' ✓ {lesson.title}: {data["prompt"][:55]}')) if stop.is_set(): self.stdout.write(self.style.ERROR( '\nسقفِ مصرفِ حسابِ سرویس پر شد؛ اجرا نیمه‌کاره ماند. ' 'اجرای بعدی از همین‌جا ادامه می‌دهد.')) self.stdout.write(self.style.SUCCESS( f'\n{created} سوال ساخته شد، {skipped} رد شد، {failed} خطا.' )) # ------------------------------------------------------------------ def _export(self, targets, opts): """ جفت‌ها را در فایل‌های کوچکِ JSON بنویس؛ هیچ مدلی صدا زده نمی‌شود. چرا تکه‌تکه و نه یک فایلِ بزرگ: هر فایل سهمِ یک subagent است و باید در یک نشست جا شود. و چرا `prompt` را هم کنارش می‌نویسیم: تا کسی که سوال‌ها را می‌سازد قواعد را از حافظه بازنویسی نکند — همان متنی که به مدلِ پولی می‌رفت عیناً در دسترسِ subagent است. """ directory = opts['export'] os.makedirs(directory, exist_ok=True) book = self._book size = max(1, opts['batch_size']) slug = f'{book.subject}-{book.grade}' written = [] for index in range(0, len(targets), size): rows = [ { 'lessonId': lesson.pk, 'unit': lesson.node.unit.title, 'node': lesson.node.title, 'lessonTitle': lesson.title, 'problemType': chunk.problem_type, 'problem': chunk.problem, 'solution': chunk.solution, } for lesson, chunk in targets[index:index + size] ] path = os.path.join(directory, f'{slug}-{index // size + 1:02d}.json') with open(path, 'w', encoding='utf-8') as handle: json.dump(rows, handle, ensure_ascii=False, indent=1) written.append((path, len(rows))) rules = os.path.join(directory, f'{slug}-RULES.txt') with open(rules, 'w', encoding='utf-8') as handle: handle.write(PROMPT.format( grade=book.grade, problem_type='', problem='', solution='', )) self.stdout.write(self.style.SUCCESS( f'\n{len(targets)} جفت در {len(written)} فایل نوشته شد: {directory}' )) self.stdout.write(f' قواعدِ ساختِ سوال: {rules}') self.stdout.write( ' خروجیِ هر subagent باید قالبِ content.question_io باشد ' '({"lessonId", "prompt", "choices", "correct"}) و با ' '«ورودِ سوال» در پنلِ ادمین یا import_questions وارد شود.' ) # ------------------------------------------------------------------ def _page_span(self, lesson): """بازه‌ی صفحه‌ی PDF برای نودِ این درس، از فهرستِ خودِ کتاب.""" node = lesson.node book = self._book if book is None: return None for chapter in book.chapters: if chapter.title != node.unit.title: continue for section in chapter.resolved_sections(): if section.title == node.title: off = book.page_offset return section.first_page + off, section.last_page + off # نودی که بخشِ متناظر ندارد: کلِ فصل را بازه بگیر. return book.pdf_pages(chapter) return None def _chunks_by_page(self, lesson, collection, need, opts, used, skip=0): span = self._page_span(lesson) if span is None: return [] first, last = span qs = (LessonChunk.objects .filter(collection=collection, source_page__gte=first, source_page__lte=last) .exclude(problem='').exclude(pk__in=used)) if opts['verified_only']: qs = qs.filter(solution_verdict=LessonChunk.VERDICT_AGREES) return list(qs.order_by('source_page', 'order')[skip:skip + need]) def _targets(self, grade, collection, opts): """[(lesson, chunk), ...] — فقط درس‌هایی که هنوز سوال کم دارند.""" lessons = ( Lesson.objects .filter(node__unit__grade=grade, node__unit__subject__title=self._book.subject) .select_related('node__unit') .order_by('node__unit__order', 'node__order', 'id') ) if opts['chapters']: lessons = lessons.filter(node__unit__order__in=opts['chapters']) targets = [] used: set[int] = set() # یک chunk فقط به یک درس داده شود for lesson in lessons: have = lesson.questions.count() need = opts['per_lesson'] - have if need <= 0: continue # فقط chunkهایی که راه‌حلشان بازبینی و تأیید شده. ساختنِ سوال از # راه‌حلی که می‌دانیم غلط است، خطای کتاب را به دانش‌آموز منتقل # می‌کند — و ۱۲۳ موردِ این‌چنینی در همین کتاب‌ها پیدا شد. candidates = LessonChunk.objects.filter( collection=collection, problem_type=lesson.title).exclude(problem='') if opts['verified_only']: candidates = candidates.filter( solution_verdict=LessonChunk.VERDICT_AGREES) # از `have` تا به بعد، نه از اول: ترتیب قطعی است، پس اجرای قبلی # دقیقاً `have` تکه‌ی نخست را مصرف کرده. بدونِ این پرش، هر اجرای # دوم با --per-lesson بالاتر همان مسئله را دوباره تحویل می‌داد و # سوالِ تقریباً تکراری می‌ساخت — که نه خطا می‌دهد و نه در import # گیر می‌افتد، چون صورتِ سوال بازنویسی شده و رشته‌اش یکی نیست. chunks = list(candidates.order_by('source_page', 'order')[have:have + need]) # تطبیقِ عنوان شکننده است: عنوانِ درس از یک استخراجِ قدیمی آمده و # هر بار که کتاب دوباره خوانده شود مدل عنوانِ تازه می‌سازد. بعد از # بازخوانیِ ۱۴۰۵/۰۶/۲۹ فقط ۵ تا ۸ درصدِ عنوان‌ها می‌خوردند، یعنی # ۹۵٪ درس‌ها بی‌سروصدا هیچ نامزدی پیدا نمی‌کردند. # # لنگرِ پایدار، صفحه است: هر Node دقیقاً یک بخشِ فهرستِ کتاب با # بازه‌ی صفحه‌ی مشخص است، و هر chunk هم `source_page` دارد. پس # chunkهای همان بازه به درس‌های همان نود می‌خورند — بدونِ هیچ # فراخوانیِ مدلی. if not chunks and opts['by_page']: chunks = self._chunks_by_page(lesson, collection, need, opts, used, skip=have) for chunk in chunks: used.add(chunk.pk) targets.append((lesson, chunk)) if opts['limit'] and len(targets) >= opts['limit']: return targets return targets def _ask_model(self, chunk, grade): # noqa: D401 prompt = PROMPT.format( grade=grade, problem_type=chunk.problem_type, problem=chunk.problem, solution=chunk.solution, ) client = clients.get_openai_client() completion = client.chat.completions.create( model=self.model, messages=[ {'role': 'system', 'content': 'You always answer with a single valid JSON object.'}, {'role': 'user', 'content': prompt}, ], ) raw = (completion.choices[0].message.content or '').strip() for fence in ('```json', '```'): if raw.startswith(fence): raw = raw[len(fence):].strip() if raw.endswith('```'): raw = raw[:-3].strip() data = json.loads(raw) if data.get('skip'): return None # ⚠️ گزینه‌های خالی را حذف نمی‌کنیم و بعد اعتبارسنجی می‌کنیم: اگر مدل # پنج گزینه بدهد که یکی‌شان خالی است، حذفِ آن یکی چهارتا باقی می‌گذارد # ولی correctIndex همچنان به فهرستِ اصلی اشاره می‌کند — یعنی سوال # می‌سازیم و گزینه‌ی *غلط* را درست علامت می‌زنیم. پس اول ساختار را # همان‌طور که آمده می‌سنجیم. raw_options = data.get('options') index = data.get('correctIndex') if not isinstance(raw_options, list) or len(raw_options) != 4: raise ValueError(f'دقیقاً چهار گزینه لازم است: {raw_options!r}') if not isinstance(index, int) or isinstance(index, bool) or not 0 <= index < 4: raise ValueError(f'correctIndex نامعتبر: {index!r}') options = [str(o).strip()[:100] for o in raw_options] if not all(options) or len(set(options)) != 4: raise ValueError(f'گزینه‌ی خالی یا تکراری: {options}') if not (data.get('prompt') or '').strip(): raise ValueError('صورتِ سوال خالی بود') return {'prompt': data['prompt'].strip(), 'options': options, 'correctIndex': index} @transaction.atomic def _persist(self, lesson, data): question = Question.objects.create( lesson=lesson, type='multiple_choice', prompt=data['prompt'], ) Choice.objects.bulk_create([ Choice(question=question, value=value, is_correct=(i == data['correctIndex'])) for i, value in enumerate(data['options']) ])