نسخهٔ معتبر سند در RAG؛ مرجع، انقضا و امتناع از بازگشت خودکار با پایتون

در پاسخ‌گوی متصل به پایگاه دانش، سندی که امتیاز جست‌وجوی بالاتری دارد لزوماً مرجع معتبر امروز نیست. نسخهٔ پیش‌نویس، اطلاعیهٔ آینده و سند پس‌گرفته‌شده می‌توانند متن کاملاً مرتبطی داشته باشند. این درس یک سیاست محدود برای انتخاب سند روی تصویر ثابتِ ساختگی می‌سازد. کد کامل، پاسخ دستی و هجده بررسی داریم؛ نمایه، مدل زبانی، تماس تلفنی و سند واقعی اجرا یا ارزیابی نمی‌شوند.

۱. مرجع، شمارهٔ نسخه و تاریخ تغییر یک مفهوم نیستند

واژگان DCMI Metadata Terms، زمان اعتبار منبع، تاریخ تغییر آن و رابطهٔ نسخه‌ها را جدا تعریف می‌کنند. این تفکیک برای طراحی مفید است، اما به‌خودی‌خود نمی‌گوید کدام واحد سازمانی تصمیم‌گیرنده است. سیاست مرجع و رفتار هنگام ابهام باید برای کاربرد مورد نظر نوشته شود؛ افزودن برچسب metadata جای اجرای سیاست را نمی‌گیرد.

در گزارش NVIDIA در ۲۳ مهر ۱۴۰۵، گردش‌کارهایی شامل خواندن اسناد و پیروی از سیاست سازمان توضیح داده می‌شوند. این نوشتهٔ سازنده زمینهٔ روز بحث است؛ قاعدهٔ انتخاب مرجعِ مثال ما از آن استخراج نشده و هیچ نتیجهٔ عملکرد مدل به کد حاضر نسبت داده نمی‌شود.

موضوع ثابت را ivr-power می‌نامیم؛ نام، مشخصات یک محصول واقعی نیست. مرجع engineering رتبهٔ بیست و support رتبهٔ ده دارد. عدد بزرگ‌تر، صرفاً اولویتِ سیاست ساختگی است. رتبه‌ها باید مثبت، صحیح و متمایز باشند. سند با مرجع ثبت‌نشده وارد انتخاب نمی‌شود؛ با این حال، دانستن نام مرجع یا درج آن در یک رکورد، هویت صادرکننده را احراز نمی‌کند.

۲. اول سرنسخه را تعیین کنیم، سپس اعتبار زمانی را بسنجیم

سیاست ما محافظه‌کارانه و صریح است: ابتدا رکوردهای همان موضوع، با مرجع ثبت‌شده و وضعیت غیردرفت را در نظر می‌گیریم؛ سپس بالاترین مرجع و در همان مرجع، بزرگ‌ترین شمارهٔ نسخه را انتخاب می‌کنیم. «سرنسخه» همین نامزد برتر است، نه لزوماً سند مجاز برای پاسخ. پس‌گرفتن را از فهرست نامزدها حذف نمی‌کنیم تا نسخهٔ قدیمی بی‌صدا زنده نشود.

پس از تعیین سرنسخه، وضعیت و بازهٔ زمانی آن کنترل می‌شود. آغاز بازه شامل و پایان آن خارج است: start <= now < end. در لحظهٔ دقیق انقضا، متن وارد خروجی نمی‌شود. یک ثانیه پیش از آن مجاز است، اگر باقی شرایط برقرار باشند. بازهٔ خالی یا وارونه، خطای قرارداد داده است و با نتیجهٔ عادیِ «منقضی» یکی نیست.

