پنجرهٔ حافظهٔ مکالمه با پایتون؛ جفت کامل نوبت‌ها و بودجهٔ نقاط کد

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

۱. تاریخچهٔ یک گفتگو را از سیاست ثابت جدا کنیم

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

راهنمای رسمی حافظهٔ کوتاه‌مدت LangChain تاریخچه را در محدودهٔ یک thread یا گفتگو توضیح می‌دهد و حذف، کوتاه‌سازی و خلاصه‌سازی را از هم جدا می‌کند. در کد ما این چارچوب نصب یا اجرا نشده است؛ state گراف، checkpointer یا پایگاه داده نداریم. یک قرارداد کوچکِ کتابخانهٔ استاندارد را مستقل بررسی می‌کنیم.

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

۲. فقط جفت کامل کاربر و دستیار را نگه داریم

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

جفت‌ها به ترتیب ورود نگه داشته می‌شوند. اگر تعداد یا مجموع طول از سقف بگذرد، قدیمی‌ترین جفت کامل حذف می‌شود؛ هرگز فقط نیمهٔ کاربر یا دستیار را کنار نمی‌گذاریم. تاریخچهٔ باقی‌مانده در محدودهٔ همین قرارداد از user آغاز و با assistant تمام می‌شود، اما این ساختار به‌تنهایی کامل‌بودن معنای گفتگو را اثبات نمی‌کند.

سقف نمونه دو جفت است و max_pairs می‌تواند از یک تا چهار باشد. این ترتیب، ترتیب استفادهٔ اخیرِ کش نیست: خواندن یا export کردنِ یک جفت آن را تازه‌تر نمی‌کند. حذف مدخل در کش LRU قرارداد دیگری دارد؛ جابه‌جا کردن تاریخچه به علت خواندن، ترتیب مکالمه را تغییر می‌دهد.

۳. بودجهٔ نقاط کد را با توکن و بایت یکی نگیریم

بودجه برابر مجموع len متن user و assistant در جفت‌های نگه‌داشته‌شده است. رشتهٔ پایتون دنبالهٔ نقاط کد یونیکد است. در آزمایش، e همراه علامت ترکیبی و یک ایموجی، سه نقطهٔ کد ولی هفت بایت UTF-8 دارد؛ تعداد نویسهٔ دیداری هم الزاماً سه نیست. هیچ شمارندهٔ توکن مدل در برنامه وجود ندارد.

سیاست ثابت، نام نقش‌ها، کلیدهای خروجی، جداکننده‌ها و قالب نهاییِ ارائه‌دهنده در این بودجه نیستند. بنابراین پرشدن بودجهٔ پانزده، به معنی پانزده توکن یا محدودبودن کل درخواست به پانزده بایت نیست. حتی یک سیاستِ ۲۵۶ نقطه‌ای می‌تواند کنار تاریخچهٔ دو نقطه‌ای قرار بگیرد؛ خروجی کلی طبیعتاً بزرگ‌تر می‌شود.

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

۴. کد را با قدیمی‌ترین جفت حذف‌شونده مقایسه کنیم

فایل conversation_window.py با Python 3.14.5 اجرا شده است. برای ذخیره از deque و popleft استفاده می‌کنیم. maxlen خودکار تعیین نشده، چون باید هم شرط تعداد و هم مجموع نقاط کد را کنترل کنیم و تعداد جفت‌های حذف‌شده را بدانیم. هر حذف، یک جفت از سمت قدیمی‌تر برمی‌دارد.

"""Original whole-pair conversation window; codepoints are not model tokens."""
from collections import deque
from itertools import product
import copy
import json
import platform
import re


def text(value):
    if type(value) is not str or not value.strip() or len(value) > 256:
        raise ValueError('nonblank actual str of at most 256 codepoints required')
    if any(ord(c) < 32 or 0xD800 <= ord(c) <= 0xDFFF for c in value):
        raise ValueError('control or surrogate outside text contract')


