Voice Agent API

1
"""AssemblyAI Native Voice Agent Plugin for PipeCat."""
2
3
from __future__ import annotations
4
5
import asyncio
6
import base64
7
import json
8
import logging
9
from datetime import datetime, timezone
10
11
import websockets
12
import websockets.asyncio.client
13
14
from pipecat.frames.frames import (
15
    BotStartedSpeakingFrame,
16
    BotStoppedSpeakingFrame,
17
    CancelFrame,
18
    EndFrame,
19
    ErrorFrame,
20
    Frame,
21
    InputAudioRawFrame,
22
    InterimTranscriptionFrame,
23
    LLMContextFrame,
24
    StartFrame,
25
    TranscriptionFrame,
26
    TTSAudioRawFrame,
27
    TTSStartedFrame,
28
    TTSStoppedFrame,
29
    UserStartedSpeakingFrame,
30
    UserStoppedSpeakingFrame,
31
)
32
from pipecat.processors.frame_processor import FrameDirection
33
from pipecat.services.llm_service import LLMService
34
35
logger = logging.getLogger(__name__)
36
37
SAMPLE_RATE = 24000
38
NUM_CHANNELS = 1
39
40
41
class AssemblyAIRealtimeLLMService(LLMService):
42
    def __init__(self, *, url: str, api_key: str, system_prompt: str = "", **kwargs) -> None:
43
        super().__init__(**kwargs)
44
        self._url = url
45
        self._api_key = api_key
46
        self._system_prompt = system_prompt
47
        self._websocket = None
48
        self._receive_task = None
49
        self._session_ready = False
50
        self._current_response_id = None
51
        self._bot_speaking = False
52
        self._user_transcript_buf = ""
53
        self._pending_tools: list[dict] = []
54
55
    async def start(self, frame: StartFrame) -> None:
56
        await super().start(frame)
57
        await self._connect()
58
59
    async def stop(self, frame: EndFrame) -> None:
60
        await super().stop(frame)
61
        await self._disconnect()
62
63
    async def cancel(self, frame: CancelFrame) -> None:
64
        await super().cancel(frame)
65
        await self._disconnect()
66
67
    async def _connect(self) -> None:
68
        headers = {"Authorization": self._api_key}
69
        try:
70
            self._websocket = await websockets.asyncio.client.connect(
71
                self._url, additional_headers=headers,
72
            )
73
            self._receive_task = asyncio.create_task(
74
                self._receive_task_handler(), name="AssemblyAIRealtime._recv",
75
            )
76
            logger.info(f"Connected to AssemblyAI Voice Agent API at {self._url}")
77
            if self._system_prompt:
78
                await self._send({"type": "session.update", "session": {"system_prompt": self._system_prompt}})
79
        except Exception as e:
80
            logger.error(f"Failed to connect: {e}")
81
            await self.push_frame(ErrorFrame(str(e)))
82
83
    async def _disconnect(self) -> None:
84
        if self._receive_task:
85
            self._receive_task.cancel()
86
            try:
87
                await self._receive_task
88
            except (asyncio.CancelledError, Exception):
89
                pass
90
            self._receive_task = None
91
        if self._websocket:
92
            await self._websocket.close()
93
            self._websocket = None
94
        self._session_ready = False
95
        self._bot_speaking = False
96
        self._current_response_id = None
97
98
    async def _send(self, msg: dict) -> None:
99
        if self._websocket:
100
            try:
101
                await self._websocket.send(json.dumps(msg))
102
            except websockets.exceptions.ConnectionClosedOK:
103
                pass
104
            except Exception as e:
105
                logger.error(f"WebSocket send error: {e}")
106
                await self.push_frame(ErrorFrame(str(e)))
107
108
    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
109
        await super().process_frame(frame, direction)
110
        if isinstance(frame, InputAudioRawFrame):
111
            await self._send_user_audio(frame)
112
        elif isinstance(frame, LLMContextFrame):
113
            await self._handle_context(frame)
114
        else:
115
            await self.push_frame(frame, direction)
116
117
    async def _send_user_audio(self, frame: InputAudioRawFrame) -> None:
118
        if not self._session_ready:
119
            return
120
        await self._send({"type": "input.audio", "audio": base64.b64encode(frame.audio).decode()})