طبق مستند زمان دارای منطقه در پایتون، ساعت بدون اطلاعات منطقه، یک لحظهٔ بی‌ابهام را مشخص نمی‌کند. کد، datetime دارای منطقه می‌خواهد و همه را برای مقایسه به UTC می‌برد. دو نمایش با offset متفاوت از یک لحظه، تصمیم یکسان می‌دهند. زمان آزمون ورودی ثابت است؛ ساعت سیستم خوانده نمی‌شود و رابط نمایش تاریخ شمسی نیز در این تابع پیاده نشده است.

۳. تصویر ثابت و پاسخ دستیِ انتخاب سند

چهار رکورد داریم. A1 نسخهٔ یک engineering و A2 نسخهٔ دو همان مرجع هستند؛ S99 نسخهٔ نودونه support و D9 پیش‌نویس نسخهٔ نه engineering است. در لحظهٔ آزمون، A2 تاییدشده و داخل بازه است. پاسخ دستی A2 است: نودونه بودن شمارهٔ سندِ مرجع ضعیف‌تر، رتبهٔ آن را بالاتر نمی‌برد و پیش‌نویس نیز سرنسخهٔ موثر نمی‌شود.

اسنپ‌شات، tuple کوچکی از رکوردهای ثابت است. پیش از تصمیم، همهٔ شناسه‌ها، شماره‌ها، زمان‌ها و وضعیت‌ها کنترل می‌شوند؛ شناسهٔ تکراری یا شمارهٔ بولی پذیرفته نیست. حتی خرابی رکورد نامرتبط، در این قرارداد سخت‌گیرانه کل تصویر را نامعتبر می‌کند. برنامه اصلاح خودکار، حذف بی‌صدای دادهٔ خراب یا اتصال به مخزن سند ندارد.

برای بازبینی، رکوردها بر پایهٔ شناسه مرتب، زمان‌ها یکسان‌سازی و همراه جدول اولویت سریال می‌شوند. SHA256 این نمایش در خروجی می‌آید. جابه‌جایی ترتیب ورودی، اثر انگشت را عوض نمی‌کند؛ تغییر متن یا سیاست آن را عوض می‌کند. این برچسب برای اتصال نتیجه به ورودی معلوم مفید است؛ امضای دیجیتال، اثبات اصالت یا شاهد صحت متن نیست. نسخهٔ قابل اعتمادِ ورودی همچنان باید جدا نگهداری شود.

۴. کد کامل و هجده بررسی اجراشده

فایل rag_document_validity.py را ذخیره و با پایتون اجرا کنید. اجرای ثبت‌شده Python 3.14.5 و فقط کتابخانهٔ استاندارد است. ورودی از صد سند بیشتر پذیرفته نمی‌شود؛ این سقف آموزشی است و ظرفیت تولیدی را نشان نمی‌دهد. جدول رتبه، زمان درخواست و موضوع، ورودی‌های آشکار تصمیم هستند.

"""Authored snapshot policy only; no retrieval, authenticated registry or live clock."""
import copy
from dataclasses import asdict, dataclass, replace
from datetime import datetime, timedelta, timezone
import hashlib
import json
import platform

UTC = timezone.utc


@dataclass(frozen=True)
class Document:
    doc_id: str
    topic: str
    revision: int
    authority: str
    status: str
    start: datetime
    end: datetime
    text: str


def aware_utc(value):
    if type(value) is not datetime or value.tzinfo is None or value.utcoffset() is None:
        raise ValueError('Exact aware datetime required; timezone is not guessed')
    return value.astimezone(UTC)


