Add LiveKit audio transport

[joachimchauvet]

This might not be perfect yet but I would love to get some initial feedback.
It looks like there are quite a few people interested #325

[aconchillo]

We probably want to add livekit to pyproject.toml with the right dependencies.

[aconchillo]

This looks really good! I'll take a final look later, thank you!!!

[cyrilS-dev]

This looks amazing @joachimchauvet ! I'm particularly interested in this PR as I'm using LiveKit.
I will give it a try as soon as possible and provide feedback. Looking forward to testing it out!
Many thanks

[cyrilS-dev]

@joachimchauvet I just tested and I can't play the audio when I join the room using the LK playground. I join the same room as my agent, I have both participants active in the room, the room events are consistent, but the 'Agent connected' status on the playground remains pending, as does the audio track.

[joachimchauvet]

@joachimchauvet I just tested and I can't play the audio when I join the room using the LK playground. I join the same room as my agent, I have both participants active in the room, the room events are consistent, but the 'Agent connected' status on the playground remains pending, as does the audio track.

It seems to work fine on my side.
However, I noticed that I forgot to adjust the sample rate in https://github.com/joachimchauvet/pipecat-livekit/blob/main/examples/foundational/01b-livekit-audio.py
Could it be that the voice was playing so fast that you did not really hear it if you were using that example?
Are you using a different token for the bot and the user?

[aconchillo]

This looks great! Please, rebase and make sure the pipeline is green. We are now using Ruff as our formatter. Make sure everything is green and I think we cna merge after that. Thank you! 🙏

[cyrilS-dev]

@joachimchauvet I encountered an error while using the LiveKitInputTransport in my application :

Error in audio input task: 'LiveKitInputTransport' object has no attribute '_internal_push_frame'

_internal_push_frame method is not defined in the LiveKitInputTransport class

[joachimchauvet]

@joachimchauvet I encountered an error while using the LiveKitInputTransport in my application :

Error in audio input task: 'LiveKitInputTransport' object has no attribute '_internal_push_frame'

_internal_push_frame method is not defined in the LiveKitInputTransport class

Were you doing something special or made changes to the LiveKitInputTransport? It inherits the BaseInputTransport so _internal_push_frame should be defined.
The method works fine on my side 🤔

[cyrilS-dev]

@joachimchauvet I encountered an error while using the LiveKitInputTransport in my application :
Error in audio input task: 'LiveKitInputTransport' object has no attribute '_internal_push_frame'
_internal_push_frame method is not defined in the LiveKitInputTransport class

Were you doing something special or made changes to the LiveKitInputTransport? It inherits the BaseInputTransport so _internal_push_frame should be defined. The method works fine on my side 🤔

There is no _internal_push_frame method into BaseInputTransport since #436

[cyrilS-dev]

@joachimchauvet I encountered an error in the _audio_in_task_handler method of the LiveKitInputTransport class., using Deepgram as STT. :

AttributeError: 'AsyncListenWebSocketClient' object has no attribute '_socket'
2024-09-25 15:49:15.863 | ERROR    | pipecat.processors.frame_processor:push_frame:203 - Uncaught exception in LiveKitInputTransport#0: 'AsyncListenWebSocketClient' object has no attribute '_socket'

The issue arises when await self.push_frame(pipecat_audio_frame) attempts to push frames before the Deepgram WebSocket connection is fully initialized.

[aconchillo]

@joachimchauvet I encountered an error in the _audio_in_task_handler method of the LiveKitInputTransport class., using Deepgram as STT. :

AttributeError: 'AsyncListenWebSocketClient' object has no attribute '_socket'
2024-09-25 15:49:15.863 | ERROR    | pipecat.processors.frame_processor:push_frame:203 - Uncaught exception in LiveKitInputTransport#0: 'AsyncListenWebSocketClient' object has no attribute '_socket'

The issue arises when await self.push_frame(pipecat_audio_frame) attempts to push frames before the Deepgram WebSocket connection is fully initialized.

In theory, all processors should receive the StartFrame first (this is guaranteed now in main) and they should make sure any connection or whatever is needed is established so it's safe to push frames after.

It's possible that this branch hasn't been rebased so it doesn't have that StartFrame guarantee. Would it be possible for you to rebase locally and try that? I could be wrong and maybe it's something else...

[cyrilS-dev]

@joachimchauvet I encountered an error in the _audio_in_task_handler method of the LiveKitInputTransport class., using Deepgram as STT. :

AttributeError: 'AsyncListenWebSocketClient' object has no attribute '_socket'
2024-09-25 15:49:15.863 | ERROR    | pipecat.processors.frame_processor:push_frame:203 - Uncaught exception in LiveKitInputTransport#0: 'AsyncListenWebSocketClient' object has no attribute '_socket'

The issue arises when await self.push_frame(pipecat_audio_frame) attempts to push frames before the Deepgram WebSocket connection is fully initialized.

In theory, all processors should receive the StartFrame first (this is guaranteed now in main) and they should make sure any connection or whatever is needed is established so it's safe to push frames after.

It's possible that this branch hasn't been rebased so it doesn't have that StartFrame guarantee. Would it be possible for you to rebase locally and try that? I could be wrong and maybe it's something else...

You're absolutely right @aconchillo. Thanks for pointing that out!

Now, regarding this part:

async def send_metrics(self, frame: MetricsFrame):
        metrics = {}
        if frame.ttfb:
            metrics["ttfb"] = frame.ttfb
        if frame.processing:
            metrics["processing"] = frame.processing
        if hasattr(frame, "tokens"):
            metrics["tokens"] = frame.tokens
        if hasattr(frame, "characters"):
            metrics["characters"] = frame.characters

        message = LiveKitTransportMessageFrame(message={"type": "pipecat-metrics", "metrics": metrics})
        await self._client.send_data(str(message.message).encode())

This needs to be updated due to changes introduced in #474

[cyrilS-dev]

Everything is running perfectly; just one detail: LiveKitInputTransport stops when the room is disconnected, but LiveKitOutputTransport doesn't; this triggers an error after a while from the TTS service: no close frame received or sent.

[aconchillo]

Everything is running perfectly; just one detail: LiveKitInputTransport stops when the room is disconnected, but LiveKitOutputTransport doesn't; this triggers an error after a while from the TTS service: no close frame received or sent.

I'll wait on @joachimchauvet to answer this before merging.

[aconchillo]

I'll go ahead and merge this one. If there's any error you can provide a fix later. Thank you!!!

[joachimchauvet]

Everything is running perfectly; just one detail: LiveKitInputTransport stops when the room is disconnected, but LiveKitOutputTransport doesn't; this triggers an error after a while from the TTS service: no close frame received or sent.

I don't have this error with my agents. I have an idea where that might come from but I was not able to reproduce that no close frame received or sent directly. @cyrilS-dev could you share a pipeline that triggers this error?

[cyrilS-dev]

@joachimchauvet The execution is blocked when calling await super().stop(frame) in the stop method of the LiveKitOutputTransport class.

The stop method in the BaseOutputTransport class is causing the execution to hang due to :

if self._sink_task:
     await self._sink_task
if self._sink_clock_task:
     await self._sink_clock_task

As a result, the await self._client.disconnect() call is never reached.

@aconchillo do you have any insights on this? Thanks