121
122
    async def _handle_context(self, frame: LLMContextFrame) -> None:
123
        for msg in frame.context.messages:
124
            if msg.get("role") == "system":
125
                content = msg.get("content", "")
126
                if isinstance(content, list):
127
                    content = " ".join(c.get("text", "") for c in content if isinstance(c, dict))
128
                if content:
129
                    await self._send({"type": "session.update", "session": {"system_prompt": content}})
130
                break
131
        await self.push_frame(frame, FrameDirection.DOWNSTREAM)
132
133
    async def _receive_task_handler(self) -> None:
134
        try:
135
            async for raw in self._websocket:
136
                try:
137
                    await self._handle_event(json.loads(raw))
138
                except Exception:
139
                    logger.exception("Error handling event")
140
        except websockets.exceptions.ConnectionClosed:
141
            logger.info("AssemblyAI Voice Agent API connection closed")
142
143
    async def _handle_event(self, event: dict) -> None:
144
        t = event.get("type", "")
145
        if t != "reply.audio":
146
            logger.debug(f"← {t}  {event}")
147
        dispatch = {
148
            "session.ready": self._on_session_ready,
149
            "input.speech.started": lambda _: self._on_speech_started(),
150
            "input.speech.stopped": lambda _: self._on_speech_stopped(),
151
            "transcript.user.delta": self._on_user_transcript_delta,
152
            "transcript.user": self._on_user_transcript,
153
            "reply.started": self._on_response_started,
154
            "reply.audio": self._on_response_audio,
155
            "transcript.agent": self._on_response_transcript,
156
            "reply.done": lambda _: self._on_response_done(),
157
            "tool.call": self._on_function_call,
158
            "session.error": self._on_error,
159
        }
160
        if handler := dispatch.get(t):
161
            await handler(event)
162
163
    async def _on_session_ready(self, event: dict) -> None:
164
        self._session_ready = True
165
        logger.info(f"Session ready | session_id={event.get('session_id', 'unknown')}")
166
        if self._pending_tools:
167
            await self._send({"type": "session.update", "session": {"tools": self._pending_tools}})
168
            self._pending_tools = []
169
170
    @staticmethod
171
    def _ts() -> str:
172
        return datetime.now(timezone.utc).isoformat()
173
174
    async def _on_speech_started(self) -> None:
175
        self._user_transcript_buf = ""
176
        await self.push_frame(UserStartedSpeakingFrame(), FrameDirection.UPSTREAM)
177
178
    async def _on_speech_stopped(self) -> None:
