LiveKit and Murf

1
from livekit.plugins import murf
2
3
# Add Murf TTS to your existing session
4
session = AgentSession(
5
    stt="your-stt-provider",  # e.g., "deepgram/nova-3"
6
    llm="your-llm-provider",  # e.g., "openai/gpt-4o"
7
    tts=murf.TTS(voice="Matthew", style="Conversation", model="Falcon")
8
    # ... your existing configuration
9
)

1
from dotenv import load_dotenv
2
load_dotenv()

1
import logging
2
import os
3
from dotenv import load_dotenv
4
5
load_dotenv()
6
7
from livekit.agents import (
8
    Agent,
9
    AgentServer,
10
    AgentSession,
11
    JobContext,
12
    JobProcess,
13
    MetricsCollectedEvent,
14
    cli,
15
    metrics,
16
    room_io,
17
)
18
from livekit.plugins import silero, deepgram, murf
19
from livekit.plugins import openai as openai_plugin
20
21
logger = logging.getLogger("murf-agent")
22
23
class MyAgent(Agent):
24
    def __init__(self) -> None:
25
        super().__init__(
26
            instructions="You are a voice agent built using Murf TTS. Keep responses short and natural."
27
        )
28
29
    async def on_enter(self):
30
        await self.session.say(
31
            "Hi, I am a voice agent powered by Murf, how can I help you?"
32
        )
33
34
server = AgentServer()
35
36
def prewarm(proc: JobProcess):
37
    proc.userdata["vad"] = silero.VAD.load()
38
39
server.setup_fnc = prewarm
40
41
@server.rtc_session()
42
async def entrypoint(ctx: JobContext):
43
    ctx.log_context_fields = {"room": ctx.room.name}
44
45
    session = AgentSession(
46
        stt="deepgram/nova-3",
47
        llm="openai/gpt-4o",
48
        tts=murf.TTS(voice="Matthew", style="Conversation"),
49
        vad=ctx.proc.userdata["vad"],
50
        preemptive_generation=True,
51
        resume_false_interruption=True,
52
        false_interruption_timeout=1.0,
53
    )
54
55
    usage_collector = metrics.UsageCollector()
56
57
    @session.on("metrics_collected")
58
    def on_metrics(ev: MetricsCollectedEvent):
59
        metrics.log_metrics(ev.metrics)
60
        usage_collector.collect(ev.metrics)
61
62
    async def log_usage():
63
        logger.info(f"Usage: {usage_collector.get_summary()}")
64
65
    ctx.add_shutdown_callback(log_usage)
66
67
    await session.start(
68
        agent=MyAgent(),
69
        room=ctx.room,
70
        room_options=room_io.RoomOptions(
71
            audio_input=room_io.AudioInputOptions()
72
        ),
73
    )
74
75
if __name__ == "__main__":
76
    cli.run_app(server)

1
import logging
2
import os
3
from dotenv import load_dotenv
4
5
load_dotenv()
6
7
from livekit.agents import (
8
    Agent,
9
    AgentServer,
10
    AgentSession,
11
    JobContext,
12
    JobProcess,
13
    MetricsCollectedEvent,
14
    cli,
15
    metrics,
16
    room_io,
17
)
18
from livekit.plugins import silero, deepgram, murf
19
from livekit.plugins import openai as openai_plugin
20
21
logger = logging.getLogger("murf-agent")
22
23
class CustomAgent(Agent):
24
    def __init__(self) -> None:
25
        super().__init__(
26
            instructions="You are a helpful assistant with a natural speaking style. Provide detailed but concise responses."
27
        )
28
29
    async def on_enter(self):
30
        await self.session.say(
31
            "Hello! I'm powered by Murf's high-quality voice synthesis. How can I assist you today?"
32
        )
33
34
server = AgentServer()
35
36
def prewarm(proc: JobProcess):
37
    proc.userdata["vad"] = silero.VAD.load()
38
39
server.setup_fnc = prewarm
40
41
@server.rtc_session()
42
async def entrypoint(ctx: JobContext):
43
    ctx.log_context_fields = {"room": ctx.room.name}
44
45
    session = AgentSession(
46
        stt="deepgram/nova-3",
47
        llm="openai/gpt-4o",
48
        tts=murf.TTS(
49
            voice="Matthew",
50
            style="Conversation",
51
            speed=5,
52
            pitch=0,
53
            model="FALCON",
54
            sample_rate=24000,
55
            locale="en-US",
56
        ),
57
        vad=ctx.proc.userdata["vad"],
58
        preemptive_generation=True,
59
        resume_false_interruption=True,
60
        false_interruption_timeout=1.0,
61
    )
62
63
    usage_collector = metrics.UsageCollector()
64
65
    @session.on("metrics_collected")
66
    def on_metrics(ev: MetricsCollectedEvent):
67
        metrics.log_metrics(ev.metrics)
68
        usage_collector.collect(ev.metrics)
69
70
    async def log_usage():
71
        summary = usage_collector.get_summary()
72
        logger.info(f"Session usage: {summary}")
73
74
    ctx.add_shutdown_callback(log_usage)
75
76
    await session.start(
77
        agent=CustomAgent(),
78
        room=ctx.room,
79
        room_options=room_io.RoomOptions(
80
            audio_input=room_io.AudioInputOptions()
81
        ),
82
    )
83
84
if __name__ == "__main__":
85
    cli.run_app(server)