""" راستی‌آزماییِ راه‌حل‌هایی که vision از کتابِ درسی درآورده. چرا لازم است: استخراج گاهی راه‌حلِ غلط می‌سازد و این خطا در بازیابی دیده نمی‌شود — فقط وقتی کسی به `solution` اعتماد کند (ساختِ سؤال، یا خودِ دستیار هنگامِ نقلِ جواب) خودش را نشان می‌دهد. نمونه‌های واقعی از همین کتاب‌ها: BMI = ۱۰۰/(۲٫۷۴)² کتاب: ۳۳٫۲ درست: ۱۳٫۳ ۱۶ آبی + ۲۴ صورتی کتاب: «کل ۵۰» درست: ۴۰ بزرگ‌ترین عددِ ۴رقمیِ بی‌تکرار کتاب: ۸۵۳۲ درست: ۹۸۷۶ g(x) = x³ − ۲ کتاب: «نمایی» درست: چندجمله‌ایِ درجه ۳ `solution` هرگز بازنویسی نمی‌شود. کتاب باید کتاب بماند: اگر جوابِ مدل را جایش می‌گذاشتیم و مدل اشتباه می‌کرد، منبعِ حقیقت بی‌صدا خراب می‌شد. ────────────────────────────────────────────────────────────────────────── درس‌هایی که از اجرای اولِ این دستور گرفته شد و در طراحیِ فعلی نشسته‌اند: ۱. نتیجه باید سه‌حالته باشد. با بله/خیر، «نمی‌دانم» ناچار به «غلط است» ترجمه می‌شد: ۳۵٪ از گزارش‌ها مسئله‌هایی بودند که به شکل یا جدولِ کتاب ارجاع می‌دادند و اصلاً قابلِ قضاوت نبودند. ۲. «۳/۱۰۴» در این کتاب‌ها عددِ اعشاریِ ۳٫۱۰۴ است، نه کسرِ ۳÷۱۰۴. بدونِ گفتنِ این به مدل، ۲۴٪ از گزارش‌ها هشدارِ کاذب بود — بیشترشان در پایه ششم که پر از اعشار است. ۳. مسئله‌های وابسته به شکل اصلاً فرستاده نمی‌شوند. هزینه دارند و جوابشان بی‌معناست؛ همان‌جا برچسبِ unjudgeable می‌خورند. ۴. دو مدل، دو نقش. deepseek-reasoner ارزان است و تشخیصش خوب، ولی جوابِ خودش گاهی غلط است (مسئله‌ی سرعتِ نور را ۱۰^۱۷ داد به‌جای ۱۰^۱۴). پس او تور می‌اندازد و o4-mini فقط روی صیدها داوری می‌کند. ────────────────────────────────────────────────────────────────────────── python manage.py rag_verify_solutions --all python manage.py rag_verify_solutions --all --adjudicate python manage.py rag_verify_solutions --collection "ریاضی ششم" --limit 20 python manage.py rag_verify_solutions --report """ import json import re import threading import time from concurrent.futures import ThreadPoolExecutor from django.core.management.base import BaseCommand, CommandError from django.db.models import Count, Q from django.utils import timezone from rag.models import LessonChunk from rag.services import clients from rag.services.config import rag_is_configured SCOUT_MODEL = 'deepseek-reasoner' JUDGE_MODEL = 'o4-mini' # دو خطای کاملاً متفاوت که هر دو ۴۲۹‌اند و نباید یکسان با آن‌ها رفتار کرد: # # • فشارِ لحظه‌ایِ موازی‌کاری — گذراست، عقب‌نشینی حلش می‌کند. # • سقفِ مصرفِ حساب — ساعت‌ها باز نمی‌شود. تلاشِ دوباره نه‌تنها بی‌فایده # است، پنجره‌ی محدودیت را هم تمدید می‌کند: یک اجرا ۱۱۹ بار دوباره تلاش # کرد، ۶۵ دقیقه طول کشید، هیچ کاری نکرد، و مهلت را از ۱۴ ساعت جلوتر برد. class AccountLimitReached(Exception): """حسابِ سرویس جواب نمی‌دهد؛ ادامه دادن بی‌معناست.""" 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 # مسئله‌ای که به شکل/جدولِ کتاب ارجاع می‌دهد بدونِ آن شکل قابلِ حل نیست. # فرستادنش هم پول می‌سوزاند هم آمار را خراب می‌کند. NEEDS_FIGURE = re.compile( r'شکل|جدول|نمودار|روبه‌رو|روبرو|مقابل|تصویر|نقشه|رسم کنید|بکشید|' r'رنگ[‌ ]?آمیزی|زیر را|بالا را|وِن' ) PROMPT = """تو یک ریاضی‌دانِ دقیق هستی که راه‌حل‌های کتابِ درسیِ ایران را بازبینی می‌کند. ⚠️ سه قاعده‌ی مهم درباره‌ی نگارشِ این متن‌ها: ۱. علامتِ «/» بینِ دو عدد، در این کتاب‌ها تقریباً همیشه **جداکننده‌ی اعشار** است، نه خطِ کسر. یعنی «۳/۱۰۴» عددِ ۳٫۱۰۴ است و «۲/۵» عددِ ۲٫۵. کسرها معمولاً به شکلِ «۳ بر ۴» یا با کسرِ عمودی می‌آیند. ۲. صفر در ریاضیِ مدرسه‌ی ایران عددِ طبیعی **نیست**. ۳. متن با OCR از PDF درآمده؛ فاصله و نیم‌فاصله ممکن است به‌هم‌ریخته باشد. تفاوتِ صرفاً نگارشی، گِردکردن، یا ترتیبِ نوشتن **خطا نیست**. مسئله: {problem} راه‌حلِ کتاب: {solution} کارَت: ۱. اول خودت مسئله را قدم‌به‌قدم حل کن. ۲. بعد راه‌حلِ کتاب را با جوابِ خودت بسنج. یکی از این سه را برگردان: • "agrees" — راه‌حلِ کتاب از نظرِ ریاضی درست است (حتی اگر نگارشش فرق دارد) • "disagrees" — خطای قطعیِ ریاضی دارد. فقط وقتی مطمئنی. • "unjudgeable" — بدونِ شکل/جدولِ کتاب یا داده‌ی جاافتاده نمی‌شود قضاوت کرد، یا مسئله جوابِ عددیِ مشخصی ندارد (پرسشِ باز یا توضیحی). اگر شک داری بینِ disagrees و unjudgeable، unjudgeable را انتخاب کن. فقط این JSON را برگردان، بدونِ هیچ متنِ اضافه: {{"verdict": "agrees|disagrees|unjudgeable", "answer": "جوابِ خودت، کوتاه", "why": "حداکثر ۱۲۰ نویسه"}}""" def first_json(text: str): """ اولین شیءِ JSON را از متن بیرون می‌کشد. مدل‌های استدلالی گاهی قبل یا بعد از JSON توضیح می‌چسبانند و json.loads خام روی همان می‌افتد. در محکِ اولیه همین باعث شد deepseek به‌اشتباه ۷ از ۸ به‌نظر برسد در حالی که ۸ از ۸ بود. """ depth = 0 start = None for i, ch in enumerate(text or ''): if ch == '{': if depth == 0: start = i depth += 1 elif ch == '}': depth -= 1 if depth == 0 and start is not None: try: return json.loads(text[start:i + 1]) except json.JSONDecodeError: start = None return None class Command(BaseCommand): help = 'بازبینیِ درستیِ راه‌حل‌های استخراج‌شده از کتاب (هزینه دارد، کم).' def add_arguments(self, parser): parser.add_argument('--collection', default=None) parser.add_argument('--all', action='store_true') parser.add_argument('--limit', type=int, default=None) parser.add_argument('--workers', type=int, default=12) parser.add_argument('--adjudicate', action='store_true', help='پاسِ دوم: مشکوک‌ها را با مدلِ دقیق‌تر بسنج.') parser.add_argument('--recheck', action='store_true') parser.add_argument('--report', action='store_true', help='فقط گزارشِ وضعیت؛ هیچ فراخوانی‌ای نمی‌کند.') parser.add_argument('--sync-qdrant', action='store_true', help='برچسب‌ها را در payloadِ Qdrant بنویس (رایگان).') def handle(self, *args, **opts): if opts['report']: return self._report() if opts['sync_qdrant']: return self._sync_qdrant(opts) if not rag_is_configured(): raise CommandError('RAG پیکربندی نشده (کلیدها خالی‌اند).') if opts['adjudicate']: return self._adjudicate(opts) qs = LessonChunk.objects.exclude(problem='').exclude(solution='') if opts['collection']: qs = qs.filter(collection=opts['collection']) elif not opts['all']: raise CommandError('یا --collection بدهید یا --all.') if not opts['recheck']: qs = qs.filter(solution_checked_at__isnull=True) rows = list(qs.order_by('collection', 'source_page', 'order')) skipped = self._mark_figure_dependent(rows) rows = [r for r in rows if not NEEDS_FIGURE.search(r.problem or '')] if opts['limit']: rows = rows[:opts['limit']] if skipped: self.stdout.write( f' {skipped} مسئله‌ی وابسته به شکل/جدول بدونِ فراخوانی ' 'برچسبِ unjudgeable گرفتند.' ) if not rows: self.stdout.write(self.style.WARNING('چیزی برای بررسی نماند.')) return self.stdout.write(self.style.MIGRATE_HEADING( f'\n{len(rows)} chunk با {SCOUT_MODEL} (~${len(rows) * 0.00014:.2f})\n')) self._run(rows, SCOUT_MODEL, opts['workers']) def _mark_figure_dependent(self, rows) -> int: """قبل از خرجِ یک ریال، آن‌هایی که ذاتاً قابلِ قضاوت نیستند کنار می‌روند.""" marked = [] for chunk in rows: if NEEDS_FIGURE.search(chunk.problem or ''): chunk.solution_verdict = LessonChunk.VERDICT_UNJUDGEABLE chunk.check_note = 'به شکل/جدولِ کتاب ارجاع می‌دهد؛ از متن قابلِ قضاوت نیست.' chunk.checked_by = 'pre-filter' chunk.solution_checked_at = timezone.now() marked.append(chunk) if marked: LessonChunk.objects.bulk_update( marked, ['solution_verdict', 'check_note', 'checked_by', 'solution_checked_at']) return len(marked) def _adjudicate(self, opts): qs = LessonChunk.objects.filter( solution_verdict=LessonChunk.VERDICT_DISAGREES).exclude(problem='') if opts['collection']: qs = qs.filter(collection=opts['collection']) if not opts['recheck']: qs = qs.exclude(checked_by=JUDGE_MODEL) rows = list(qs.order_by('collection', 'source_page')[:opts['limit'] or 10_000]) if not rows: self.stdout.write(self.style.WARNING('مشکوکی برای داوری نمانده.')) return # مدلِ داور روی این پروکسی سخت‌گیرتر نرخ‌بندی می‌شود؛ نصفِ کارگرها. workers = max(1, min(opts['workers'], 4)) self.stdout.write(self.style.MIGRATE_HEADING( f'\nداوریِ {len(rows)} مورد با {JUDGE_MODEL} ({workers} کارگر) ' f'(~${len(rows) * 0.0018:.2f})\n')) self._run(rows, JUDGE_MODEL, workers) def _run(self, rows, model, workers): """ فراخوانی‌ها موازی، نوشتن دسته‌ای از نخِ اصلی. هر chunk چند ثانیه منتظرِ شبکه است، پس ترتیبی یعنی ساعت‌ها؛ ولی نوشتنِ هم‌زمان از چند نخ روی SQLite به قفلِ نوشتن می‌خورد. """ client = clients.get_openai_client() stop = threading.Event() def ask(chunk): # اگر یک نخ به سقفِ حساب خورده، بقیه بی‌درنگ برمی‌گردند؛ ادامه دادن # فقط پنجره‌ی محدودیت را طولانی‌تر می‌کند. if stop.is_set(): return chunk, None, None delay = 4.0 last = None for _ in range(5): try: completion = client.chat.completions.create( model=model, messages=[{'role': 'user', 'content': PROMPT.format( problem=chunk.problem[:900], solution=chunk.solution[:900])}], ) data = first_json(completion.choices[0].message.content) verdict = (data or {}).get('verdict') if verdict not in dict(LessonChunk.VERDICT_CHOICES): raise ValueError(f'verdictِ نامعتبر: {verdict!r}') return chunk, data, None except Exception as exc: # noqa: BLE001 last = exc text = str(exc).lower() if is_account_limit(exc): stop.set() return chunk, None, AccountLimitReached(str(exc)[:200]) if '429' not in text and 'rate' not in text: break if stop.is_set(): break time.sleep(delay) delay *= 2 return chunk, None, last counts = {v: 0 for v, _ in LessonChunk.VERDICT_CHOICES} failed = 0 aborted: AccountLimitReached | None = None batch: list[LessonChunk] = [] t0 = time.time() done = 0 with ThreadPoolExecutor(max_workers=workers) as pool: for chunk, data, err in pool.map(ask, rows): done += 1 if isinstance(err, AccountLimitReached): aborted = err continue if err is None and data is None: continue # بعد از توقف، نوبتش نرسید if err is not None: failed += 1 self.stdout.write(self.style.ERROR(f' ✗ #{chunk.pk}: {str(err)[:75]}')) else: chunk.solution_verdict = data['verdict'] chunk.checked_answer = str(data.get('answer') or '')[:2000] chunk.check_note = str(data.get('why') or '')[:300] chunk.checked_by = model chunk.solution_checked_at = timezone.now() batch.append(chunk) counts[data['verdict']] += 1 if data['verdict'] == LessonChunk.VERDICT_DISAGREES: self.stdout.write( f' ⚠ {chunk.collection} ص{chunk.source_page}: ' f'{chunk.problem[:50]} → {chunk.checked_answer[:40]}') if len(batch) >= 40: self._flush(batch) if done % 100 == 0: rate = done / max(time.time() - t0, 1) self.stdout.write( f' … {done}/{len(rows)} (~{(len(rows)-done)/max(rate,.001)/60:.0f} دقیقه مانده)') self._flush(batch) self.stdout.write(self.style.SUCCESS( f"\n{counts['agrees']} درست، {counts['disagrees']} مشکوک، " f"{counts['unjudgeable']} غیرقابلِ قضاوت، {failed} خطا — {time.time()-t0:.0f}s")) if aborted is not None: # آنچه تا اینجا سنجیده شده ذخیره شده و اجرای بعدی از همین‌جا ادامه # می‌دهد، چون فقط بررسی‌نشده‌ها برداشته می‌شوند. raise CommandError( 'سقفِ مصرفِ حسابِ سرویس پر شده و اجرا همین‌جا متوقف شد.\n' f' پیامِ سرویس: {aborted}\n' ' تا باز شدنِ سقف دوباره تلاش نکنید؛ هر تلاش مهلت را جلوتر می‌برد.\n' ' وضعیت با: python manage.py rag_verify_solutions --report' ) @staticmethod def _flush(batch): if batch: LessonChunk.objects.bulk_update( batch, ['solution_verdict', 'checked_answer', 'check_note', 'checked_by', 'solution_checked_at']) batch.clear() def _sync_qdrant(self, opts): """ برچسب‌ها را روی نقاطِ موجود می‌نویسد، بدونِ بردار و بدونِ هزینه. نقطه‌ها بر اساسِ مقدارِ برچسب دسته می‌شوند، پس برای هر collection سه فراخوانی کافی است نه هزاران‌تا. """ from rag.services.qdrant_store import set_payload_fields qs = LessonChunk.objects.exclude(solution_verdict='') if opts['collection']: qs = qs.filter(collection=opts['collection']) buckets: dict[tuple[str, str], list[str]] = {} for collection, verdict, point_id in qs.values_list( 'collection', 'solution_verdict', 'qdrant_point_id'): buckets.setdefault((collection, verdict), []).append(str(point_id)) if not buckets: self.stdout.write(self.style.WARNING('برچسبی برای همگام‌سازی نیست.')) return total = 0 for (collection, verdict), ids in sorted(buckets.items()): set_payload_fields(collection, ids, {'solution_verdict': verdict}) total += len(ids) self.stdout.write(f' {collection:14s} {verdict:12s} {len(ids):4d} نقطه') self.stdout.write(self.style.SUCCESS(f'\n{total} نقطه در Qdrant به‌روز شد.')) def _report(self): rows = (LessonChunk.objects.exclude(problem='').exclude(solution='') .values('collection') .annotate(n=Count('id'), ok=Count('id', filter=Q(solution_verdict='agrees')), bad=Count('id', filter=Q(solution_verdict='disagrees')), unk=Count('id', filter=Q(solution_verdict='unjudgeable')), todo=Count('id', filter=Q(solution_checked_at__isnull=True))) .order_by('collection')) self.stdout.write(f"\n{'کتاب':14s} {'کل':>5s} {'درست':>6s} {'مشکوک':>7s} " f"{'بی‌قضاوت':>9s} {'مانده':>6s}") for r in rows: self.stdout.write(f"{r['collection']:14s} {r['n']:5d} {r['ok']:6d} " f"{r['bad']:7d} {r['unk']:9d} {r['todo']:6d}")