ingest whole textbooks chapter by chapter, resumably

نقشه‌ی فیزیکیِ کتاب‌ها (افستِ صفحه، بازه‌ی هر فصل، درس‌های داخلِ فصل از
فهرستِ خودِ کتاب) در rag/services/textbooks.py ثبت شد تا عنوانِ فصل از
کتاب بیاید نه از حدسِ مدل — رشته‌ی فصل باید دقیقاً بینِ ingest و retrieval
یکی باشد وگرنه نتیجه بی‌سروصدا خالی برمی‌گردد.

rag_ingest_book کتاب را فصل‌به‌فصل ingest می‌کند و قابلِ ازسرگیری است:
rag.IngestedPage نشانه‌ی «این صفحه پردازش شد» را نگه می‌دارد حتی وقتی
صفحه chunkی نساخته، وگرنه هر بار اجرای دوباره بابتِ صفحه‌های بی‌مسئله
دوباره پول می‌گرفت. --plan هزینه را رایگان برآورد می‌کند.

rag_import_collection نقاطِ موجودِ Qdrant را در LessonChunk می‌ریزد —
فقط می‌خواند، رایگان و idempotent، برای پایه‌هایی که تیمِ دیگر ingest
کرده بود.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-13 18:59:03 +03:30
parent 52d9694583
commit c79be0a336
11 changed files with 1139 additions and 35 deletions

View File

