قطع پاسخ تلفن هوشمند با پایتون؛ نسل پاسخ و رویداد صوتی قدیمی

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

۱. وقفهٔ مکالمه را به قرارداد رخداد تبدیل کنیم

در بخش barge-in استاندارد VoiceXML 2.0، امکان قطع پیام به سیاست برنامه و قابلیت بستر وابسته است. این زمینه، ضرورت تعریف رفتار وقفه را روشن می‌کند؛ کد ما پیاده‌سازی آن استاندارد نیست. اینکه صدای ورودی واقعاً گفتار باشد یا کاربر قصد قطع پاسخ داشته باشد، مسئلهٔ جداگانه‌ای است. ما رخداد interrupt را از پیش در فهرست ورودی می‌نویسیم.

چهار نوع رخداد داریم: start آغاز پاسخ تازه، interrupt ابطال پاسخ، ready آماده‌شدن یک قطعه و played تأیید قراردادی مصرف سر صف. دو نوع آخر، نسل و شناسهٔ قطعه دارند. نام played نباید با مشاهدهٔ شنیدن کاربر اشتباه شود؛ در این مدل فقط یک فرمان تألیفی است. هیچ میکروفن، بلندگو، ساعت واقعی یا بستهٔ شبکه‌ای وجود ندارد که چنین مشاهده‌ای را فراهم کند.

در مستندات رسمی پیام‌های Twilio Media Streams، پیام mark هم پس از اتمام پخش و هم پس از پاک‌کردن بافر می‌تواند بازگردد. پس معنای اعلان باید با قرارداد بستر خوانده شود. اینجا پیام mark را تجزیه نمی‌کنیم و played را جایگزین مستقیم آن نمی‌نامیم؛ فقط نشان می‌دهیم چرا اعلان دیررس نباید بی‌قید، وضعیت تازه را عوض کند.

۲. نسل، شناسهٔ قطعه و ترتیب مرزی

حالت اولیه نسل صفر، غیرفعال و دارای صف خالی است. هر start یا interrupt نسل را یک واحد زیاد می‌کند و صف معوق را پاک می‌کند. start حالت را فعال می‌کند؛ interrupt آن را غیرفعال نگه می‌دارد. بنابراین پس از start نخست، نسل یک داریم؛ وقفه نسل دو می‌سازد و start بعدی نسل سه. این شماره‌ها شمار پاسخ‌های پخش‌شده نیستند؛ شمارهٔ دورهٔ اعتبارند.

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

صف بر اساس ترتیب رسیدن ready ساخته می‌شود، نه اندازهٔ عدد شناسهٔ قطعه. played باید دقیقاً به سر صف جاری اشاره کند؛ تأیید قطعهٔ عقب‌تر نادیده گرفته می‌شود و ترتیب را عوض نمی‌کند. اگر سازندهٔ واقعی قطعات را بی‌ترتیب تحویل دهد، مرتب‌سازی رسانه به قرارداد دیگری نیاز دارد. این کد با نام FIFO، مسئلهٔ بازسازی ترتیب نمونه‌های صوتی را پنهان نمی‌کند.

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

۳. کد کامل و هجده بررسی با پاسخ دستی

فایل barge_in_generation.py با Python 3.14.5 و کتابخانهٔ استاندارد اجرا شده است. رخدادها dataclass هستند، ولی مستندات dataclasses پایتون تصریح می‌کند که annotation معمولاً نوع ورودی را اعتبارسنجی نمی‌کند. تابع validate پیش از تغییر حالت، نوع دقیق اعضا، زمان و شناسه‌ها را می‌آزماید؛ bool، NaN، زمان اعشاری و شناسهٔ صفر پذیرفته نمی‌شوند.

"""Offline authored generation-gated queue; no audio, clocks or network."""
from dataclasses import dataclass, replace
import json
import platform


@dataclass(frozen=True)
class Event:
    at_ms: int
    kind: str
    generation: int | None = None
    chunk: int | None = None


@dataclass(frozen=True)
class State:
    generation: int = 0
    active: bool = False
    queue: tuple = ()
    seen: tuple = ()
    played: tuple = ()
    cleared_chunks: int = 0


