real RAG answers on POST /api/ai/ask, plus SSE streaming and answer cache

When RAG is configured the endpoint retrieves lesson-scoped chunks and
generates a grounded answer; otherwise the previous deterministic stub
runs unchanged, so keyless runs and existing tests are unaffected.
Quota / idempotency / user-row lock / clean-503 behaviour is preserved on
every path.

models.py
  - Conversation.subject (CharField) and Conversation.lesson_number
    (IntegerField): retrieval metadata sent on the first ask and reused
    for the rest of the conversation.
migrations/0004_conversation_lesson_number_conversation_subject.py
  - migration for the two new fields.
serializers.py
  - AskRequestSerializer: optional subject / grade / lessonNumber to pin
    the Qdrant collection and lesson/chapter.
  - AskResponseSerializer: usedChunks (retrieved [{title, summary}]) and
    retrievalDebug (collection / filterKey / chapter-or-lesson / counts,
    emitted only under DEBUG).
views.py
  - generate_tutor_answer(): unchanged signature; RAG-configured branch
    calls rag.services generation, stub branch otherwise.
  - AskAIView.post(): resolve subject/grade/lessonNumber (falling back to
    the Conversation), check the answer cache via resolve_target(), run
    retrieve_context + generate_answer on a miss and populate the cache,
    return usedChunks (+ retrievalDebug under DEBUG). Failures still
    release the reserved quota slot and surface as 503.
  - AskAIView._streaming_response(): opt-in "stream": true returns a
    text/event-stream of chunk / done / error events; same retrieval,
    phase-3 persist, idempotency and quota accounting as the JSON path,
    and it consults the answer cache first. Non-stream JSON contract is
    untouched.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-29 02:30:33 +03:30
parent ff47c8d4ba
commit 5416fbb2ba
4 changed files with 271 additions and 35 deletions

View File

@@ -0,0 +1,23 @@
# Generated by Django 5.2.17 on 2026-08-28 18:56
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ai_tutor', '0003_askidempotency_suggested_follow_ups'),
]
operations = [
migrations.AddField(
model_name='conversation',
name='lesson_number',
field=models.IntegerField(blank=True, null=True),
),
migrations.AddField(
model_name='conversation',
name='subject',
field=models.CharField(blank=True, default='', max_length=100),
),
]

View File

@@ -6,6 +6,12 @@ class Conversation(models.Model):
lesson_id = models.IntegerField(null=True, blank=True) # ارجاع سبک به content.Lesson
chapter_id = models.IntegerField(null=True, blank=True) # شناسه‌ی فصل RAG (خارج از content/)
grade_id = models.IntegerField(null=True, blank=True) # ارجاع سبک به content.Grade
# metadataِ retrieval برای لایه‌ی RAG (اپ rag): نامِ درس (رشته) و
# شماره‌ی درس. اولین بار از بدنه‌ی POST /api/ai/ask می‌آن و اینجا ذخیره
# می‌شن تا ادامه‌ی همون مکالمه دوباره لازم نباشه فرستاده بشن. نامِ
# collectionِ Qdrant از روی subject + عنوانِ گرید ساخته می‌شه.
subject = models.CharField(max_length=100, blank=True, default='')
lesson_number = models.IntegerField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):

View File