class ConversationWindow:
    def __init__(self, conversation_id, policy, *, max_pairs=2, budget=15):
        if type(conversation_id) is not str or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]{0,15}', conversation_id):
            raise ValueError('invalid conversation identifier')
        text(policy)
        if type(max_pairs) is not int or not 1 <= max_pairs <= 4:
            raise ValueError('max_pairs outside 1..4')
        if type(budget) is not int or not 1 <= budget <= 256:
            raise ValueError('history budget outside 1..256 codepoints')
        self._id = conversation_id
        self._policy = policy
        self._max_pairs = max_pairs
        self._budget = budget
        self._pairs = deque()
        self._dropped = 0

    def _scope(self, conversation_id):
        if type(conversation_id) is not str or conversation_id != self._id:
            raise ValueError('wrong logical conversation scope')

    def append_pair(self, conversation_id, pair):
        self._scope(conversation_id)
        if type(pair) is not dict or set(pair) != {'user', 'assistant'}:
            raise ValueError('one complete user/assistant pair required')
        text(pair['user'])
        text(pair['assistant'])
        incoming = (pair['user'], pair['assistant'])
        incoming_size = sum(map(len, incoming))
        if incoming_size > self._budget:
            raise ValueError('new pair alone exceeds history budget')
        # Work on a private candidate: invalid input never trims existing state.
        candidate = deque(self._pairs)
        candidate.append(incoming)
        count = sum(len(u) + len(a) for u, a in candidate)
        dropped_now = 0
        while len(candidate) > self._max_pairs or count > self._budget:
            u, a = candidate.popleft()
            count -= len(u) + len(a)
            dropped_now += 1
        self._pairs = candidate
        self._dropped += dropped_now

    def export(self, conversation_id):
        self._scope(conversation_id)
        return {'conversation_id': self._id, 'policy': self._policy,
                'pairs': [{'user': u, 'assistant': a} for u, a in self._pairs],
                'history_codepoints': sum(len(u) + len(a) for u, a in self._pairs),
                'dropped_pairs': self._dropped}

    def reset(self, conversation_id):
        self._scope(conversation_id)
        self._pairs.clear()
        self._dropped = 0


def pair(u, a):
    return {'user': u, 'assistant': a}


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