def validate(events):
    if type(events) not in (list, tuple):
        raise TypeError('events must be a list or tuple')
    previous = -1
    for event in events:
        if type(event) is not Event:
            raise TypeError('each item must be Event')
        if type(event.at_ms) is not int or event.at_ms < 0:
            raise ValueError('at_ms must be a nonnegative exact int in ms')
        if event.at_ms < previous:
            raise ValueError('events must be nondecreasing in at_ms')
        previous = event.at_ms
        if type(event.kind) is not str or event.kind not in ('start', 'interrupt', 'ready', 'played'):
            raise ValueError('unknown event kind')
        if event.kind in ('ready', 'played'):
            if any(type(v) is not int or v <= 0 for v in (event.generation, event.chunk)):
                raise ValueError('generation and chunk must be positive exact ints')
        elif event.generation is not None or event.chunk is not None:
            raise ValueError('control event must not carry generation/chunk')


def transition(state, event):
    """Called only after whole-stream validation by simulate()."""
    if event.kind in ('start', 'interrupt'):
        return replace(state, generation=state.generation + 1,
                       active=event.kind == 'start', queue=(), seen=(),
                       cleared_chunks=state.cleared_chunks + len(state.queue)), event.kind
    if not state.active or event.generation != state.generation:
        return state, 'ignored_generation'
    pair = (event.generation, event.chunk)
    if event.kind == 'ready':
        if pair in state.seen:
            return state, 'ignored_duplicate'
        return replace(state, queue=state.queue + (pair,), seen=state.seen + (pair,)), 'queued'
    if not state.queue or state.queue[0] != pair:
        return state, 'ignored_not_head'
    return replace(state, queue=state.queue[1:], played=state.played + (pair,)), 'played'


def simulate(events):
    validate(events)
    state, audit = State(), []
    for position, event in enumerate(events):
        state, outcome = transition(state, event)
        audit.append({'position': position, 'at_ms': event.at_ms,
                      'kind': event.kind, 'outcome': outcome,
                      'current_generation': state.generation,
                      'queue': [list(pair) for pair in state.queue]})
    return {'generation': state.generation, 'active': state.active,
            'queue': [list(pair) for pair in state.queue],
            'played': [list(pair) for pair in state.played],
            'cleared_chunks': state.cleared_chunks, 'audit': audit}


def rejects(value, exc=(TypeError, ValueError)):
    try:
        simulate(value)
    except exc:
        return True
    return False