@@ -7,16 +7,37 @@ class AskRequestSerializer(serializers.Serializer):
chapterId = serializers.CharField(required=False, allow_null=True, allow_blank=True)
gradeId = serializers.IntegerField(required=False, allow_null=True)
questionId = serializers.CharField(required=False, allow_null=True, allow_blank=True)
# metadataِ retrieval برای لایه‌ی RAG (اپ rag): برای زدن به دیتای واقعیِ
# کتاب‌ها در Qdrant باید subject (مثلاً «علوم تجربی») + gradeId + lessonNumber
# همگی مطابقِ دادهٔ موجود ارسال بشن؛ اگه نباشن، پاسخ به‌صورتِ تدریسِ عمومی
# (بدونِ استناد به کتاب) تولید می‌شه. در ادامه‌ی همون مکالمه لازم نیست
# دوباره فرستاده بشن (روی Conversation ذخیره می‌شن).
subject = serializers.CharField(required=False, allow_null=True, allow_blank=True)
# جزءِ «پایه» در نامِ collectionِ Qdrant (مثلاً «ششم»). اگه بیاد مستقیم
# استفاده می‌شه؛ اگه نه، از روی gradeId و content.Grade.title / نگاشتِ
# RAG_GRADE_TITLE_OVERRIDES ساخته می‌شه.
grade = serializers.CharField(required=False, allow_null=True, allow_blank=True)
lessonNumber = serializers.IntegerField(required=False, allow_null=True)
userQuestion = serializers.CharField()
stream = serializers.BooleanField(required=False, default=False)
class UsedChunkSerializer(serializers.Serializer):
title = serializers.CharField(allow_blank=True)
summary = serializers.CharField()
class AskResponseSerializer(serializers.Serializer):
conversationId = serializers.CharField()
messageId = serializers.CharField()
answer = serializers.CharField()
suggestedFollowUps = serializers.ListField(child=serializers.CharField())
remainingQuota = serializers.IntegerField()
# تکه‌های درسیِ بازیابی‌شده از Qdrant که پاسخ بر پایه‌شون ساخته شده؛
# خالی یعنی retrieval انجام نشده (نبودِ subject/lessonNumber یا خاموش‌بودنِ RAG).
usedChunks = UsedChunkSerializer(many=True, required=False, default=list)
# فقط وقتی DEBUG=True: از کدام collection/فصل چانک‌ها برداشته شد (بازبینیِ داخلی).
retrievalDebug = serializers.DictField(required=False)
class MessageSerializer(serializers.Serializer):

View File