def run_example():
    fixture = [pair('دما', 'کم'), pair('ولتاژ', 'خوب'), pair('جریان', 'کم')]
    before = copy.deepcopy(fixture)
    window = ConversationWindow('Case7', 'پاسخ‌ها آموزشی‌اند')
    for item in fixture:
        window.append_pair('Case7', item)
    hand = {'conversation_id': 'Case7', 'policy': 'پاسخ‌ها آموزشی‌اند',
            'pairs': [pair('ولتاژ', 'خوب'), pair('جریان', 'کم')],
            'history_codepoints': 15, 'dropped_pairs': 1}
    initial = window.export('Case7')
    oversized_rejected = rejects(lambda: window.append_pair('Case7', pair('X' * 15, 'Y')))
    unchanged_after_reject = window.export('Case7') == initial
    window.append_pair('Case7', pair('پرسش', 'پاسخ'))
    after_four = window.export('Case7')
    limit_count = ConversationWindow('A', 'policy', max_pairs=1, budget=64)
    for item in fixture[:2]:limit_count.append_pair('A', item)
    unicode = ConversationWindow('U', 'policy', budget=3)
    unicode.append_pair('U', pair('e\u0301', '🙂'))
    big_policy = ConversationWindow('P', 'P' * 256, budget=2)
    big_policy.append_pair('P', pair('u', 'a'))
    scope_before = window.export('Case7')
    scope_reject = all(rejects(fn) for fn in [lambda: window.export('Other'), lambda: window.reset('Other'), lambda: window.append_pair('Other', pair('u', 'a'))])
    exported = window.export('Case7');exported['pairs'][0]['user'] = 'changed';exported['policy'] = 'changed'
    separate = ConversationWindow('Other', 'different policy')
    separate.append_pair('Other', pair('u', 'a'))
    raw_pair = pair('u', 'a'); copied = ConversationWindow('Copy', 'fixed')
    copied.append_pair('Copy', raw_pair); raw_pair['user'] = 'changed'
    reset = ConversationWindow('R', 'fixed')
    reset.append_pair('R', pair('u', 'a'));reset.reset('R')
    oracle_checks = []
    # Independent reference enumerates every suffix and chooses the longest fit;
    # it does not run a popleft loop or call append_pair to compute expected data.
    for lengths in product([2, 4, 6], repeat=3):
        for budget in [4, 8, 12]:
            for cap in [1, 2, 4]:
                current = []
                accepted = 0
                tested = ConversationWindow('O', 'fixed', max_pairs=cap, budget=budget)
                for step, size in enumerate(lengths):
                    item = pair(chr(65 + step) * (size - 1), 'a')
                    if size > budget:
                        old = tested.export('O')
                        oracle_checks.append(rejects(lambda: tested.append_pair('O', item)) and tested.export('O') == old)
                    else:
                        all_pairs = current + [item]
                        suffixes = [all_pairs[i:] for i in range(len(all_pairs) + 1)]
                        valid = [s for s in suffixes if len(s) <= cap and sum(len(x['user']) + len(x['assistant']) for x in s) <= budget]
                        current = max(valid, key=len)
                        accepted += 1
                        tested.append_pair('O', item)
                        got = tested.export('O')
                        oracle_checks.append(got['pairs'] == current and got['dropped_pairs'] == accepted - len(current))
    invalid_pairs = [{'user': 'u'}, {'user': 'u', 'assistant': 'a', 'extra': 1}, pair('', 'a'), pair(' ', 'a'), pair(1, 'a'), pair('u', '\ud800'), pair('u', 'a\n'), pair('u', 'A' * 257)]
    invalid_before = window.export('Case7')
    invalid_rejects = all(rejects(lambda item=item: window.append_pair('Case7', item)) for item in invalid_pairs)
    checks = {
        'literal_three_pair_hand_answer': initial == hand,
        'exact_budget_equality_is_retained': initial['history_codepoints'] == 15 and len(initial['pairs']) == 2,
        'new_pair_oversize_rejects_without_mutation': oversized_rejected and unchanged_after_reject,
        'oldest_whole_pair_is_trimmed': after_four['pairs'] == [pair('جریان', 'کم'), pair('پرسش', 'پاسخ')] and after_four['dropped_pairs'] == 2,
        'pair_limit_independent_of_character_budget': limit_count.export('A')['pairs'] == [fixture[1]] and limit_count.export('A')['history_codepoints'] == 8,
        'fixed_policy_is_outside_history_budget': big_policy.export('P')['policy'] == 'P' * 256 and big_policy.export('P')['history_codepoints'] == 2,
        'codepoints_are_not_graphemes_or_utf8_bytes': unicode.export('U')['history_codepoints'] == 3 and len('e\u0301🙂'.encode('utf-8')) == 7,
        'logical_scope_mismatch_rejects_unchanged': scope_reject and window.export('Case7') == scope_before,
        'export_is_detached_from_internal_pairs': window.export('Case7')['pairs'][0]['user'] == 'جریان' and window.export('Case7')['policy'] == 'پاسخ‌ها آموزشی‌اند',
        'separate_conversation_instances_do_not_share_pairs': separate.export('Other')['pairs'] == [pair('u', 'a')] and window.export('Case7') == scope_before,
        'reset_preserves_policy_clears_history_and_drop_count': reset.export('R') == {'conversation_id': 'R', 'policy': 'fixed', 'pairs': [], 'history_codepoints': 0, 'dropped_pairs': 0},
        'exact_pair_fields_and_text_validation': invalid_rejects and window.export('Case7') == invalid_before,
        'constructor_integer_bounds_exclude_bool': all(rejects(fn) for fn in [lambda: ConversationWindow('A', 'p', max_pairs=True), lambda: ConversationWindow('A', 'p', max_pairs=0), lambda: ConversationWindow('A', 'p', max_pairs=5), lambda: ConversationWindow('A', 'p', budget=True), lambda: ConversationWindow('A', 'p', budget=0), lambda: ConversationWindow('A', 'p', budget=257)]),
        'identifier_and_policy_contract': all(rejects(fn) for fn in [lambda: ConversationWindow('', 'p'), lambda: ConversationWindow('A B', 'p'), lambda: ConversationWindow(True, 'p'), lambda: ConversationWindow('A', ''), lambda: ConversationWindow('A', 'p' * 257)]),
        'pair_input_is_copied_to_scalar_tuple': fixture == before and copied.export('Copy')['pairs'] == [pair('u', 'a')],
        'all_stored_rows_are_complete_pairs': all(set(x) == {'user', 'assistant'} for x in window.export('Case7')['pairs']),
        'independent_enumerated_suffix_oracle': all(oracle_checks),
        'reset_one_scope_does_not_change_other_instance': (separate.reset('Other') is None) and window.export('Case7') == scope_before,
    }
    assert len(checks) == 18 and all(type(x) is bool and x for x in checks.values()), checks
    return {'python': platform.python_version(), 'dependencies': 'Python standard library only',
            'example_kind': 'Offline synthetic whole-pair codepoint window, no framework/LLM/token counting',
            'pair_sizes_hand': [5, 8, 7], 'after_three_pairs': initial, 'after_four_pairs': after_four,
            'oversized_new_pair_rejected_unchanged': oversized_rejected and unchanged_after_reject,
            'independent_suffix_steps_inside_one_group': len(oracle_checks), 'checks': checks}


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

سه جفتِ دستی «دما/کم»، «ولتاژ/خوب» و «جریان/کم» به‌ترتیب پنج، هشت و هفت نقطهٔ کد دارند. واژه‌ها دادهٔ آزمایش‌اند و نتیجهٔ حسگر نیستند. با سقف دو و بودجهٔ پانزده، جفت اول حذف و دو جفت آخر با مجموع دقیق پانزده نگه داشته می‌شوند. dropped_pairs برابر یک است و خود سیاست تغییر نکرده است.

جفت چهارم «پرسش/پاسخ» هشت نقطه دارد؛ جفت قدیمیِ هشت‌تایی حذف می‌شود و هفت به‌علاوهٔ هشت باقی می‌ماند، پس شمار حذف‌ها دو می‌شود. جفتِ پانزده X و یک Y، شانزده نقطه دارد و رد می‌شود. آزمون snapshot قبل و بعد نشان می‌دهد چنین ردشدنی، تاریخچهٔ قبلی را پاک یا کوتاه نکرده است.