def run_example():
    events = [Event(0, 'start'), Event(10, 'ready', 1, 1),
              Event(20, 'played', 1, 1), Event(30, 'ready', 1, 2),
              Event(40, 'interrupt'), Event(40, 'played', 1, 2),
              Event(50, 'ready', 1, 3), Event(60, 'start'),
              Event(70, 'ready', 3, 1), Event(80, 'played', 1, 2),
              Event(90, 'played', 3, 1)]
    result = simulate(events)
    hand_outcomes = ['start', 'queued', 'played', 'queued', 'interrupt',
                     'ignored_generation', 'ignored_generation', 'start',
                     'queued', 'ignored_generation', 'played']
    checks = {}
    checks['hand_written_event_outcomes'] = [r['outcome'] for r in result['audit']] == hand_outcomes
    checks['hand_written_final_state'] = (result['generation'], result['active'], result['queue'],
        result['played'], result['cleared_chunks']) == (3, True, [], [[1, 1], [3, 1]], 1)
    checks['initial_idle_ignores_callback'] = simulate([Event(0, 'ready', 1, 1)])['queue'] == []
    checks['interrupt_clears_pending_queue'] = simulate(events[:5])['queue'] == [] and simulate(events[:5])['cleared_chunks'] == 1
    checks['ready_after_interrupt_is_ignored'] = simulate(events[:7])['audit'][-1]['outcome'] == 'ignored_generation'
    checks['stale_play_cannot_remove_new_head'] = simulate(events[:10])['queue'] == [[3, 1]]
    checks['ready_duplicate_before_and_after_play'] = [r['outcome'] for r in simulate([
        Event(0, 'start'), Event(1, 'ready', 1, 1), Event(2, 'ready', 1, 1),
        Event(3, 'played', 1, 1), Event(4, 'ready', 1, 1)])['audit']] == ['start', 'queued', 'ignored_duplicate', 'played', 'ignored_duplicate']
    checks['duplicate_play_is_not_replayed'] = simulate([
        Event(0, 'start'), Event(1, 'ready', 1, 1), Event(2, 'played', 1, 1),
        Event(3, 'played', 1, 1)])['played'] == [[1, 1]]
    fifo = simulate([Event(0, 'start'), Event(1, 'ready', 1, 8), Event(2, 'ready', 1, 2),
                     Event(3, 'played', 1, 2), Event(4, 'played', 1, 8)])
    checks['arrival_fifo_not_numeric_chunk_sort'] = fifo['queue'] == [[1, 2]] and fifo['played'] == [[1, 8]] and fifo['audit'][3]['outcome'] == 'ignored_not_head'
    before = simulate([Event(0, 'start'), Event(1, 'ready', 1, 1), Event(2, 'played', 1, 1), Event(2, 'interrupt')])
    after = simulate([Event(0, 'start'), Event(1, 'ready', 1, 1), Event(2, 'interrupt'), Event(2, 'played', 1, 1)])
    checks['equal_time_uses_input_order'] = before['played'] == [[1, 1]] and after['played'] == []
    checks['repeated_interrupts_invalidate_each_epoch'] = simulate([Event(0, 'interrupt'), Event(0, 'interrupt'), Event(1, 'start')])['generation'] == 3
    checks['new_start_cancels_unfinished_previous'] = simulate([Event(0, 'start'), Event(1, 'ready', 1, 1), Event(2, 'start')])['cleared_chunks'] == 1
    checks['future_generation_cannot_enqueue'] = simulate([Event(0, 'start'), Event(1, 'ready', 2, 1)])['queue'] == []
    checks['empty_input_returns_idle'] = simulate([]) == {'generation': 0, 'active': False, 'queue': [], 'played': [], 'cleared_chunks': 0, 'audit': []}
    checks['rejects_time_type_nan_negative_and_order'] = all(rejects([Event(v, 'start')]) for v in (True, 1.0, float('nan'), -1)) and rejects([Event(2, 'start'), Event(1, 'interrupt')])
    checks['rejects_identity_and_control_payload_errors'] = all(rejects([Event(0, 'ready', g, c)]) for g, c in ((True, 1), (1, False), (1.0, 1), (1, 0), (0, 1))) and rejects([Event(0, 'start', 1, 1)])
    checks['rejects_unknown_event_and_container'] = rejects([Event(0, 'other')]) and rejects([{}]) and rejects(iter(events))
    saved = tuple(events)
    checks['repeatable_without_mutating_events'] = simulate(events) == simulate(tuple(events)) and tuple(events) == saved
    if len(checks) != 18 or not all(type(v) is bool and v for v in checks.values()):
        raise AssertionError(checks)
    return {'python_version': platform.python_version(), 'runtime': 'Python standard library only',
            'units': {'at_ms': 'authored milliseconds; not clock/latency measurement', 'generation': 'dimensionless epoch identifier'},
            'result': result, 'checks': checks, 'unique_checks': 18,
            'limitations': ['No VAD, TTS, sound, network, async tasks, locks or telephone execution.',
                            'Equal timestamps retain input position; FIFO is callback arrival order, not chunk-number order.',
                            'Played events are authored acknowledgements, not evidence of what a human heard.',
                            'Already recorded plays are history and cannot be undone; generation cancellation affects future accepted events.']}


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

تابع transition پس از اعتبارسنجی کل فهرست فراخوانی می‌شود. حالت تازه با replace ساخته می‌شود؛ این روش خواندن مسیر تغییر را آسان می‌کند، ولی قفل، تراکنش یا ایمنی هم‌زمانی ایجاد نمی‌کند. frozen نیز جانشین اعتبارسنجی یا مرز امنیتی نیست. ورودی‌های این مثال، مقادیر ساده و tupleهای محدودند و تغییر مستقیم فهرست رخدادها لازم نیست؛ بازاجرای آن باید همان نتیجه را بسازد.

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

۴. اجرای سناریو و معنای تاریخچه

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

