Hacktoberfest 2026：メンテナが10月に向けて印を付けた、オープンで初心者向けの issue。 Hacktoberfest の issue を見る

StreamPacerWrapper.end_input() sets _input_ended before buffered sentences reach _sentences; last sentence(s) silently never sent to TTS

オープン
  • 何か問題がありますか？
#7,810 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

@Hugo-DDT がすでに取り組んでいます。

2026年10月14日 から。

  • #7816 @Hugo-DDT による — オープン

評価

難易度
3/5
見積もり時間
1〜2日
初心者へのやさしさ
25/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
停滞
技術スタック
python
領域
audio-video-rtc

調査の方向性

The race is in livekit/agents/tts/stream_pacer.py: end_input() sets _input_ended before the sentence tokenizer flushes, and _send_task can break on _input_ended and not _sentences. Read end_input(), _recv_task's finally block and the _send_task loop, then run the included repro_pacer_tail.py with k from -6 to 12. Done means every k reports FULL. Pull request #7816 is already open against this, so check it before starting.

索引モデルが issue の本文から書いたものです。

説明

Summary

StreamPacerWrapper.end_input() sets _input_ended = True before the sentence tokenizer has delivered the buffered tail to _recv_task. If the send loop wakes up in that window, _send_task sees _input_ended and not _sentences, breaks, and the last sentence(s) are never sent to TTS. No error is raised: the turn finishes "successfully" with the end of the reply missing.

We hit this on a deployed voice agent: the reply was cut after the first sentence, the rest of the LLM text never reached TTS.

Versions

livekit-agents 1.6.10 through 1.8.7; livekit-agents/livekit/agents/tts/stream_pacer.py on main is identical. Python 3.14, Windows and Linux.

Setup: StreamAdapter(tts=<non-streaming TTS>, text_pacing=SentenceStreamPacer(min_remaining_audio=3.0, max_text_length=200)).

Cause
def end_input(self) -> None:
    self._sent_stream.end_input()
    self._input_ended = True          # set before the tokenizer flushed its tail
    ...

self._sent_stream.end_input() only schedules the tokenizer flush; the final sentence is appended to self._sentences by _recv_task on a later event-loop iteration. Meanwhile _send_task can be woken by the pending call_later wakeup timer (or _wakeup_event), and then

if self._closing or (self._input_ended and not self._sentences):
    break

exits the loop with the tail still in flight. _event_ch closes, StreamAdapterWrapper._synthesize finishes, and the remaining text is dropped.

Deterministic repro (no network, no wall clock)

The pacer wakeup (what its timer does) is injected k event-loop iterations before/after stream.end_input(). Only k=0 loses the tail, which matches the timing window above.

repro_pacer_tail.py 
import asyncio

from livekit.agents import APIConnectOptions, tts, utils
from livekit.agents.tts import SentenceStreamPacer, StreamAdapter

SAMPLE_RATE = 24000
REPLY = (
    "I'm doing well, thanks! How are you? Today we'll practice minimal "
    "pairs, words that differ by just one sound, like sheep and ship."
)

synthesized: list[str] = []
pacers: list[object] = []

_orig_wrap = SentenceStreamPacer.wrap


def _recording_wrap(self, sent_stream, audio_emitter):
    wrapper = _orig_wrap(self, sent_stream, audio_emitter)
    pacers.append(wrapper)
    return wrapper


SentenceStreamPacer.wrap = _recording_wrap


class FakeChunkedStream(tts.ChunkedStream):
    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        synthesized.append(self._input_text)
        output_emitter.initialize(
            request_id=utils.shortuuid(),
            sample_rate=SAMPLE_RATE,
            num_channels=1,
            mime_type="audio/pcm",
        )
        await asyncio.sleep(0.6)  # network latency
        seconds = len(self._input_text) * 0.08
        sent = 0.0
        while sent < seconds:
            output_emitter.push(b"\x00\x00" * int(SAMPLE_RATE * 0.1))
            sent += 0.1
            await asyncio.sleep(0.01)
        output_emitter.flush()


class FakeTTS(tts.TTS):
    def __init__(self) -> None:
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=SAMPLE_RATE,
            num_channels=1,
        )

    def synthesize(self, text, *, conn_options=APIConnectOptions()):
        return FakeChunkedStream(tts=self, input_text=text, conn_options=conn_options)


async def run(k: int) -> tuple[int, float]:
    synthesized.clear()
    pacers.clear()
    adapter = StreamAdapter(
        tts=FakeTTS(),
        text_pacing=SentenceStreamPacer(min_remaining_audio=3.0, max_text_length=200),
    )
    stream = adapter.stream()

    async def produce() -> None:
        for i in range(0, len(REPLY), 4):
            stream.push_text(REPLY[i : i + 4])
            await asyncio.sleep(0.003)
        # first sentence is in TTS, pacer idles on its wakeup timer with an empty queue
        while not (
            pacers
            and synthesized
            and pacers[0]._wakeup_timer is not None
            and not pacers[0]._sentences
        ):
            await asyncio.sleep(0.003)
        pacer = pacers[0]
        if k >= 0:
            stream.end_input()
            for _ in range(k):
                await asyncio.sleep(0)
            pacer._wakeup_event.set()
        else:
            pacer._wakeup_event.set()
            for _ in range(-k):
                await asyncio.sleep(0)
            stream.end_input()

    async def consume() -> float:
        played = 0.0
        async for ev in stream:
            played += ev.frame.duration
        return played

    _, played = await asyncio.gather(produce(), consume())
    await stream.aclose()
    return len(synthesized), played


async def main() -> None:
    import livekit.agents

    print("livekit-agents", livekit.agents.__version__)
    for k in range(-6, 13):
        requests, played = await run(k)
        verdict = "FULL" if requests == 2 else "TAIL DROPPED"
        print(f"k={k:+3d} tts_requests={requests} {verdict} audio={played:.2f}s", flush=True)


asyncio.run(main())

Output on 1.8.7:

livekit-agents 1.8.7
k= -1 tts_requests=2 FULL audio=10.43s
k= +0 tts_requests=1 TAIL DROPPED audio=1.92s
k= +1 tts_requests=2 FULL audio=10.43s
(all other k in -6..+12: FULL)
Suggested fix

Do not set _input_ended in end_input(). _recv_task already sets it in its finally once the sentence stream is exhausted, i.e. after every sentence is in self._sentences:

def end_input(self) -> None:
    self._sent_stream.end_input()
    if self._audio_emitter._dst_ch.closed:
        self._closing = True
        self._wakeup_event.set()

With this change the deterministic repro gives FULL for all k, and a timer-based sweep (25 offsets around the wakeup deadline) went from 2 dropped tails to 0.

Related but different: #3133, #3140 (adds _wakeup_event.set() but keeps the _input_ended and not _sentences race), #4440.

主要言語
Python
スター
14.6k
フォーク
3.9k
平均マージ
3日 1時間
マージ済み PR（30日）
166

環境構築

このプロジェクトには開発コンテナ、Dockerfile、コントリビューションガイドがありません。まず README を読み、一般的な手順ははじめてのコントリビューションガイドを参照してください。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

livekit/agents のほかの issue

livekit/agents の issue をすべて見る

似ている issue

Python の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。