مرجع مستقل، همهٔ پسوندهای ممکنِ فهرست کاندیدا را می‌سازد و بلندترین پسوندِ مجاز را انتخاب می‌کند؛ حلقهٔ popleft برنامه را تکرار نمی‌کند. ۲۴۳ سناریوی محدود، هر کدام با سه گام، ۷۲۹ مقایسه در یک گروه می‌سازند. هجده گروه واقعی داریم؛ برابری مرزی، ورودی بزرگ، نوع، شناسه و کپی خروجی نیز بررسی شده‌اند.

۵. بازنشانی و خروجی را در محدودهٔ گفتگو نگه داریم

append، export و reset شناسهٔ مطابق می‌خواهند. شناسهٔ دیگر خطا می‌دهد و حالت قبلی باقی می‌ماند، ولی این کنترل، احراز هویت یا جداسازی امن کاربران نیست. نمونه‌های جدا، تاریخچهٔ مشترک ندارند؛ تغییر دیکشنری ورودی یا خروجیِ صادرشده نیز جفت ذخیره‌شده را تغییر نمی‌دهد. کلاس برای عملیات هم‌زمان یا چند فرایند طراحی نشده است.

{
  "python": "3.14.5",
  "dependencies": "Python standard library only",
  "example_kind": "Offline synthetic whole-pair codepoint window, no framework/LLM/token counting",
  "pair_sizes_hand": [
    5,
    8,
    7
  ],
  "after_three_pairs": {
    "conversation_id": "Case7",
    "policy": "پاسخ‌ها آموزشی‌اند",
    "pairs": [
      {
        "user": "ولتاژ",
        "assistant": "خوب"
      },
      {
        "user": "جریان",
        "assistant": "کم"
      }
    ],
    "history_codepoints": 15,
    "dropped_pairs": 1
  },
  "after_four_pairs": {
    "conversation_id": "Case7",
    "policy": "پاسخ‌ها آموزشی‌اند",
    "pairs": [
      {
        "user": "جریان",
        "assistant": "کم"
      },
      {
        "user": "پرسش",
        "assistant": "پاسخ"
      }
    ],
    "history_codepoints": 15,
    "dropped_pairs": 2
  },
  "oversized_new_pair_rejected_unchanged": true,
  "independent_suffix_steps_inside_one_group": 729,
  "checks": {
    "literal_three_pair_hand_answer": true,
    "exact_budget_equality_is_retained": true,
    "new_pair_oversize_rejects_without_mutation": true,
    "oldest_whole_pair_is_trimmed": true,
    "pair_limit_independent_of_character_budget": true,
    "fixed_policy_is_outside_history_budget": true,
    "codepoints_are_not_graphemes_or_utf8_bytes": true,
    "logical_scope_mismatch_rejects_unchanged": true,
    "export_is_detached_from_internal_pairs": true,
    "separate_conversation_instances_do_not_share_pairs": true,
    "reset_preserves_policy_clears_history_and_drop_count": true,
    "exact_pair_fields_and_text_validation": true,
    "constructor_integer_bounds_exclude_bool": true,
    "identifier_and_policy_contract": true,
    "pair_input_is_copied_to_scalar_tuple": true,
    "all_stored_rows_are_complete_pairs": true,
    "independent_enumerated_suffix_oracle": true,
    "reset_one_scope_does_not_change_other_instance": true
  }
}

reset تاریخچه و شمار حذف‌ها را در همین شیء خالی می‌کند و سیاست را حفظ می‌کند. این کار، پاک‌شدن داده از دیسک، گزارش، سرویس بیرونی یا نسخهٔ پشتیبان را ثابت نمی‌کند. منابع در ۲۵ مهر ۱۴۰۵ خوانده شدند؛ راهنمای جاری چارچوب، تاریخ انتشار مشخصی در بخش خوانده‌شده نداشت و مستندات 3.14.8 از اجرای 3.14.5 جدا ثبت شده‌اند.

پرسش اول: برای جا دادن جفت تازه، نصف پاسخ را ببریم؟ پاسخ: خیر؛ قرارداد جفت کامل است. اگر خود جفت جا نشود، بی‌تغییر رد می‌شود؛ سیاست خلاصه‌سازی جداست.

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

پرسش سوم: reset سیاست ثابت را هم حذف می‌کند؟ پاسخ: خیر؛ فقط تاریخچه و شمار حذف‌ها صفر می‌شوند. این تصمیمِ صریح همین رابط است.

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

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

• LangChain؛ حافظهٔ کوتاه‌مدت و مرز حذف پیام‌ها
• Python؛ deque و حذف از ابتدا
• Python؛ str و نقاط کد یونیکد

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