179
        await self.push_frame(UserStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
180
181
    async def _on_user_transcript_delta(self, event: dict) -> None:
182
        text = event.get("text", "")
183
        if text and text != self._user_transcript_buf:
184
            self._user_transcript_buf = text
185
            await self.push_frame(InterimTranscriptionFrame(text=text, user_id="user", timestamp=self._ts()), FrameDirection.UPSTREAM)
186
187
    async def _on_user_transcript(self, event: dict) -> None:
188
        text = event.get("text", "")
189
        self._user_transcript_buf = ""
190
        if text:
191
            logger.info(f"[User] {text}")
192
            await self.push_frame(TranscriptionFrame(text=text, user_id="user", timestamp=self._ts()), FrameDirection.UPSTREAM)
193
194
    async def _on_response_started(self, event: dict) -> None:
195
        self._current_response_id = event.get("reply_id", "")
196
        self._bot_speaking = True
197
        await self.push_frame(TTSStartedFrame())
198
        await self.push_frame(BotStartedSpeakingFrame())
199
200
    async def _on_response_audio(self, event: dict) -> None:
201
        data = base64.b64decode(event.get("data", ""))
202
        if data:
203
            await self.push_frame(TTSAudioRawFrame(audio=data, sample_rate=SAMPLE_RATE, num_channels=NUM_CHANNELS))
204
205
    async def _on_response_transcript(self, event: dict) -> None:
206
        if text := event.get("text", ""):
207
            logger.info(f"[Agent] {text}")
208
209
    async def _on_response_done(self) -> None:
210
        self._current_response_id = None
211
        self._bot_speaking = False
212
        await self.push_frame(TTSStoppedFrame())
213
        await self.push_frame(BotStoppedSpeakingFrame())
214
215
    async def _on_function_call(self, event: dict) -> None:
216
        call_id = event.get("call_id", "")
217
        name = event.get("name", "")
218
        raw = event.get("arguments", event.get("args", {}))
219
        args = raw if isinstance(raw, str) else json.dumps(raw)
220
        logger.info(f"Tool call: {name}({args})")
221
        result = ""
222
        if self.has_function(name):
223
            try:
224
                result = await self._execute_function(name, call_id, args)
225
            except Exception as e:
226
                result = f"Error: {e}"
227
        await self._send({"type": "tool.result", "call_id": call_id, "result": result or ""})
228
229
    async def _execute_function(self, name: str, call_id: str, arguments: str) -> str:
230
        from pipecat.adapters.schemas.direct_function import DirectFunctionWrapper
231
        from pipecat.processors.aggregators.llm_context import LLMContext
232
        from pipecat.services.llm_service import FunctionCallParams
233
        args = json.loads(arguments) if isinstance(arguments, str) else arguments
234
        result_holder: list = []
235
        async def result_callback(result, **_kwargs):
236
            result_holder.append(result)
237
        params = FunctionCallParams(function_name=name, tool_call_id=call_id, arguments=args, llm=self, context=LLMContext(), result_callback=result_callback)
238
        item = self._functions[name]
239
        if isinstance(item.handler, DirectFunctionWrapper):
240
            await item.handler.invoke(args=args, params=params)
241
        else:
242
            await item.handler(params)
243
        return str(result_holder[0]) if result_holder else ""
244
245
    async def _on_error(self, event: dict) -> None:
246
        msg = event.get("message") or str(event)
247
        logger.error(f"AssemblyAI Voice Agent API error: {msg}")
248
        await self.push_frame(ErrorFrame(msg))
249
250
    async def set_tools(self, tools: list[dict]) -> None:
251
        if self._session_ready:
252
            await self._send({"type": "session.update", "session": {"tools": tools}})
253
        else:
254
            self._pending_tools = tools

1
"""AssemblyAI Native Realtime Plugin — livekit.agents RealtimeModel/RealtimeSession."""
2
3
from __future__ import annotations
4
5
import asyncio
6
import base64
7
import json
8
import logging
9
import time
10
from dataclasses import dataclass
11
from typing import Literal
12
13
import aiohttp
14
from livekit import rtc
15
from livekit.agents import APIConnectionError, llm, utils
16
from livekit.agents.types import NOT_GIVEN, NotGivenOr
17
from livekit.agents.utils import is_given
18
19
logger = logging.getLogger(__name__)
20
21
SAMPLE_RATE = 24000
22
NUM_CHANNELS = 1
23
_SAMPLES_PER_CHUNK = SAMPLE_RATE // 10
24
25
26
@dataclass
27
class _Generation:
28
    response_id: str
29
    msg_ch: utils.aio.Chan[llm.MessageGeneration]
30
    fn_ch: utils.aio.Chan[llm.FunctionCall]
31
    text_ch: utils.aio.Chan[str]
32
    audio_ch: utils.aio.Chan[rtc.AudioFrame]
33
    modalities_fut: asyncio.Future[list[Literal["text", "audio"]]]
34
35
36
def _serialize_tool(tool: llm.Tool) -> dict | None:
37
    if isinstance(tool, llm.FunctionTool):
38
        return llm.utils.build_legacy_openai_schema(tool, internally_tagged=True)
39
    elif isinstance(tool, llm.RawFunctionTool):
40
        raw = dict(tool.info.raw_schema)
41
        raw.pop("meta", None)
42
        raw["type"] = "function"
43
        return raw
44
    return None
45
46
47
class _SessionExpiredError(Exception):
48
    pass
49
50
51
class RealtimeModel(llm.RealtimeModel):
52
    def __init__(self, *, url: str, api_key: str, sample_rate: int = SAMPLE_RATE, greeting: str | None = None) -> None:
53
        super().__init__(
54
            capabilities=llm.RealtimeCapabilities(
55
                turn_detection=True, user_transcription=True, audio_output=True,
56
                manual_function_calls=False, auto_tool_reply_generation=True, message_truncation=False,
57
            )
58
        )
59
        self._url = url
60
        self._api_key = api_key
61
        self._sample_rate = sample_rate
62
        self._greeting = greeting
63
        self._http_session: aiohttp.ClientSession | None = None
64
65
    def _ensure_http_session(self) -> aiohttp.ClientSession:
66
        if self._http_session is None or self._http_session.closed:
67
            self._http_session = aiohttp.ClientSession()
68
        return self._http_session
69
70
    def session(self) -> RealtimeSession:
71
        return RealtimeSession(self)
72
73
    async def aclose(self) -> None:
74
        if self._http_session and not self._http_session.closed:
75
            await self._http_session.close()
76
77
78
class RealtimeSession(llm.RealtimeSession[Literal[()]]):
79
    def __init__(self, realtime_model: RealtimeModel) -> None:
80
        super().__init__(realtime_model)
81
        self._model = realtime_model
82
        self._tools = llm.ToolContext.empty()
83
        self._chat_ctx = llm.ChatContext.empty()
84
        self._msg_ch: utils.aio.Chan[dict] = utils.aio.Chan()
85
        self._current_gen: _Generation | None = None
86
        self._pending_reply_fut: asyncio.Future[llm.GenerationCreatedEvent] | None = None
87
        self._current_response_id: str | None = None
88
        self._pending_call_ids: set[str] = set()
89
        self._session_ready: bool = False
90
        self._session_id: str | None = None
91
        self._bstream = utils.audio.AudioByteStream(SAMPLE_RATE, NUM_CHANNELS, samples_per_channel=_SAMPLES_PER_CHUNK)
92
        self._input_resampler: rtc.AudioResampler | None = None
93
        self._main_task = asyncio.create_task(self._run(), name="AssemblyAIRealtimeSession._run")
94
95
    @property
96
    def chat_ctx(self) -> llm.ChatContext:
97
        return self._chat_ctx
98
99
    @property
100
    def tools(self) -> llm.ToolContext:
101
        return self._tools.copy()
102
103
    def _send(self, msg: dict) -> None:
104
        try:
105
            self._msg_ch.send_nowait(msg)
106
        except Exception:
107
            pass
108
109
    def push_audio(self, frame: rtc.AudioFrame) -> None:
110
        for f in self._resample_audio(frame):
111
            for chunk in self._bstream.write(f.data.tobytes()):
112
                self._send({"type": "input.audio", "audio": base64.b64encode(chunk.data).decode()})
113
114
    def push_video(self, frame: rtc.VideoFrame) -> None:
115
        pass
116
117
    def generate_reply(self, *, instructions: NotGivenOr[str] = NOT_GIVEN) -> asyncio.Future[llm.GenerationCreatedEvent]:
118
        loop = asyncio.get_event_loop()
119
        fut: asyncio.Future[llm.GenerationCreatedEvent] = loop.create_future()
120
        self._pending_reply_fut = fut
121
        self._send({"type": "reply.create"})
122
        handle = loop.call_later(5.0, lambda: fut.done() or fut.set_exception(llm.RealtimeError("generate_reply timed out.")))
123
        fut.add_done_callback(lambda _: handle.cancel())
124
        return fut
125
126
    def interrupt(self) -> None:
127
        if self._current_response_id:
128
            self._send({"type": "reply.cancel", "reply_id": self._current_response_id})
129
130
    def commit_audio(self) -> None:
131
        pass
132
133
    def clear_audio(self) -> None:
134
        pass
135
136
    def truncate(self, *, message_id: str, modalities: list[Literal["text", "audio"]], audio_end_ms: int, audio_transcript: NotGivenOr[str] = NOT_GIVEN) -> None:
137
        pass
138
139
    async def update_instructions(self, instructions: str) -> None:
140
        payload: dict = {"system_prompt": instructions}
141
        if self._model._greeting:
142
            payload["greeting"] = self._model._greeting
143
        self._send({"type": "session.update", "session": payload})
144
145
    async def update_tools(self, tools: list[llm.Tool]) -> None:
146
        serialized = [s for t in tools if (s := _serialize_tool(t)) is not None]
147
        self._send({"type": "session.update", "session": {"tools": serialized}})
148
        self._tools = llm.ToolContext([t for t in tools if isinstance(t, (llm.FunctionTool, llm.RawFunctionTool))])
149
150
    def update_options(self, *, tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN) -> None:
151
        if is_given(tool_choice):
152
            self._send({"type": "session.update", "session": {"tool_choice": tool_choice}})
153
154
    async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
155
        if self._session_ready:
156
            existing_ids = {item.id for item in self._chat_ctx.items}
157
            for item in chat_ctx.items:
158
                if item.id in existing_ids:
159
                    continue
160
                if isinstance(item, llm.FunctionCallOutput) and item.call_id in self._pending_call_ids:
161
                    self._send({"type": "tool.result", "call_id": item.call_id, "result": item.output})
162
                    self._pending_call_ids.discard(item.call_id)
163
                elif isinstance(item, llm.ChatMessage) and item.role in ("user", "system") and item.text_content:
164
                    self._send({"type": "conversation.message", "role": item.role, "content": item.text_content})
165
        self._chat_ctx = chat_ctx
166
167
    async def aclose(self) -> None:
168
        self._msg_ch.close()
169
        await self._main_task
170
171
    @utils.log_exceptions(logger=logger)
172
    async def _run(self) -> None:
173
        for attempt in range(4):
174
            try:
175
                headers = {"Authorization": f"Bearer {self._model._api_key}"}
176
                ws = await self._model._ensure_http_session().ws_connect(self._model._url, headers=headers)
177
                if self._session_id is not None:
178
                    await ws.send_str(json.dumps({"type": "session.resume", "session_id": self._session_id}))
179
                if self._model._greeting:
180
                    await ws.send_str(json.dumps({"type": "session.update", "session": {"greeting": self._model._greeting}}))
181
                self._session_ready = False
182
                self._current_response_id = None
183
                self._close_current_gen()
184
                if self._pending_reply_fut:
185
                    self._pending_reply_fut.cancel()
186
                    self._pending_reply_fut = None
187
                self._bstream = utils.audio.AudioByteStream(SAMPLE_RATE, NUM_CHANNELS, samples_per_channel=_SAMPLES_PER_CHUNK)
188
                closing = False
189
190
                @utils.log_exceptions(logger=logger)
191
                async def _send_task() -> None:
192
                    nonlocal closing
193
                    async for msg in self._msg_ch:
194
                        try:
195
                            await ws.send_str(json.dumps(msg))
196
                        except Exception:
197
                            logger.exception("failed to send event")
198
                    closing = True
199
                    await ws.close()
200
201
                @utils.log_exceptions(logger=logger)
202
                async def _recv_task() -> None:
203
                    while True:
204
                        msg = await ws.receive()
205
                        if msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING):
