How to Clone Any Voice in Minutes Using Voxtral TTS

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
"""
Voxtral TTS Example

This example demonstrates Voxtral TTS (Mistral) integration with Vision Agents.

This example creates an agent that uses:
- Voxtral TTS for text-to-speech (voxtral-mini-tts-2603)
- Deepgram for speech-to-text
- Gemini for LLM
- GetStream for edge/real-time communication
- Smart Turn for turn detection

Requirements:
- MISTRAL_API_KEY environment variable
- DEEPGRAM_API_KEY environment variable
- GOOGLE_API_KEY environment variable
- STREAM_API_KEY and STREAM_API_SECRET environment variables
"""

import asyncio
import base64
import logging
from pathlib import Path

from dotenv import load_dotenv
from vision_agents.core import Agent, Runner, User
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import deepgram, gemini, getstream, smart_turn

import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from vision_agents.plugins.voxtral import TTS as VoxtralTTS

logger = logging.getLogger(__name__)

load_dotenv()

async def create_agent(**kwargs) -> Agent:
    """Create the agent with Voxtral TTS."""
    ref_audio_path = Path(__file__).resolve().parents[3] / "david.wav"
    ref_audio_b64 = base64.b64encode(ref_audio_path.read_bytes()).decode()

    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Voxtral AI", id="agent"),
        instructions=(
            "You are a friendly multilingual voice assistant powered by "
            "Voxtral TTS. Keep your responses short and conversational. "
            "You can speak in English, French, Spanish, Portuguese, Italian, "
            "Dutch, German, Hindi, and Arabic."
        ),
        tts=VoxtralTTS(
            model="voxtral-mini-tts-2603",
            ref_audio=ref_audio_b64,
            response_format="pcm",
        ),
        stt=deepgram.STT(eager_turn_detection=True),
        llm=gemini.LLM("gemini-3-flash-preview"),
        turn_detection=smart_turn.TurnDetection(
            silence_duration_ms=2000,
            speech_probability_threshold=0.5,
        ),
    )
    return agent

async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    """Join the call and start the agent."""
    call = await agent.create_call(call_type, call_id)

    logger.info("Starting Voxtral TTS Agent...")

    async with agent.join(call):
        logger.info("Agent joined call")

        await asyncio.sleep(3)
        await agent.llm.simple_response(
            text="Hello! I'm powered by Voxtral TTS from Mistral."
        )

        await agent.finish()

if __name__ == "__main__":
    Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli()

1
2
3
4
5
     tts=VoxtralTTS(
            model="voxtral-mini-tts-2603",
            ref_audio=ref_audio_b64,
            response_format="pcm",
        ),

1
2
 ref_audio_path = Path(__file__).resolve().parents[3] / "david.wav"
 ref_audio_b64 = base64.b64encode(ref_audio_path.read_bytes()).decode()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
"""
Voxtral TTS Voice Cloning Example

Requirements:
- MISTRAL_API_KEY environment variable
- DEEPGRAM_API_KEY environment variable
- GOOGLE_API_KEY environment variable
- STREAM_API_KEY and STREAM_API_SECRET environment variables
- A reference audio file (david.wav) for voice cloning
"""

import asyncio
import base64
import logging
import os
import sys
from pathlib import Path

from dotenv import load_dotenv
from mistralai.client import Mistral
from vision_agents.core import Agent, Runner, User
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import deepgram, gemini, getstream

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from vision_agents.plugins.voxtral import TTS as VoxtralTTS

logger = logging.getLogger(__name__)

load_dotenv()

PROJECT_ROOT = Path(__file__).resolve().parents[3]
REF_AUDIO_PATH = PROJECT_ROOT / "david.wav"

def create_saved_voice(audio_path: Path = REF_AUDIO_PATH) -> str:
    """
    Create a saved voice via the Mistral Voices API.

    Once created, the voice can be reused across requests by its ID,
    avoiding the need to pass ref_audio every time.

    Args:
        audio_path: Path to audio file (3-25s, single speaker, clean WAV/MP3).

    Returns:
        The voice ID string.
    """
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    sample_audio_b64 = base64.b64encode(audio_path.read_bytes()).decode()

    voice = client.audio.voices.create(
        name="david-cloned-voice",
        sample_audio=sample_audio_b64,
        sample_filename=audio_path.name,
        languages=["en"],
    )

    logger.info("Created voice: %s (id=%s)", voice.name, voice.id)
    return voice.id

async def create_agent_with_ref_audio(**kwargs) -> Agent:
    """
    Create an agent using on-the-fly voice cloning via ref_audio.

    Pass a base64-encoded audio clip directly to clone a voice without
    creating a saved voice first. Best for one-off use or experimentation.
    """
    ref_audio_b64 = base64.b64encode(REF_AUDIO_PATH.read_bytes()).decode()

    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Cloned Voice AI", id="agent"),
        instructions=(
            "You are a voice assistant that sounds like the person in the "
            "reference audio. Keep responses natural and conversational."
        ),
        tts=VoxtralTTS(
            model="voxtral-mini-tts-2603",
            ref_audio=ref_audio_b64,
            response_format="pcm",
        ),
        stt=deepgram.STT(eager_turn_detection=True),
        llm=gemini.LLM("gemini-3-flash-preview"),
    )
    return agent

async def create_agent_with_voice_id(**kwargs) -> Agent:
    """
    Create an agent using a previously saved voice ID.

    Saved voices provide consistent results and avoid sending the
    reference audio with every request. Set VOXTRAL_VOICE_ID in .env
    or pass voice_id in kwargs.
    """
    voice_id = kwargs.get("voice_id") or os.environ.get("VOXTRAL_VOICE_ID")
    if not voice_id:
        logger.info("No voice_id provided, creating a saved voice from %s...", REF_AUDIO_PATH.name)
        voice_id = create_saved_voice(REF_AUDIO_PATH)

    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Cloned Voice AI", id="agent"),
        instructions=(
            "You are a voice assistant with a custom cloned voice. "
            "Keep responses short, friendly, and natural."
        ),
        tts=VoxtralTTS(
            model="voxtral-mini-tts-2603",
            voice_id=voice_id,
            response_format="pcm",
        ),
        stt=deepgram.STT(eager_turn_detection=True),
        llm=gemini.LLM("gemini-3-flash-preview"),
    )
    return agent

async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    """Join the call and start the agent."""
    call = await agent.create_call(call_type, call_id)

    logger.info("Starting Voxtral Voice Cloning Agent...")

    async with agent.join(call):
        logger.info("Agent joined call with cloned voice")

        await asyncio.sleep(3)
        await agent.llm.simple_response(
            text="Hello! I'm speaking with a cloned voice powered by Voxtral TTS."
        )

        await agent.finish()

if __name__ == "__main__":
    Runner(
        AgentLauncher(
            create_agent=create_agent_with_ref_audio,
            join_call=join_call,
        )
    ).cli()