Installation
pip install cartesia netra-sdk
npm install @cartesia/cartesia-js netra-sdk
Usage
Initialize Netra before using Cartesia:import os
from netra import Netra
Netra.init(
app_name="cartesia-service",
headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}"
)
import Netra from 'netra-sdk';
await Netra.init({
appName: 'cartesia-service',
headers: `x-api-key=${process.env.NETRA_API_KEY}`
});
Examples
Text-to-Speech with Sonic Models
Track Cartesia TTS operations using Netra decorators:from cartesia import Cartesia
from netra.decorators import task, workflow
import os
client = Cartesia(api_key=os.environ.get("CARTESIA_API_KEY"))
@task()
def generate_speech(text: str, voice_id: str) -> bytes:
"""Generate speech using Cartesia Sonic model."""
response = client.tts.bytes(
model_id="sonic-english",
transcript=text,
voice={
"mode": "id",
"id": voice_id
},
output_format={
"container": "raw",
"encoding": "pcm_f32le",
"sample_rate": 44100
}
)
return response
@task()
def stream_speech(text: str, voice_id: str):
"""Stream speech using Cartesia Sonic Turbo."""
stream = client.tts.stream(
model_id="sonic-turbo",
transcript=text,
voice={
"mode": "id",
"id": voice_id
},
output_format={
"container": "raw",
"encoding": "pcm_f32le",
"sample_rate": 44100
}
)
for chunk in stream:
yield chunk
@workflow()
def process_text_batch(texts: list[str], voice_id: str) -> list[bytes]:
"""Process multiple texts to speech."""
audio_buffers = []
for text in texts:
audio = generate_speech(text, voice_id)
audio_buffers.append(audio)
return audio_buffers
# Usage
audio_data = generate_speech(
"Hello, this is Cartesia Sonic speech synthesis.",
"a0e99841-438c-4a64-b679-ae501e7d6091"
)
import Cartesia from '@cartesia/cartesia-js';
import { task, workflow } from 'netra-sdk';
const cartesia = new Cartesia({
apiKey: process.env.CARTESIA_API_KEY
});
class TTSService {
@task()
async generateSpeech(text: string, voiceId: string): Promise<Buffer> {
const response = await cartesia.tts.bytes({
model_id: 'sonic-english',
transcript: text,
voice: {
mode: 'id',
id: voiceId
},
output_format: {
container: 'raw',
encoding: 'pcm_f32le',
sample_rate: 44100
}
});
return Buffer.from(response);
}
@task()
async streamSpeech(text: string, voiceId: string): Promise<AsyncGenerator<Buffer>> {
const stream = await cartesia.tts.stream({
model_id: 'sonic-turbo',
transcript: text,
voice: {
mode: 'id',
id: voiceId
},
output_format: {
container: 'raw',
encoding: 'pcm_f32le',
sample_rate: 44100
}
});
async function* generateChunks() {
for await (const chunk of stream) {
yield Buffer.from(chunk);
}
}
return generateChunks();
}
@workflow()
async processTextBatch(texts: string[], voiceId: string): Promise<Buffer[]> {
const audioBuffers: Buffer[] = [];
for (const text of texts) {
const audio = await this.generateSpeech(text, voiceId);
audioBuffers.push(audio);
}
return audioBuffers;
}
}
// Usage
const service = new TTSService();
const audio = await service.generateSpeech(
'Hello, this is Cartesia Sonic speech synthesis.',
'a0e99841-438c-4a64-b679-ae501e7d6091'
);
Speech-to-Text with Ink Models
Track Cartesia STT operations using Netra decorators:from cartesia import Cartesia
from netra.decorators import task
import os
client = Cartesia(api_key=os.environ.get("CARTESIA_API_KEY"))
@task()
def transcribe_audio(audio_path: str) -> str:
"""Transcribe audio using Cartesia Ink model."""
with open(audio_path, "rb") as audio_file:
audio_data = audio_file.read()
response = client.stt.transcribe(
model_id="ink-whisper",
audio=audio_data,
language="en"
)
return response.transcript
@task()
def transcribe_stream(audio_stream) -> str:
"""Transcribe streaming audio in real-time."""
full_transcript = ""
stream = client.stt.stream_transcribe(
model_id="ink-whisper",
audio=audio_stream,
language="en",
interim_results=True
)
for result in stream:
if result.is_final:
full_transcript += result.transcript + " "
return full_transcript.strip()
# Usage
transcript = transcribe_audio("./audio/sample.wav")
import Cartesia from '@cartesia/cartesia-js';
import { task } from 'netra-sdk';
import * as fs from 'fs';
const cartesia = new Cartesia({
apiKey: process.env.CARTESIA_API_KEY
});
class STTService {
@task()
async transcribeAudio(audioPath: string): Promise<string> {
const audioBuffer = fs.readFileSync(audioPath);
const response = await cartesia.stt.transcribe({
model_id: 'ink-whisper',
audio: audioBuffer,
language: 'en'
});
return response.transcript;
}
@task()
async transcribeStream(audioStream: ReadableStream): Promise<string> {
let fullTranscript = '';
const stream = await cartesia.stt.streamTranscribe({
model_id: 'ink-whisper',
audio: audioStream,
language: 'en',
interim_results: true
});
for await (const result of stream) {
if (result.is_final) {
fullTranscript += result.transcript + ' ';
}
}
return fullTranscript.trim();
}
}
// Usage
const sttService = new STTService();
const transcript = await sttService.transcribeAudio('./audio/sample.wav');
Manual Span Creation with Action Tracking
For detailed control over tracing with both TTS and STT:from cartesia import Cartesia
from netra import SpanWrapper, ActionModel, UsageModel
import os
import time
client = Cartesia(api_key=os.environ.get("CARTESIA_API_KEY"))
def generate_speech_with_tracking(text: str, voice_id: str) -> bytes:
"""Generate speech with detailed tracking."""
span = SpanWrapper("cartesia-tts")
span.start()
try:
start_time = time.time_ns()
span.set_attribute("text_length", len(text))
span.set_attribute("voice_id", voice_id)
span.set_attribute("model", "sonic-turbo")
response = client.tts.bytes(
model_id="sonic-turbo",
transcript=text,
voice={"mode": "id", "id": voice_id},
output_format={
"container": "raw",
"encoding": "pcm_f32le",
"sample_rate": 44100
}
)
end_time = time.time_ns()
duration_ms = (end_time - start_time) / 1_000_000
# Track the TTS API operation
action = ActionModel(
start_time=str(start_time),
action="API",
action_type="TTS_SYNTHESIS",
metadata={
"provider": "cartesia",
"model": "sonic-turbo",
"voice_id": voice_id,
"text_length": str(len(text)),
"audio_size_bytes": str(len(response)),
"sample_rate": "44100",
"latency_ms": str(duration_ms)
},
success=True
)
span.set_action([action])
# Track usage
usage = UsageModel(
model="sonic-turbo",
usage_type="characters",
units_used=len(text),
cost_in_usd=len(text) * 0.00001
)
span.set_usage([usage])
span.set_status({"code": 1, "message": "Success"})
span.end()
return response
except Exception as e:
span.set_error(e)
span.set_status({"code": 2, "message": "Error"})
span.end()
raise
def transcribe_with_tracking(audio_path: str) -> str:
"""Transcribe audio with detailed tracking."""
span = SpanWrapper("cartesia-stt")
span.start()
try:
start_time = time.time_ns()
with open(audio_path, "rb") as audio_file:
audio_data = audio_file.read()
audio_size_bytes = len(audio_data)
span.set_attribute("audio_file", audio_path)
span.set_attribute("audio_size_bytes", audio_size_bytes)
span.set_attribute("model", "ink-whisper")
response = client.stt.transcribe(
model_id="ink-whisper",
audio=audio_data,
language="en"
)
end_time = time.time_ns()
duration_ms = (end_time - start_time) / 1_000_000
# Track the STT API operation
action = ActionModel(
start_time=str(start_time),
action="API",
action_type="STT_TRANSCRIPTION",
metadata={
"provider": "cartesia",
"model": "ink-whisper",
"audio_size_bytes": str(audio_size_bytes),
"transcript_length": str(len(response.transcript)),
"duration_ms": str(duration_ms),
"language": "en"
},
success=True
)
span.set_action([action])
# Track usage
audio_duration = getattr(response, 'duration', 0)
usage = UsageModel(
model="ink-whisper",
usage_type="audio_seconds",
units_used=audio_duration,
cost_in_usd=audio_duration * 0.036 # $0.13 per hour
)
span.set_usage([usage])
span.set_attribute("transcript_length", len(response.transcript))
span.set_status({"code": 1, "message": "Success"})
span.end()
return response.transcript
except Exception as e:
span.set_error(e)
span.set_status({"code": 2, "message": "Error"})
span.end()
raise
# Usage
audio_data = generate_speech_with_tracking(
"This is ultra-low latency speech synthesis.",
"a0e99841-438c-4a64-b679-ae501e7d6091"
)
transcript = transcribe_with_tracking("./audio/sample.wav")
import Cartesia from '@cartesia/cartesia-js';
import { SpanWrapper, ActionModel } from 'netra-sdk';
const cartesia = new Cartesia({
apiKey: process.env.CARTESIA_API_KEY
});
async function generateSpeechWithTracking(text: string, voiceId: string): Promise<Buffer> {
const span = new SpanWrapper('cartesia-tts');
span.start();
try {
const startTime = Date.now();
span.setAttribute('text_length', text.length);
span.setAttribute('voice_id', voiceId);
span.setAttribute('model', 'sonic-turbo');
const response = await cartesia.tts.bytes({
model_id: 'sonic-turbo',
transcript: text,
voice: { mode: 'id', id: voiceId },
output_format: {
container: 'raw',
encoding: 'pcm_f32le',
sample_rate: 44100
}
});
const audioBuffer = Buffer.from(response);
const duration = Date.now() - startTime;
// Track the TTS API operation
const action: ActionModel = {
start_time: (startTime * 1000000).toString(),
action: 'API',
action_type: 'TTS_SYNTHESIS',
metadata: {
provider: 'cartesia',
model: 'sonic-turbo',
voice_id: voiceId,
text_length: text.length.toString(),
audio_size_bytes: audioBuffer.length.toString(),
sample_rate: '44100',
latency_ms: duration.toString()
},
success: true
};
span.setAction([action]);
span.setUsage({
model: 'sonic-turbo',
usage_type: 'characters',
units_used: text.length,
cost_in_usd: text.length * 0.00001
});
span.setStatus({ code: 1, message: 'Success' });
span.end();
return audioBuffer;
} catch (error) {
span.setError(error as Error);
span.setStatus({ code: 2, message: 'Error' });
span.end();
throw error;
}
}
async function transcribeWithTracking(audioPath: string): Promise<string> {
const span = new SpanWrapper('cartesia-stt');
span.start();
try {
const startTime = Date.now();
const audioBuffer = fs.readFileSync(audioPath);
const audioSizeBytes = audioBuffer.length;
span.setAttribute('audio_file', audioPath);
span.setAttribute('audio_size_bytes', audioSizeBytes);
span.setAttribute('model', 'ink-whisper');
const response = await cartesia.stt.transcribe({
model_id: 'ink-whisper',
audio: audioBuffer,
language: 'en'
});
const duration = Date.now() - startTime;
// Track the STT API operation
const action: ActionModel = {
start_time: (startTime * 1000000).toString(),
action: 'API',
action_type: 'STT_TRANSCRIPTION',
metadata: {
provider: 'cartesia',
model: 'ink-whisper',
audio_size_bytes: audioSizeBytes.toString(),
transcript_length: response.transcript.length.toString(),
duration_ms: duration.toString(),
language: 'en'
},
success: true
};
span.setAction([action]);
span.setUsage({
model: 'ink-whisper',
usage_type: 'audio_seconds',
units_used: response.duration || 0,
cost_in_usd: (response.duration || 0) * 0.036
});
span.setAttribute('transcript_length', response.transcript.length);
span.setStatus({ code: 1, message: 'Success' });
span.end();
return response.transcript;
} catch (error) {
span.setError(error as Error);
span.setStatus({ code: 2, message: 'Error' });
span.end();
throw error;
}
}
// Usage
const audio = await generateSpeechWithTracking(
'This is ultra-low latency speech synthesis.',
'a0e99841-438c-4a64-b679-ae501e7d6091'
);
const transcript = await transcribeWithTracking('./audio/sample.wav');
Next Steps
- Netra Documentation - Learn more about Netra’s observability features
- Cartesia API - Explore Cartesia’s ultra-low latency TTS and STT