def select_document(snapshot, authority_rank, topic, now):
    if type(snapshot) is not tuple or len(snapshot) > 100:
        raise ValueError('A tuple of at most100 documents is required')
    if type(authority_rank) is not dict or not authority_rank:
        raise ValueError('An explicit nonempty authority policy is required')
    for label, rank in authority_rank.items():
        if type(label) is not str or not label or type(rank) is not int or rank <= 0:
            raise ValueError('Authority labels and exact positive integer ranks required')
    if len(set(authority_rank.values())) != len(authority_rank):
        raise ValueError('This authored policy requires distinct authority ranks')
    if type(topic) is not str or not topic:
        raise ValueError('Exact nonempty topic required')
    instant = aware_utc(now)
    documents, seen = [], set()
    for item in snapshot:
        if type(item) is not Document:
            raise ValueError('Exact Document records required')
        for value in (item.doc_id, item.topic, item.authority, item.text):
            if type(value) is not str or not value.strip():
                raise ValueError('Nonempty declared string metadata required')
        if item.doc_id in seen or type(item.revision) is not int or item.revision <= 0:
            raise ValueError('Unique IDs and exact positive integer revisions required')
        if item.status not in ('approved', 'draft', 'withdrawn'):
            raise ValueError('Unknown lifecycle status')
        start, end = aware_utc(item.start), aware_utc(item.end)
        if start >= end:
            raise ValueError('Nonempty forward validity interval required')
        documents.append(replace(item, start=start, end=end))
        seen.add(item.doc_id)
    serial = []
    for item in sorted(documents, key=lambda x: x.doc_id):
        record = asdict(item)
        record['start'], record['end'] = item.start.isoformat(), item.end.isoformat()
        serial.append(record)
    serialized = json.dumps({'documents': serial, 'authority_rank': authority_rank},
                            ensure_ascii=False, sort_keys=True, separators=(',', ':'))
    fingerprint = hashlib.sha256(serialized.encode('utf-8')).hexdigest()
    candidates = [x for x in documents if x.topic == topic and x.authority in authority_rank and x.status != 'draft']
    result = {'decision': 'abstain', 'reason': 'no_registered_head', 'document_id': None,
              'head_ids': [], 'text': None, 'topic': topic,
              'requested_at_utc': instant.isoformat(), 'snapshot_sha256': fingerprint}
    if not candidates:
        return result
    head_key = max((authority_rank[x.authority], x.revision) for x in candidates)
    heads = [x for x in candidates if (authority_rank[x.authority], x.revision) == head_key]
    result['head_ids'] = sorted(x.doc_id for x in heads)
    if len(heads) != 1:
        result['reason'] = 'ambiguous_head'
        return result
    head = heads[0]
    if head.status == 'withdrawn':
        result['reason'] = 'withdrawn_head'
    elif instant < head.start:
        result['reason'] = 'not_yet_valid'
    elif instant >= head.end:
        result['reason'] = 'expired_head'
    else:
        result.update(decision='selected', reason='current_registered_head', document_id=head.doc_id, text=head.text)
    return result


def rejects(call):
    try:
        call()
    except ValueError:
        return True
    return False


