diff --git a/livekit-agents/livekit/agents/ipc/channel.py b/livekit-agents/livekit/agents/ipc/channel.py index 98dbfb4a5..d44830925 100644 --- a/livekit-agents/livekit/agents/ipc/channel.py +++ b/livekit-agents/livekit/agents/ipc/channel.py @@ -146,6 +146,8 @@ def _read_thread(self) -> None: self._loop.call_soon_threadsafe(self._read_q.put_nowait, msg) except RuntimeError: break + except (OSError, EOFError, ValueError): + break except ChannelClosed: break diff --git a/livekit-agents/livekit/agents/tokenize/token_stream.py b/livekit-agents/livekit/agents/tokenize/token_stream.py index 0fec50b48..7d1df9361 100644 --- a/livekit-agents/livekit/agents/tokenize/token_stream.py +++ b/livekit-agents/livekit/agents/tokenize/token_stream.py @@ -44,7 +44,7 @@ def push_text(self, text: str) -> None: self._event_ch.send_nowait(TokenData(token=buf)) for i, tok in enumerate(buf_toks): - tok_i = self._buf.index(tok) + tok_i = self._buf.find(tok) self._buf = self._buf[tok_i + len(tok) :].lstrip() buf_toks = [] diff --git a/livekit-agents/livekit/agents/transcription/tts_forwarder.py b/livekit-agents/livekit/agents/transcription/tts_forwarder.py index 3c4f6b5f8..0dcedd47e 100644 --- a/livekit-agents/livekit/agents/transcription/tts_forwarder.py +++ b/livekit-agents/livekit/agents/transcription/tts_forwarder.py @@ -175,8 +175,7 @@ def push_text(self, text: str) -> None: def mark_text_segment_end(self) -> None: self._check_not_closed() stream = self._forming_segments.text.sentence_stream - stream.mark_segment_end() - self._task_set.create_task(stream.aclose()) + stream.end_input() # create a new segment on "mark_text_segment_end" # further text can already be pushed even if mark_audio_segment_end has not been @@ -235,8 +234,7 @@ async def _forward_task(): seg.forward_start_time = time.time() async for ev in sentence_stream: - if ev.type == tokenize.TokenEventType.TOKEN: - await self._sync_sentence_co(seg, ev.token, rtc_seg_q) + await self._sync_sentence_co(seg, ev.token, rtc_seg_q) rtc_seg_q.put_nowait(None) await forward_task diff --git a/livekit-agents/livekit/agents/voice_assistant/agent_output.py b/livekit-agents/livekit/agents/voice_assistant/agent_output.py index f6a0f2ce4..2f28267dc 100644 --- a/livekit-agents/livekit/agents/voice_assistant/agent_output.py +++ b/livekit-agents/livekit/agents/voice_assistant/agent_output.py @@ -189,8 +189,6 @@ async def _read_generated_audio_task(): handle._buf_ch.send_nowait(audio.frame) # we're only flushing once, so we know we can break at the end of the first segment - if audio.end_of_segment: - break # self._log_debug( # f"tts finished synthesising {audio_duration:.2f}s audio (streamed)" @@ -212,7 +210,7 @@ async def _read_generated_audio_task(): if handle._tr_fwd is not None: handle._tr_fwd.mark_text_segment_end() - tts_stream.flush() + tts_stream.end_input() await read_atask await tts_stream.aclose()