@@ -1,7 +1,9 @@
import json
import logging
from django.conf import settings
from django.db import IntegrityError, transaction
from django.http import Http404
from django.http import Http404, StreamingHttpResponse
from django.shortcuts import get_object_or_404
from django.utils import timezone
from drf_spectacular.types import OpenApiTypes
@@ -92,44 +94,40 @@ def generate_tutor_answer(
user_question: str, grade_id, lesson_id, context=None
) -> tuple[str, list[str]]:
"""
مرزِ (boundary) تولیدِ پاسخِ AI Tutor — محلِ اتصال به مدل واقعی (مثلاً
Claude API از طریق Anthropic SDK) در آینده.
مرزِ (boundary) تولیدِ پاسخِ AI Tutor.
🏗️ اصلاحِ معماریِ این دور: gradeId/lessonId که در تسکِ قبلی از این
امضا حذف شده بودن، برگشتن — چون اینا metadata ای هستن که لایه‌ی RAG
(که جداگانه و در آینده با Qdrant پیاده می‌شه) برای retrieve‌کردنِ حدوداً
۵ chunk مرتبط از knowledge base واقعاً بهشون نیاز داره (کنارِ خودِ سوال).
یعنی امضای این تابع الان: (user_question, grade_id, lesson_id, context).
امضا: (user_question, grade_id, lesson_id, context). context یک پارامترِ
عمومیِ اختیاریِ application-generated است که این تابع صرفاً مصرف می‌کنه،
بدونِ اینکه بدونه از کجا retrieve شده. جریان:
- grade_id و lesson_id: مستقیماً از همون فیلدهای gradeId/lessonId که در
بدنه‌ی درخواستِ POST /api/ai/ask می‌آن (v.get('gradeId')،
v.get('lessonId')) به این تابع پاس داده می‌شن؛ قراردادِ API endpoint
دست‌نخورده باقی می‌مونه.
- context: یک پارامترِ عمومیِ اختیاریِ application-generated (رشته/شیء/
لیستِ پیام — فرمتش تصمیمِ لایه‌ی RAG در آینده‌ست) که *این* تابع صرفاً
مصرف می‌کنه، بدونِ اینکه بدونه از کجا اومده (retrieval از Qdrant؟
چیزِ دیگه؟). یعنی جریانِ آینده:
gradeId + subject + lessonNumber + userQuestion → لایه‌ی RAG (اپ rag،
Qdrant) → ~۴ chunkِ انتخاب‌شده به‌عنوانِ context → این تابع (لایه‌ی
تولید) → provider (OpenAI از طریقِ GapGPT)
gradeId + lessonId + userQuestion → لایه‌ی RAG (Qdrant) →
~۵ chunkِ انتخاب‌شده به‌عنوانِ context → این تابع (لایه‌ی تولید) →
provider
AskAIView خودش retrieval را انجام می‌ده و نتیجه را به‌عنوانِ context
(لیستی از dictهای {title, summary, score}) پاس می‌ده؛ این تابع فقط
تولید می‌کنه. مالکیتِ تاریخچه‌ی مکالمه همیشه دستِ دیتابیسِ خودمونه، نه
provider.
این تابع (و provider واقعی‌ای که جایگزینِ استابِ زیر می‌شه) نباید خودش
Conversation/Message رو از دیتابیس بخونه یا تاریخچه‌ی خامِ مکالمه نگه
داره — مالکیتِ تاریخچه‌ی مکالمه همیشه دستِ دیتابیسِ خودمونه، نه لایه‌ی
provider؛ grade_id/lesson_id صرفاً metadataی retrieval هستن، نه
تاریخچه.
در همین تسک، AskAIView هنوز context ای نمی‌سازه (لایه‌ی RAG/Qdrant
جداگانه و بعداً پیاده می‌شه) و همیشه context=None می‌فرسته؛ پس رفتارِ
فعلی (پاسخِ استابِ قطعی) بدونِ تغییر باقی می‌مونه — این فقط مرز رو برای
آینده آماده می‌کنه.
اگه RAG پیکربندی نشده باشه (نبودِ OPENAI_API_KEY/QDRANT_API_KEY یا
RAG_FORCE_MOCK)، همون پاسخِ stubِ قطعیِ قبلی برمی‌گرده تا اجرای بدونِ
کلید و تست‌ها دست‌نخورده کار کنن.
"""
from rag.services.config import rag_is_configured
if rag_is_configured():
from rag.services.generation import generate_answer
context_chunks = context if isinstance(context, list) else None
# استثناهای شبکه/provider اینجا خفه نمی‌شن؛ AskAIView اون‌ها رو به
# 503 AIProviderUnavailable تبدیل می‌کنه (و اسلاتِ رزروشده رو آزاد).
return generate_answer(user_question, context_chunks)
context_note = f' (با context: {context})' if context else ''
answer = (
f'این یک پاسخ نمونه (stub) برای سوال «{user_question}»{context_note} است. '
'برای اتصال به مدل واقعی هوش مصنوعی، این تابع را در ai_tutor/views.py '
'جایگزین کنید (مثلاً با فراخوانی Anthropic API).'
'برای اتصال به مدل واقعی هوش مصنوعی، OPENAI_API_KEY و QDRANT_API_KEY '
'را در .env تنظیم کنید (لایه‌ی RAG در اپ rag).'
)
follow_ups = ['یک مثال دیگر بزن', 'بیشتر توضیح بده']
return answer, follow_ups
@@ -208,6 +206,8 @@ class AskAIView(APIView):
lesson_id=raw_lesson_id if _is_db_ref_id(raw_lesson_id) else None,
chapter_id=raw_chapter_id if _is_db_ref_id(raw_chapter_id) else None,
grade_id=v.get('gradeId'),
subject=(v.get('subject') or '').strip(),
lesson_number=v.get('lessonNumber'),
)
raw_question_id = v.get('questionId')
@@ -218,11 +218,68 @@ class AskAIView(APIView):
question_id=raw_question_id if _is_db_ref_id(raw_question_id) else None,
)
# قفل ردیف کاربر اینجا آزاد شده؛ فراخوانی AI بیرون از هر تراکنشی اجرا می‌شود.
# context برای لایه‌ی RAG/Qdrant آینده در نظر گرفته شده؛ فعلاً None است.
# لایه‌ی RAG (اپ rag) اینجا context را می‌سازد: پرسش → embedding →
# جستجوی Qdrant (محدود به subject+grade+lessonNumber) → چند خلاصه.
# اگر RAG پیکربندی نشده باشد context=None می‌ماند و generate_tutor_answer
# همان پاسخِ stub را می‌دهد. خطای شبکه/provider در همین try به 503
# تبدیل و اسلاتِ رزروشده آزاد می‌شود.
grade_id = v.get('gradeId')
lesson_id = v.get('lessonId')
subject = (v.get('subject') or conversation.subject or '').strip()
grade_title = (v.get('grade') or '').strip()
lesson_number = v.get('lessonNumber')
if lesson_number is None:
lesson_number = conversation.lesson_number
# --- مسیرِ استریم (SSE) --- فقط وقتی stream=true و RAG پیکربندی‌شده.
# قراردادِ JSONِ غیراستریم دست‌نخورده می‌ماند؛ Swagger که stream
# نمی‌فرسته همون مسیرِ قبلی رو می‌گیره.
if v.get('stream'):
from rag.services.config import rag_is_configured
if rag_is_configured():
return self._streaming_response(
request=request, v=v, user=user, conversation=conversation,
user_message=user_message, conversation_is_new=conversation_is_new,
idempotency_key=idempotency_key, used_today=used_today,
subject=subject, grade_id=grade_id, grade_title=grade_title,
lesson_number=lesson_number,
)
used_chunks: list[dict] = []
retrieval_meta: dict = {}
context = None
try:
from rag.services.config import rag_is_configured
if rag_is_configured():
from rag.services import answer_cache
from rag.services.generation import generate_answer
from rag.services.retrieval import resolve_target, retrieve_context
_col, _fk, _fv = resolve_target(subject, grade_title, grade_id, lesson_number)
cache_key = answer_cache.make_key(_col, _fv, v['userQuestion'])
cached = answer_cache.get(cache_key)
if cached is not None:
answer_text = cached['answer']
follow_ups = list(cached['followUps'])
used_chunks = cached['usedChunks']
retrieval_meta = cached.get('retrievalMeta') or {}
else:
used_chunks, retrieval_meta = retrieve_context(
v['userQuestion'], subject=subject, grade_id=grade_id,
grade_title=grade_title, lesson_number=lesson_number,
)
used_chunks = [
{'title': c.get('title', ''), 'summary': c.get('summary', '')}
for c in used_chunks
]
answer_text, follow_ups = generate_answer(v['userQuestion'], used_chunks or None)
answer_cache.set(cache_key, {
'answer': answer_text, 'followUps': follow_ups,
'usedChunks': used_chunks, 'retrievalMeta': retrieval_meta,
})
else:
answer_text, follow_ups = generate_tutor_answer(
v['userQuestion'], grade_id, lesson_id, context
)
@@ -269,9 +326,138 @@ class AskAIView(APIView):
'answer': answer_text,
'suggestedFollowUps': follow_ups,
'remainingQuota': remaining_quota,
'usedChunks': [
{'title': c.get('title', ''), 'summary': c.get('summary', '')}
for c in used_chunks
],
}
# فقط برای بازبینیِ داخلیِ خودمان (نه کاربر): از کدام collection/فصل
# چانک‌ها برداشته شد. فقط وقتی DEBUG روشن است در پاسخ می‌آید.
if settings.DEBUG and retrieval_meta:
data['retrievalDebug'] = {
'collection': retrieval_meta.get('collection'),
'filterKey': retrieval_meta.get('filterKey'),
'chapterOrLesson': retrieval_meta.get('filterValue'),
'gradeTitle': retrieval_meta.get('gradeTitle'),
'chunkCount': retrieval_meta.get('chunkCount'),
'chunkTitles': [c.get('title', '') for c in used_chunks],
}
return Response(AskResponseSerializer(data).data)
def _streaming_response(self, *, request, v, user, conversation, user_message,
conversation_is_new, idempotency_key, used_today,
subject, grade_id, grade_title, lesson_number):
"""
نسخه‌ی SSE از post(): همون منطقِ retrieval + کوتا + idempotency، ولی
پاسخ توکن‌به‌توکن استریم می‌شه. رویدادها:
event: chunk data: {"delta": "..."}
event: done data: {conversationId, messageId, remainingQuota, usedChunks[, retrievalDebug]}
event: error data: {"detail": "..."}
فاز ۱ (رزرو کوتا + conversation + user_message) قبلاً در post() انجام شده.
"""
from rag.services import answer_cache
from rag.services.generation import default_follow_ups, stream_answer
from rag.services.retrieval import resolve_target, retrieve_context
def _sse(event: str, payload: dict) -> str:
return f'event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n'
def _gen():
nonlocal used_today
_col, _fk, _fv = resolve_target(subject, grade_title, grade_id, lesson_number)
cache_key = answer_cache.make_key(_col, _fv, v['userQuestion'])
try:
cached = answer_cache.get(cache_key)
if cached is not None:
answer_text = cached['answer']
follow_ups = list(cached['followUps'])
used_chunks = cached['usedChunks']
retrieval_meta = cached.get('retrievalMeta') or {}
yield _sse('chunk', {'delta': answer_text})
else:
used_chunks, retrieval_meta = retrieve_context(
v['userQuestion'], subject=subject, grade_id=grade_id,
grade_title=grade_title, lesson_number=lesson_number,
)
used_chunks = [
{'title': c.get('title', ''), 'summary': c.get('summary', '')}
for c in used_chunks
]
parts: list[str] = []
for delta in stream_answer(v['userQuestion'], used_chunks or None):
parts.append(delta)
yield _sse('chunk', {'delta': delta})
answer_text = ''.join(parts).strip() or 'متأسفم، الان نتونستم پاسخِ مناسبی بسازم.'
follow_ups = default_follow_ups()
answer_cache.set(cache_key, {
'answer': answer_text, 'followUps': follow_ups,
'usedChunks': used_chunks, 'retrievalMeta': retrieval_meta,
})
except Exception:
logger.exception('stream generate failed for user_id=%s', user.id)
self._release_reserved_slot(conversation, user_message, conversation_is_new)
yield _sse('error', {'detail': AIProviderUnavailable.default_detail})
return
# فاز ۳: ثبتِ پیامِ دستیار + idempotency (همون منطقِ post()).
try:
with transaction.atomic():
assistant_message = Message.objects.create(
conversation=conversation, role='assistant', text=answer_text,
question_id=v.get('questionId') or None,
)
if idempotency_key:
AskIdempotency.objects.create(
user_id=user.id, idempotency_key=idempotency_key,
conversation=conversation, user_message=user_message,
assistant_message=assistant_message,
suggested_follow_ups=follow_ups,
)
except IntegrityError:
existing = AskIdempotency.objects.filter(
user_id=user.id, idempotency_key=idempotency_key,
).select_related('assistant_message').first() if idempotency_key else None
if existing is None:
self._release_reserved_slot(conversation, user_message, conversation_is_new)
yield _sse('error', {'detail': AIProviderUnavailable.default_detail})
return
self._release_reserved_slot(conversation, user_message, conversation_is_new)
yield _sse('done', {
'conversationId': str(existing.conversation_id),
'messageId': str(existing.assistant_message_id),
'suggestedFollowUps': existing.suggested_follow_ups,
'remainingQuota': max(DAILY_AI_QUOTA - used_today, 0),
'usedChunks': [],
})
return
used_today += 1
done = {
'conversationId': str(conversation.id),
'messageId': str(assistant_message.id),
'suggestedFollowUps': follow_ups,
'remainingQuota': max(DAILY_AI_QUOTA - used_today, 0),
'usedChunks': [
{'title': c.get('title', ''), 'summary': c.get('summary', '')}
for c in used_chunks
],
}
if settings.DEBUG and retrieval_meta:
done['retrievalDebug'] = {
'collection': retrieval_meta.get('collection'),
'filterKey': retrieval_meta.get('filterKey'),
'chapterOrLesson': retrieval_meta.get('filterValue'),
'gradeTitle': retrieval_meta.get('gradeTitle'),
'chunkCount': retrieval_meta.get('chunkCount'),
'chunkTitles': [c.get('title', '') for c in used_chunks],
}
yield _sse('done', done)
response = StreamingHttpResponse(_gen(), content_type='text/event-stream')
response['Cache-Control'] = 'no-cache'
response['X-Accel-Buffering'] = 'no'
return response
def _release_reserved_slot(self, conversation, user_message, conversation_is_new):
"""پاک‌کردن رزرو فاز ۱ وقتی فاز بعدی شکست می‌خورد، تا نه پیام یتیم بماند و نه کوتا مصرف‌شده حساب شود."""
with transaction.atomic():