def run_example():
    now = datetime(2026, 10, 16, 10, tzinfo=UTC)
    start, end = datetime(2026, 10, 1, tzinfo=UTC), datetime(2026, 11, 1, tzinfo=UTC)
    old = Document('A1', 'ivr-power', 1, 'engineering', 'approved', start, end, 'Authored old procedure')
    head = Document('A2', 'ivr-power', 2, 'engineering', 'approved', now - timedelta(hours=1), now + timedelta(hours=14), 'Authored revision2 procedure')
    support = Document('S99', 'ivr-power', 99, 'support', 'approved', start, end, 'Authored support note')
    draft = replace(head, doc_id='D9', revision=9, status='draft')
    snapshot, ranks = (old, head, support, draft), {'engineering': 20, 'support': 10}
    before_snapshot, before_ranks = copy.deepcopy(snapshot), copy.deepcopy(ranks)
    run = lambda docs=snapshot, policy=ranks, instant=now, topic='ivr-power': select_document(docs, policy, topic, instant)
    baseline = run()
    future = replace(head, doc_id='F3', revision=3, start=now + timedelta(seconds=1))
    expired = replace(head, doc_id='X3', revision=3, start=start, end=now)
    withdrawn = replace(head, doc_id='W3', revision=3, status='withdrawn')
    ambiguous = replace(head, doc_id='A2-copy')
    unregistered = replace(head, doc_id='U100', authority='unregistered', revision=100)
    scenarios = {name: run(snapshot + (item,)) for name, item in
                 [('future', future), ('expired', expired), ('withdrawn', withdrawn), ('ambiguous', ambiguous)]}
    checks = {
        'hand_oracle_selects_registered_engineering_revision_two': baseline['decision'] == 'selected' and baseline['document_id'] == 'A2' and baseline['text'] == 'Authored revision2 procedure',
        'higher_authority_outweighs_lower_authority_revision_99': baseline['document_id'] != 'S99' and run((head, support))['document_id'] == 'A2',
        'highest_revision_in_chosen_authority_wins_regardless_of_input_order': run(tuple(reversed(snapshot))) == baseline,
        'newer_draft_is_not_an_effective_revision_head': baseline['head_ids'] == ['A2'] and run((head, draft))['document_id'] == 'A2',
        'exact_validity_start_is_included': run(instant=head.start)['document_id'] == 'A2',
        'exact_expiry_end_is_excluded': run(instant=head.end)['reason'] == 'expired_head',
        'one_second_before_expiry_remains_selected': run(instant=head.end - timedelta(seconds=1))['document_id'] == 'A2',
        'future_head_blocks_automatic_old_or_lower_priority_fallback': scenarios['future']['reason'] == 'not_yet_valid' and scenarios['future']['document_id'] is None,
        'expired_head_blocks_automatic_old_or_lower_priority_fallback': scenarios['expired']['reason'] == 'expired_head' and scenarios['expired']['text'] is None,
        'withdrawn_head_blocks_automatic_fallback': scenarios['withdrawn']['reason'] == 'withdrawn_head' and scenarios['withdrawn']['document_id'] is None,
        'unregistered_authority_cannot_win': run(snapshot + (unregistered,))['document_id'] == 'A2',
        'topic_must_match_exactly': run(topic='IVR-power')['reason'] == 'no_registered_head',
        'equal_head_priority_and_revision_abstains_even_without_lexical_tiebreak': scenarios['ambiguous']['reason'] == 'ambiguous_head' and scenarios['ambiguous']['head_ids'] == ['A2', 'A2-copy'],
        'empty_or_draft_only_snapshot_has_no_registered_head': run(())['reason'] == run((draft,))['reason'] == 'no_registered_head',
        'same_instant_at_other_explicit_offset_produces_same_decision': run(instant=now.astimezone(timezone(timedelta(hours=3, minutes=30)))) == baseline,
        'snapshot_digest_order_invariant_but_binds_text_and_policy': run(tuple(reversed(snapshot)))['snapshot_sha256'] == baseline['snapshot_sha256'] and run((old, replace(head, text='Changed authored text'), support, draft))['snapshot_sha256'] != baseline['snapshot_sha256'] and run(policy={'engineering': 21, 'support': 10})['snapshot_sha256'] != baseline['snapshot_sha256'],
        'invalid_metadata_duplicates_naive_time_and_policy_rejected': all((rejects(lambda: run(snapshot + (head,))), rejects(lambda: run((replace(head, revision=True),))), rejects(lambda: run((replace(head, status='unknown'),))), rejects(lambda: run((replace(head, end=head.start),))), rejects(lambda: run((replace(head, text=' '),))), rejects(lambda: run(instant=now.replace(tzinfo=None))), rejects(lambda: run(policy={'engineering': True})), rejects(lambda: run(policy={'engineering': 20, 'support': 20})), rejects(lambda: run(list(snapshot))))),
        'inputs_and_policy_not_mutated': snapshot == before_snapshot and ranks == before_ranks,
    }
    assert len(checks) == 18 and all(type(value) is bool and value for value in checks.values()), checks
    return {'python': platform.python_version(), 'dependencies': 'standard library only',
            'policy': 'Authored authority-first revision head, then lifecycle/time gate; no fallback; [start,end)',
            'scope': 'Synthetic snapshot only; labels/hash do not authenticate authority or prove content truth',
            'baseline': baseline, 'abstention_scenarios': scenarios, 'checks': checks}


