Community is at the core of every voice agent

1# Install the speechmatics package using the command "pip install speechmatics-rt"
2
3#!/usr/bin/env python3
4"""Real-time transcription with microphone."""
5
6import asyncio
7import os
8from dotenv import load_dotenv
9from speechmatics.rt import (
10    AsyncClient,
11    ServerMessageType,
12    TranscriptionConfig,
13    TranscriptResult,
14    OperatingPoint,
15    AudioFormat,
16    AudioEncoding,
17    Microphone,
18    AuthenticationError,
19)
20
21load_dotenv()
22
23
24async def main():
25    api_key = os.getenv("SPEECHMATICS_API_KEY")
26
27    transcript_parts = []
28
29    audio_format = AudioFormat(
30        encoding=AudioEncoding.PCM_S16LE,
31        chunk_size=4096,
32        sample_rate=16000,
33    )
34
35    transcription_config = TranscriptionConfig(
36        language="en",
37        enable_partials=True,
38        operating_point=OperatingPoint.ENHANCED,
39    )
40
41    mic = Microphone(
42        sample_rate=audio_format.sample_rate,
43        chunk_size=audio_format.chunk_size,
44    )
45
46    if not mic.start():
47        print("PyAudio not installed. Install: pip install pyaudio")
48        return
49
50    try:
51        async with AsyncClient(api_key=api_key) as client:
52            @client.on(ServerMessageType.ADD_TRANSCRIPT)
53            def handle_final_transcript(message):
54                result = TranscriptResult.from_message(message)
55                transcript = result.metadata.transcript
56                if transcript:
57                    print(f"[final]: {transcript}")
58                    transcript_parts.append(transcript)
59
60            @client.on(ServerMessageType.ADD_PARTIAL_TRANSCRIPT)
61            def handle_partial_transcript(message):
62                result = TranscriptResult.from_message(message)
63                transcript = result.metadata.transcript
64                if transcript:
65                    print(f"[partial]: {transcript}")
66
67            try:
68                print("Connected! Start speaking (Ctrl+C to stop)...\n")
69
70                await client.start_session(
71                    transcription_config=transcription_config,
72                    audio_format=audio_format,
73                )
74
75                while True:
76                    frame = await mic.read(audio_format.chunk_size)
77                    await client.send_audio(frame)
78
79            except KeyboardInterrupt:
80                pass
81            finally:
82                mic.stop()
83                print(f"\n\nFull transcript: {' '.join(transcript_parts)}")
84
85    except (AuthenticationError, ValueError) as e:
86        print(f"\nAuthentication Error: {e}")
87
88
89if __name__ == "__main__":
90    asyncio.run(main())
91