Translate Streaming STT Transcripts with LeMUR

1
import assemblyai as aai
2
3
aai.settings.api_key = "API_KEY"
4
5
def on_open(session_opened: aai.RealtimeSessionOpened):
6
    "This function is called when the connection has been established."
7
8
    print("Session ID:", session_opened.session_id)
9
10
def on_error(error: aai.RealtimeError):
11
    "This function is called when an error has been thrown."
12
13
    print("An error occurred:", error)
14
15
def on_close():
16
    "This function is called when the connection has been closed."
17
18
    print("Closing Session")
19
20
21
def on_data(transcript: aai.RealtimeTranscript):
22
    "This function is called when a new transcript has been received."
23
24
    if not transcript.text:
25
        return
26
27
    if isinstance(transcript, aai.RealtimeFinalTranscript):
28
        result = aai.Lemur().task("Translate the following text into Spanish. Do not write a preamble. Just return the translated text.", input_text=transcript.text)
29
        print(result.response, end="\r\n")
30
31
transcriber = aai.RealtimeTranscriber(
32
    on_data=on_data,
33
    on_error=on_error,
34
    sample_rate=44_100,
35
    on_open=on_open,
36
    on_close=on_close,
37
)
38
39
# Start the connection
40
transcriber.connect()
41
42
# Open a microphone stream
43
microphone_stream = aai.extras.MicrophoneStream()
44
45
# Press Ctrl+C to abort
46
transcriber.stream(microphone_stream)
47
48
transcriber.close()

1
# (Mac)
2
brew install portaudio
3
4
# (Debian/Ubuntu)
5
apt install portaudio19-dev
6
7
pip install "assemblyai[extras]"

1
import assemblyai as aai
2
3
aai.settings.api_key = "API_KEY"

1
def on_open(session_opened: aai.RealtimeSessionOpened):
2
    "This function is called when the connection has been established."
3
4
    print("Session ID:", session_opened.session_id)
5
6
def on_error(error: aai.RealtimeError):
7
    "This function is called when an error has been thrown."
8
9
    print("An error occurred:", error)
10
11
def on_close():
12
    "This function is called when the connection has been closed."
13
14
    print("Closing Session")

1
def on_data(transcript: aai.RealtimeTranscript):
2
    "This function is called when a new transcript has been received."
3
4
    if not transcript.text:
5
        return
6
7
    if isinstance(transcript, aai.RealtimeFinalTranscript):
8
        result = aai.Lemur().task(
9
            "Translate the following text into Spanish. Do not write a preamble. Just return the translated text.",
10
            input_text=transcript.text,
11
            final_model=aai.LemurModel.claude3_haiku)
12
        print(result.response, end="\r\n")

1
transcriber = aai.RealtimeTranscriber(
2
    on_data=on_data,
3
    on_error=on_error,
4
    sample_rate=44_100,
5
    on_open=on_open, # Optional
6
    on_close=on_close, # Optional
7
)

1
# Start the connection
2
transcriber.connect()
3
4
# Open a microphone stream
5
microphone_stream = aai.extras.MicrophoneStream()
6
7
# Press Ctrl+C to abort
8
transcriber.stream(microphone_stream)
9
10
transcriber.close()