if __name__ == '__main__':
    print(json.dumps(run_example(), ensure_ascii=False, indent=2))

هجده بررسی نام‌دار، پاسخ دستی A2، مرز آغاز و انقضا، اختلاف اولویت، پیش‌نویس، پس‌گرفتن، ابهام، منطقهٔ زمانی، اثر انگشت و ورودی نامعتبر را می‌آزمایند. اجرای دوباره، تعداد بررسی‌های مستقل را افزایش نمی‌دهد. شناسهٔ سند انتخاب‌شده فقط هنگام selected پر می‌شود؛ در حالت امتناع، نامزدهای سرنسخه و علت باقی می‌مانند ولی متن پاسخ None است.

{
  "python": "3.14.5",
  "dependencies": "standard library only",
  "policy": "Authored authority-first revision head, then lifecycle/time gate; no fallback; [start,end)",
  "scope": "Synthetic snapshot only; labels/hash do not authenticate authority or prove content truth",
  "baseline": {
    "decision": "selected",
    "reason": "current_registered_head",
    "document_id": "A2",
    "head_ids": [
      "A2"
    ],
    "text": "Authored revision2 procedure",
    "topic": "ivr-power",
    "requested_at_utc": "2026-10-16T10:00:00+00:00",
    "snapshot_sha256": "e208545f2f21cdf13e48489d0704ad6d8b3a68fd282a47bd9b1d81a9feca6147"
  },
  "abstention_scenarios": {
    "future": {
      "decision": "abstain",
      "reason": "not_yet_valid",
      "document_id": null,
      "head_ids": [
        "F3"
      ],
      "text": null,
      "topic": "ivr-power",
      "requested_at_utc": "2026-10-16T10:00:00+00:00",
      "snapshot_sha256": "bfe110868fe6376a8019c45468fc22b825a56e65821cad201ed7206511ffed98"
    },
    "expired": {
      "decision": "abstain",
      "reason": "expired_head",
      "document_id": null,
      "head_ids": [
        "X3"
      ],
      "text": null,
      "topic": "ivr-power",
      "requested_at_utc": "2026-10-16T10:00:00+00:00",
      "snapshot_sha256": "fcbc3b6e467b36d8cfcc9fcf84ea0f8192dfd8d3ef2f84b9ee49d9902471f8bd"
    },
    "withdrawn": {
      "decision": "abstain",
      "reason": "withdrawn_head",
      "document_id": null,
      "head_ids": [
        "W3"
      ],
      "text": null,
      "topic": "ivr-power",
      "requested_at_utc": "2026-10-16T10:00:00+00:00",
      "snapshot_sha256": "e0fcf591ea3dfb9617aae7215668e3fe8ab43750730d9035ba4a6261fa1c60eb"
    },
    "ambiguous": {
      "decision": "abstain",
      "reason": "ambiguous_head",
      "document_id": null,
      "head_ids": [
        "A2",
        "A2-copy"
      ],
      "text": null,
      "topic": "ivr-power",
      "requested_at_utc": "2026-10-16T10:00:00+00:00",
      "snapshot_sha256": "a7eda5ae38a35824ae8403be7a31992d7719436d6404dbafac1b2d7c78a734fb"
    }
  },
  "checks": {
    "hand_oracle_selects_registered_engineering_revision_two": true,
    "higher_authority_outweighs_lower_authority_revision_99": true,
    "highest_revision_in_chosen_authority_wins_regardless_of_input_order": true,
    "newer_draft_is_not_an_effective_revision_head": true,
    "exact_validity_start_is_included": true,
    "exact_expiry_end_is_excluded": true,
    "one_second_before_expiry_remains_selected": true,
    "future_head_blocks_automatic_old_or_lower_priority_fallback": true,
    "expired_head_blocks_automatic_old_or_lower_priority_fallback": true,
    "withdrawn_head_blocks_automatic_fallback": true,
    "unregistered_authority_cannot_win": true,
    "topic_must_match_exactly": true,
    "equal_head_priority_and_revision_abstains_even_without_lexical_tiebreak": true,
    "empty_or_draft_only_snapshot_has_no_registered_head": true,
    "same_instant_at_other_explicit_offset_produces_same_decision": true,
    "snapshot_digest_order_invariant_but_binds_text_and_policy": true,
    "invalid_metadata_duplicates_naive_time_and_policy_rejected": true,
    "inputs_and_policy_not_mutated": true
  }
}