206
                            if closing:
207
                                return
208
                            raise APIConnectionError("AssemblyAI Realtime connection closed unexpectedly")
209
                        if msg.type != aiohttp.WSMsgType.TEXT:
210
                            continue
211
                        try:
212
                            self._handle_event(json.loads(msg.data))
213
                        except _SessionExpiredError:
214
                            self._session_id = None
215
                            self._pending_call_ids.clear()
216
                            raise APIConnectionError("Session expired; retrying as fresh session")
217
                        except Exception:
218
                            logger.exception("failed to handle event")
219
220
                tasks = [asyncio.create_task(_recv_task()), asyncio.create_task(_send_task())]
221
                try:
222
                    done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
223
                    for task in done:
224
                        task.result()
225
                    return
226
                finally:
227
                    await utils.aio.cancel_and_wait(*tasks)
228
                    await ws.close()
229
            except APIConnectionError:
230
                if attempt >= 3:
231
                    raise
232
                await asyncio.sleep([0.5, 1.0, 2.0][attempt])
233
234
    def _handle_event(self, event: dict) -> None:
235
        self.emit(f"aai.{event.get('type', '')}", event)
236
        t = event.get("type", "")
237
        if t == "session.ready":
238
            self._session_ready = True
239
            self._session_id = event.get("session_id")