{
  "python_version": "3.14.5",
  "runtime": "Python standard library only",
  "units": {
    "at_ms": "authored milliseconds; not clock/latency measurement",
    "generation": "dimensionless epoch identifier"
  },
  "result": {
    "generation": 3,
    "active": true,
    "queue": [],
    "played": [
      [
        1,
        1
      ],
      [
        3,
        1
      ]
    ],
    "cleared_chunks": 1,
    "audit": [
      {
        "position": 0,
        "at_ms": 0,
        "kind": "start",
        "outcome": "start",
        "current_generation": 1,
        "queue": []
      },
      {
        "position": 1,
        "at_ms": 10,
        "kind": "ready",
        "outcome": "queued",
        "current_generation": 1,
        "queue": [
          [
            1,
            1
          ]
        ]
      },
      {
        "position": 2,
        "at_ms": 20,
        "kind": "played",
        "outcome": "played",
        "current_generation": 1,
        "queue": []
      },
      {
        "position": 3,
        "at_ms": 30,
        "kind": "ready",
        "outcome": "queued",
        "current_generation": 1,
        "queue": [
          [
            1,
            2
          ]
        ]
      },
      {
        "position": 4,
        "at_ms": 40,
        "kind": "interrupt",
        "outcome": "interrupt",
        "current_generation": 2,
        "queue": []
      },
      {
        "position": 5,
        "at_ms": 40,
        "kind": "played",
        "outcome": "ignored_generation",
        "current_generation": 2,
        "queue": []
      },
      {
        "position": 6,
        "at_ms": 50,
        "kind": "ready",
        "outcome": "ignored_generation",
        "current_generation": 2,
        "queue": []
      },
      {
        "position": 7,
        "at_ms": 60,
        "kind": "start",
        "outcome": "start",
        "current_generation": 3,
        "queue": []
      },
      {
        "position": 8,
        "at_ms": 70,
        "kind": "ready",
        "outcome": "queued",
        "current_generation": 3,
        "queue": [
          [
            3,
            1
          ]
        ]
      },
      {
        "position": 9,
        "at_ms": 80,
        "kind": "played",
        "outcome": "ignored_generation",
        "current_generation": 3,
        "queue": [
          [
            3,
            1
          ]
        ]
      },
      {
        "position": 10,
        "at_ms": 90,
        "kind": "played",
        "outcome": "played",
        "current_generation": 3,
        "queue": []
      }
    ]
  },
  "checks": {
    "hand_written_event_outcomes": true,
    "hand_written_final_state": true,
    "initial_idle_ignores_callback": true,
    "interrupt_clears_pending_queue": true,
    "ready_after_interrupt_is_ignored": true,
    "stale_play_cannot_remove_new_head": true,
    "ready_duplicate_before_and_after_play": true,
    "duplicate_play_is_not_replayed": true,
    "arrival_fifo_not_numeric_chunk_sort": true,
    "equal_time_uses_input_order": true,
    "repeated_interrupts_invalidate_each_epoch": true,
    "new_start_cancels_unfinished_previous": true,
    "future_generation_cannot_enqueue": true,
    "empty_input_returns_idle": true,
    "rejects_time_type_nan_negative_and_order": true,
    "rejects_identity_and_control_payload_errors": true,
    "rejects_unknown_event_and_container": true,
    "repeatable_without_mutating_events": true
  },
  "unique_checks": 18,
  "limitations": [
    "No VAD, TTS, sound, network, async tasks, locks or telephone execution.",
    "Equal timestamps retain input position; FIFO is callback arrival order, not chunk-number order.",
    "Played events are authored acknowledgements, not evidence of what a human heard.",
    "Already recorded plays are history and cannot be undone; generation cancellation affects future accepted events."
  ]
}

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

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

۵. چهار تمرین پاسخ‌دار و حدود استفاده

تمرین اول: چرا پس از وقفه فقط صف را خالی نکنیم؟ پاسخ: ready قدیمی می‌تواند بعدتر بازگردد و صف خالی را دوباره پر کند. تغییر نسل، دلیل مستقل رد آن رخداد است؛ پاک‌کردن صف و کنترل اعتبار، دو مسئولیت مکمل‌اند.

تمرین دوم: اگر played و interrupt هر دو زمان چهل داشته باشند، پاسخ یکتا چیست؟ پاسخ: فقط با تعیین ترتیب ورودی یکتاست. played نخست، مصرف را ثبت می‌کند؛ interrupt نخست، نسل را عوض می‌کند و played قبلی را نامعتبر می‌سازد. زمان برابر به‌تنهایی ترتیب تعیین نمی‌کند.

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

تمرین چهارم: آیا موفقیت هجده بررسی، قطع بی‌درنگ صدای تلفن را ثابت می‌کند؟ پاسخ: خیر؛ کد نه صوت تولید می‌کند و نه صف دستگاه را می‌بیند. رشد تاریخچه و seen نیز برای ورودی طولانی محدود نشده است. منابع در ۲۴ مهر ۱۴۰۵ بازخوانی شده‌اند؛ آزمون واقعی صدا، هم‌زمانی، حافظه و بازگشت اعلان‌ها همچنان لازم است.

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

• W3C VoiceXML 2.0؛ بخش 4.1.5 Bargein
• Twilio Media Streams؛ پیام‌های mark و clear
• Python 3.14؛ dataclass و کنترل صریح نوع

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