@@ -8,19 +8,36 @@ DB (LessonChunk، منبعِ حقیقت) + upsert به Qdrant (index).
import logging
import os
import tempfile
from pathlib import Path
from django.conf import settings
from django.db import transaction
from ..models import LessonChunk
from ..models import IngestedPage, LessonChunk
from .chunking import chunk_and_summarize
from .config import build_collection_name, resolve_grade_title
from .embedding import embed_text
from .known_collections import is_math_subject, resolve_math_chapter
from . import pdf as pdf_service
from .qdrant_store import ensure_collection, upsert_points
from .textbooks import Textbook, get_textbook
logger = logging.getLogger(__name__)
# قیمتِ هر یک میلیون توکن (دلار) برای گزارشِ هزینه‌ی ingest. اندازه‌گیریِ
# ۱۱۱ صفحه‌ی واقعیِ ریاضی نهم: ۱٬۴۲۵ توکن ورودی و ~۴۰۰ توکن خروجی در هر صفحه،
# یعنی حدود ۰٫۰۰۷۵ دلار — نه ۰٫۰۱۷ که اول تخمین زده بودیم.
MODEL_PRICE_PER_MTOK = {
'gpt-4o': (2.50, 10.00),
'gpt-4o-mini': (0.15, 0.60),
}
def estimate_page_cost(model: str) -> float:
"""هزینه‌ی تقریبیِ یک صفحه بر پایه‌ی مصرفِ اندازه‌گیری‌شده‌ی واقعی."""
price_in, price_out = MODEL_PRICE_PER_MTOK.get(model, (2.50, 10.00))
return 1425 / 1e6 * price_in + 400 / 1e6 * price_out
def _payload(chunk: LessonChunk) -> dict:
"""
@@ -29,17 +46,44 @@ def _payload(chunk: LessonChunk) -> dict:
⚠️ کلیدِ `chapter` (رشته‌ی کاملِ فصل) حتماً باید نوشته بشه: جستجوی ریاضی در
retrieval.py دقیقاً روی همین کلید فیلتر می‌کنه (نه `chapter_number`)، پس
بدونش هر چیزی که ingest کنیم برای همیشه نامرئی می‌مونه — بدونِ هیچ خطایی.
بقیه‌ی کلیدهای ساختاری هم عمداً هم‌نامِ قراردادِ تیمِ دیگه‌ان تا collectionِ
ما و اونا یک شکل بمونن.
دو شکلِ payload داریم و باید هر کدام دقیقاً مثلِ نقاطِ موجودِ همان نوع
بمانند (نقاطِ «ریاضی هفتم» و «علوم تجربی سوم» را مقایسه کنید):
- ریاضی: grade / chapter / type / keywords / generalSolution / problem /
solution — و عمداً **بدونِ `summary`**. اگر summary بنویسیم،
`retrieval._payload_text` آن را به generalSolution ترجیح می‌دهد و
به‌جای راه‌حل، متنِ صافِ «Type: ... Keywords: ...» را به‌عنوانِ context
به مدل می‌دهیم — یعنی نقاطِ ما بی‌سروصدا بدتر از نقاطِ تیمِ دیگر جواب
می‌دهند. `lesson_title` هم به همین دلیل نمی‌آید: `_payload_title`
ترجیحش می‌دهد و عنوانِ عمومیِ فصل جایِ نامِ نوعِ مسئله را می‌گیرد.
- درس‌های متنی: lesson_number + summary، همان‌طور که بود.
کلیدهای دفتری (chapter_number / chunk_order / source_page) در هر دو حالت
می‌آیند؛ روی فیلتر و متنِ بازیابی‌شده اثری ندارند و ردیابیِ «این نقطه از
کدام صفحه‌ی کتاب آمده» را ممکن می‌کنند.
"""
payload = {
'lesson_id': chunk.lesson_id,
# منشأ را از `problem_type` تشخیص می‌دهیم، نه از پر بودنِ راه‌حل: صفحه‌ای
# که vision فقط `type` و `keywords` از آن درآورده هم ردیفِ ساختاریافته است،
# ولی با شرطِ قبلی «متنی» حساب می‌شد و `summary` می‌گرفت — یعنی همان آلودگیِ
# «Type: … Keywords: …» که این تفکیک برای جلوگیری از آن ساخته شده بود.
is_structured = bool(
chunk.problem_type or chunk.general_solution or chunk.problem or chunk.solution
)
payload: dict = {
'lesson_number': chunk.lesson_number,
'chapter_number': chunk.chapter,
'chunk_order': chunk.order,
'lesson_title': chunk.title,
'summary': chunk.summary,
}
if chunk.lesson_id is not None:
payload['lesson_id'] = chunk.lesson_id
if chunk.source_page is not None:
payload['source_page'] = chunk.source_page
if not is_structured:
payload['lesson_title'] = chunk.title
payload['summary'] = chunk.summary
for key, value in (
('chapter', chunk.chapter_title),
('grade', chunk.grade_title),
@@ -57,13 +101,16 @@ def _payload(chunk: LessonChunk) -> dict:
def _persist_and_upsert(rows: list[dict], *, lesson_id, grade_id, subject,
lesson_number, chapter, title, collection) -> list[LessonChunk]:
ensure_collection(collection)
created: list[LessonChunk] = []
points: list[dict] = []
# embedding قبل از باز کردنِ transaction: هر فراخوانی چند صد میلی‌ثانیه
# شبکه است و نگه داشتنِ قفلِ نوشتنِ SQLite در آن مدت، بقیه‌ی درخواست‌ها را
# پشتِ «database is locked» می‌خواباند.
vectors = [embed_text(row['summary']) for row in rows]
created: list[LessonChunk] = []
with transaction.atomic():
for row in rows:
vector = embed_text(row['summary'])
chunk = LessonChunk.objects.create(
for row, vector in zip(rows, vectors):
created.append(LessonChunk.objects.create(
lesson_id=lesson_id,
grade_id=grade_id,
subject=subject,
@@ -75,14 +122,30 @@ def _persist_and_upsert(rows: list[dict], *, lesson_id, grade_id, subject,
summary=row['summary'],
embedding_vector=vector,
collection=collection,
source_page=row.get('source_page'),
# فیلدهای ساختاری فقط از مسیرِ PDF/vision می‌آن؛ ingest_text
# اون‌ها رو نداره و خالی می‌مونن (payload هم اون‌وقت نمی‌نویسدشون).
**row.get('structured', {}),
)
created.append(chunk)
points.append({'id': chunk.qdrant_point_id, 'vector': vector, 'payload': _payload(chunk)})
))
upsert_points(collection, points)
# ⚠️ upsert بعد از commit است تا نقطه‌ای در Qdrant بدونِ ردیفِ متناظر در DB
# نماند. ولی اگر همین upsert شکست بخورد، ردیف‌ها commit شده‌اند و
# `_done_pages` آن صفحه را «انجام‌شده» می‌بیند — یعنی chunkها برای همیشه در
# DB می‌مانند و هیچ‌وقت به Qdrant نمی‌رسند، بی‌آنکه کسی بفهمد. پس در صورتِ
# خطا ردیف‌های همین دسته را پس می‌گیریم تا صفحه واقعاً ناتمام بماند و اجرای
# بعدی دوباره سراغش برود.
try:
upsert_points(collection, [
{'id': c.qdrant_point_id, 'vector': v, 'payload': _payload(c)}
for c, v in zip(created, vectors)
])
except Exception:
LessonChunk.objects.filter(pk__in=[c.pk for c in created]).delete()
logger.exception(
'rag: upsert به collection %r شکست خورد؛ %d ردیفِ همین دسته پس گرفته شد',
collection, len(created),
)
raise
return created
@@ -189,3 +252,208 @@ def ingest_pdf(*, lesson_id: int, grade_id, subject: str, lesson_number: int,
'chunks_created': len(created),
'chunk_ids': [c.id for c in created],
}
# ─── ingestِ کتابِ کامل ────────────────────────────────────────────────────
#
# `ingest_pdf` بالا یک PDF را یک‌تکه و زیرِ یک نامِ فصل می‌گیرد؛ برای کتابِ
# درسیِ کامل (چند فصل، ۱۵۰+ صفحه) سه چیز کم دارد و همین‌ها اینجا اضافه شده‌اند:
# ۱. فصل‌بندی — بازه‌ی صفحه‌ی هر فصل از rag/services/textbooks.py می‌آید تا
# هر نقطه زیرِ فصلِ درستِ خودش بنشیند (فیلترِ retrieval روی همین است).
# ۲. ازسرگیری — صفحه‌ای که قبلاً chunk ساخته دوباره پردازش نمی‌شود، پس قطعیِ
# شبکه وسطِ کار پول و زمانِ رفته را دور نمی‌ریزد و نقطه‌ی تکراری نمی‌سازد.
# ۳. صفحه‌به‌صفحه بودن — هم مصرفِ حافظه ثابت می‌ماند، هم هر صفحه بلافاصله
# ذخیره می‌شود، هم خطای یک صفحه بقیه را از کار نمی‌اندازد.
def _textbook_path(book: Textbook) -> Path:
root = Path(getattr(settings, 'TEXTBOOK_DIR', Path(settings.BASE_DIR) / 'data' / 'textbooks'))
return root / book.filename
def _done_pages(collection: str, chapter_title: str) -> set[int]:
"""
صفحه‌هایی که نباید دوباره پردازش شوند.
هم نشانه‌های IngestedPage (شاملِ صفحه‌های بی‌مسئله) و هم chunkهای موجود
را می‌گیرد؛ دومی برای دادهٔ ingest‌شده پیش از افزودنِ جدولِ نشانه لازم است.
"""
marked = set(
IngestedPage.objects
.filter(collection=collection, chapter_title=chapter_title)
.values_list('page', flat=True)
)
chunked = set(
LessonChunk.objects
.filter(collection=collection, chapter_title=chapter_title, source_page__isnull=False)
.values_list('source_page', flat=True)
)
return marked | chunked
def ingest_book(*, subject: str, grade: str, chapters=None, dpi: int = 200,
max_pages: int | None = None, resume: bool = True,
progress=None) -> dict:
"""
یک کتابِ درسیِ کامل از data/textbooks/ را فصل‌به‌فصل ingest می‌کند.
chapters : فهرستِ شماره‌ی فصل‌ها؛ None یعنی همه‌ی فصل‌های کتاب.
max_pages : سقفِ تعدادِ صفحه‌ی پردازش‌شده در این اجرا (برای کنترلِ هزینه).
resume : صفحه‌هایی که قبلاً chunk ساخته‌اند دوباره پردازش نشوند.
progress : callback(dict) برای گزارشِ زنده به فرمانِ مدیریتی.
خروجی: خلاصه‌ی اجرا؛ هیچ استثنایی برای خطای یک صفحه بالا نمی‌آید — صفحه‌ی
خطادار در `failed_pages` گزارش می‌شود و اجرای بعدی دوباره سراغش می‌رود.
"""
book = get_textbook(subject, grade)
if book is None:
raise ValueError(f'کتابی برای «{subject} {grade}» در textbooks.py ثبت نشده است.')
# کتابِ بدونِ filename (ششم/هفتم — دادهٔ واقعی‌شان روی Qdrantِ تیمِ دیگر
# است) از این مسیر نمی‌آید. بدونِ این بررسی، مسیر به خودِ پوشه‌ی
# data/textbooks/ می‌رسید که وجود دارد، پس `exists()` رد می‌شد و کار
# بی‌سروصدا جلو می‌رفت تا جایی که pdf2image روی یک پوشه شکست بخورد.
if not book.filename:
raise FileNotFoundError(
f'«{book.key}» فایلِ PDF ندارد و از مسیرِ ingest نمی‌آید. '
f'برای واردکردنِ دادهٔ موجودش: '
f'python manage.py rag_import_collection --collection "{book.key}"'
)
path = _textbook_path(book)
if not path.exists() or not path.is_file():
raise FileNotFoundError(f'فایلِ کتاب پیدا نشد: {path}')
collection = build_collection_name(book.subject, book.grade)
ensure_collection(collection)
wanted = list(book.chapters) if chapters is None else [
c for n in chapters if (c := book.chapter(n)) is not None
]
def emit(**event):
if progress is not None:
progress(event)
totals = {'pages_done': 0, 'pages_skipped': 0, 'chunks_created': 0,
'prompt_tokens': 0, 'completion_tokens': 0}
failed_pages: list[int] = []
budget = max_pages
for chapter in wanted:
first_pdf, last_pdf = book.pdf_pages(chapter)
already = _done_pages(collection, chapter.title) if resume else set()
emit(kind='chapter', chapter=chapter.number, title=chapter.title,
pages=(first_pdf, last_pdf), already=len(already))
# order در هر فصل ادامه‌دار می‌ماند تا اجرای دوم شماره‌ها را از نو نزند.
order = LessonChunk.objects.filter(
collection=collection, chapter_title=chapter.title,
).count()
for page in range(first_pdf, last_pdf + 1):
if page in already:
totals['pages_skipped'] += 1
continue
if budget is not None and budget <= 0:
emit(kind='budget_reached')
return _book_result(book, collection, totals, failed_pages, partial=True)
try:
images = pdf_service.extract_page_images(
str(path), first_page=page, last_page=page, dpi=dpi,
)
if not images:
raise RuntimeError('صفحه به تصویر تبدیل نشد')
parsed = pdf_service.extract_problem_types(
images[0], book.grade_label, chapter.title,
)
except Exception as exc: # noqa: BLE001
logger.exception('rag.ingest_book: صفحه %s شکست خورد', page)
failed_pages.append(page)
emit(kind='page_failed', page=page, error=str(exc))
if budget is not None:
budget -= 1
continue
page_usage = parsed.pop('_usage', None) or {}
totals['prompt_tokens'] += int(page_usage.get('promptTokens') or 0)
totals['completion_tokens'] += int(page_usage.get('completionTokens') or 0)
rows = []
for problem_type in parsed.get('problemTypes', []) or []:
text = pdf_service.problem_type_to_text(problem_type)
if not text:
continue
order += 1
example = problem_type.get('exampleFromImage', {}) or {}
rows.append({
'order': order,
'source_page': page,
'original_text': text,
'summary': text,
'structured': {
# عنوانِ فصل از فهرستِ خودمان می‌آید، نه از چیزی که
# مدل در تصویر دیده: فیلترِ retrieval روی رشته‌ی دقیق
# است و یک حرفِ جابه‌جا کلِ فصل را نامرئی می‌کند.
'chapter_title': chapter.title,
'grade_title': book.grade_label,
'problem_type': (problem_type.get('type') or '').strip(),
'keywords': problem_type.get('keywords') or [],
'general_solution': (problem_type.get('generalSolution') or '').strip(),
'problem': (example.get('problem') or '').strip(),
'solution': (example.get('solution') or '').strip(),
},
})
if rows:
try:
created = _persist_and_upsert(
rows, lesson_id=None, grade_id=None, subject=book.subject,
lesson_number=chapter.number, chapter=chapter.number,
title=chapter.title, collection=collection,
)
except Exception as exc: # noqa: BLE001
logger.exception('rag.ingest_book: ذخیره‌ی صفحه %s شکست خورد', page)
failed_pages.append(page)
emit(kind='page_failed', page=page, error=str(exc))
if budget is not None:
budget -= 1
continue
totals['chunks_created'] += len(created)
# نشانه بعد از ذخیره‌ی موفق زده می‌شود، نه قبلش: اگر ذخیره شکست
# بخورد صفحه باید در اجرای بعدی دوباره دیده شود.
IngestedPage.objects.update_or_create(
collection=collection, page=page,
defaults={'chapter_title': chapter.title, 'chunk_count': len(rows)},
)
totals['pages_done'] += 1
if budget is not None:
budget -= 1
emit(kind='page', page=page, chapter=chapter.number, chunks=len(rows))
return _book_result(book, collection, totals, failed_pages, partial=False)
def _book_result(book: Textbook, collection: str, totals: dict,
failed_pages: list[int], *, partial: bool) -> dict:
model = settings.OPENAI_VISION_MODEL
price_in, price_out = MODEL_PRICE_PER_MTOK.get(model, (2.50, 10.00))
cost = (
totals['prompt_tokens'] / 1e6 * price_in
+ totals['completion_tokens'] / 1e6 * price_out
)
return {
'status': 'partial' if partial or failed_pages else 'success',
'book': book.key,
'collection': collection,
'pages_processed': totals['pages_done'],
'pages_skipped': totals['pages_skipped'],
'chunks_created': totals['chunks_created'],
'failed_pages': failed_pages,
'prompt_tokens': totals['prompt_tokens'],
'completion_tokens': totals['completion_tokens'],
'cost_usd': round(cost, 4),
'model': model,
}