240
            if self._pending_reply_fut and not self._pending_reply_fut.done():
241
                self._send({"type": "reply.create"})
242
        elif t == "input.speech.started":
243
            self.emit("input_speech_started", llm.InputSpeechStartedEvent())
244
        elif t == "input.speech.stopped":
245
            self.emit("input_speech_stopped", llm.InputSpeechStoppedEvent(user_transcription_enabled=True))
246
        elif t == "transcript.user":
247
            self.emit("input_audio_transcription_completed", llm.InputTranscriptionCompleted(item_id=event.get("item_id", ""), transcript=event.get("text", ""), is_final=True))
248
        elif t == "reply.started":
249
            self._close_current_gen()
250
            response_id = event.get("reply_id", "")
251
            self._current_response_id = response_id
252
            text_ch: utils.aio.Chan[str] = utils.aio.Chan()
253
            audio_ch: utils.aio.Chan[rtc.AudioFrame] = utils.aio.Chan()
254
            modalities_fut = asyncio.get_event_loop().create_future()
255
            modalities_fut.set_result(["audio"])
256
            msg_gen = llm.MessageGeneration(message_id=response_id, text_stream=text_ch, audio_stream=audio_ch, modalities=modalities_fut)
257
            msg_ch: utils.aio.Chan[llm.MessageGeneration] = utils.aio.Chan()