چهار سناریوی اضافه‌شده به تصویر اصلی به ترتیب not_yet_valid، expired_head، withdrawn_head و ambiguous_head می‌سازند. در ابهام، دو شناسهٔ هم‌مرجع و هم‌نسخه داریم؛ حتی برابر بودن متن، مجوز انتخاب تصادفی یکی از آن‌ها نیست. عیب‌یابی محتوا، مرحله‌ای جدا از مرتب‌سازی واژه‌نامه‌ای شناسه‌هاست.

۵. چهار پرسش پاسخ‌دار دربارهٔ بازگشت و محدودیت‌ها

پرسش اول: چرا با انقضای نسخهٔ سه به نسخهٔ دو برنمی‌گردیم؟ پاسخ: سیاست این درس، بازگشت خودکار را ممنوع کرده است؛ نسخهٔ قدیمی یا مرجع ضعیف‌تر ممکن است دیگر پاسخ مناسب نباشد. این انتخاب عمومی همهٔ سازمان‌ها نیست. در راهنمای به‌روزرسانی پایگاه دانش، تعیین چرخهٔ اعتبار و آماده‌شدن نمایه جدا بررسی می‌شود.

پرسش دوم: اگر سرنسخه برای فردا برنامه‌ریزی شده باشد، نسخهٔ امروز پاسخ می‌دهد؟ پاسخ: در همین سیاست خیر؛ ابتدا سرنسخه تعیین شده و سپس دروازهٔ زمان بسته می‌ماند. برای فعال ماندن نسخهٔ قبلی تا موعد انتقال، باید سیاست دیگری طراحی و با آزمون مرزی مستند شود.

پرسش سوم: آیا برچسب approved یا هش، سند را معتبر می‌کند؟ پاسخ: خیر؛ این فیلدها دادهٔ ورودی‌اند. احراز مرجع، مجوز مخاطب، امضا، سلامت مخزن و صحت محتوایی اجرا نشده‌اند. این درس قاعدهٔ حقوقی یا تایید اعتبار عمومی یک سند ارائه نمی‌کند.

پرسش چهارم: انتخاب سرنسخه، تعارض ادعاهای متن را هم حل می‌کند؟ پاسخ: خیر؛ آزمون تعارض شواهد RAG مسئلهٔ دیگری است. انتخاب منبع، بازیابی بخش مناسب و درستی پاسخ باید جدا بررسی شوند. منابع در ۲۴ مهر ۱۴۰۵ خوانده شده‌اند؛ تاریخ توصیهٔ DCMI، گزارش روز سازنده و نسخهٔ پایتونِ اجرا، سه اطلاعات متفاوت‌اند.

منابع و مطالعهٔ بیشتر

• DCMI Metadata Terms؛ تمایز اعتبار، تغییر و رابطهٔ نسخه، توصیهٔ ۲۰۲۰
• Python 3.14؛ زمان دارای منطقه و مقایسهٔ لحظه‌ها
• گزارش NVIDIA دربارهٔ گردش‌کارهای اسنادی Nemotron، ۱۵ اکتبر ۲۰۲۶

بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.