258
            fn_ch: utils.aio.Chan[llm.FunctionCall] = utils.aio.Chan()
259
            self._current_gen = _Generation(response_id=response_id, msg_ch=msg_ch, fn_ch=fn_ch, text_ch=text_ch, audio_ch=audio_ch, modalities_fut=modalities_fut)
260
            msg_ch.send_nowait(msg_gen)
261
            gen_ev = llm.GenerationCreatedEvent(message_stream=msg_ch, function_stream=fn_ch, user_initiated=False, response_id=response_id)
262
            if self._pending_reply_fut and not self._pending_reply_fut.done():
263
                gen_ev.user_initiated = True
264
                self._pending_reply_fut.set_result(gen_ev)
265
                self._pending_reply_fut = None
266
            self.emit("generation_created", gen_ev)
267
        elif t == "reply.audio":
268
            if gen := self._current_gen:
269
                if data := base64.b64decode(event.get("data", "")):
270
                    gen.audio_ch.send_nowait(rtc.AudioFrame(data=data, sample_rate=SAMPLE_RATE, num_channels=NUM_CHANNELS, samples_per_channel=len(data) // 2))
271
        elif t == "transcript.agent":
272
            if gen := self._current_gen:
273
                if text := event.get("text", ""):
274
                    gen.text_ch.send_nowait(text)
275
                gen.text_ch.close()
276
        elif t == "reply.done":
277
            self._close_current_gen()
278
            self._current_response_id = None
279
        elif t == "tool.call":
280
            if gen := self._current_gen:
281
                call_id = event.get("call_id", "")
282
                if call_id not in self._pending_call_ids:
283
                    self._pending_call_ids.add(call_id)
284
                    gen.fn_ch.send_nowait(llm.FunctionCall(call_id=call_id, name=event.get("name", ""), arguments=json.dumps(event.get("args", {}))))
285
        elif t == "session.error":
286
            code = event.get("code", "")
287
            if code in ("session_not_found", "session_forbidden"):
288
                raise _SessionExpiredError(event.get("message", ""))
289
            self.emit("error", llm.RealtimeModelError(timestamp=time.time(), label=self._model.label, error=Exception(event.get("message", "unknown error")), recoverable=False))
290
291
    def _close_current_gen(self) -> None:
292
        if gen := self._current_gen:
293
            for ch in (gen.audio_ch, gen.text_ch, gen.fn_ch, gen.msg_ch):
294
                ch.close()
295
            self._current_gen = None
296
297
    def _resample_audio(self, frame: rtc.AudioFrame):
298
        if self._input_resampler is not None and frame.sample_rate != self._input_resampler._input_rate:
299
            self._input_resampler = None
300
        if self._input_resampler is None and (frame.sample_rate != SAMPLE_RATE or frame.num_channels != NUM_CHANNELS):
301
            self._input_resampler = rtc.AudioResampler(input_rate=frame.sample_rate, output_rate=SAMPLE_RATE, num_channels=NUM_CHANNELS)
302
        if self._input_resampler:
303
            yield from self._input_resampler.push(frame)
304
        else:
305
            yield frame