2d3f61aa07
Changes Split out module attributes to make engine settings clearer Removed internal audio buffer to use latest Speechmatics python SDK (0.4.0) Use diarization for improved VAD in multi-speaker situations Support custom dictionary / vocabulary with attributes Deprecated attributes superseded by re-organised attributes Diarization Enhancements Focus on specific speakers (using speaker labels) Ignore specific speakers (using speaker labels) Separate transcription formats for active and inactive speakers Support for known speakers
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
#
|
||
# Copyright (c) 2024–2025, Daily
|
||
#
|
||
# SPDX-License-Identifier: BSD 2-Clause License
|
||
#
|
||
|
||
import argparse
|
||
import os
|
||
|
||
from dotenv import load_dotenv
|
||
from loguru import logger
|
||
|
||
from pipecat.frames.frames import Frame, TranscriptionFrame
|
||
from pipecat.pipeline.pipeline import Pipeline
|
||
from pipecat.pipeline.runner import PipelineRunner
|
||
from pipecat.pipeline.task import PipelineTask
|
||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||
from pipecat.services.speechmatics.stt import SpeechmaticsSTTService
|
||
from pipecat.transcriptions.language import Language
|
||
from pipecat.transports.base_transport import BaseTransport, TransportParams
|
||
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
|
||
from pipecat.transports.services.daily import DailyParams
|
||
|
||
load_dotenv(override=True)
|
||
|
||
|
||
class TranscriptionLogger(FrameProcessor):
|
||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
if isinstance(frame, TranscriptionFrame):
|
||
print(f"Transcription: {frame.text}")
|
||
|
||
|
||
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
|
||
# instantiated. The function will be called when the desired transport gets
|
||
# selected.
|
||
transport_params = {
|
||
"daily": lambda: DailyParams(audio_in_enabled=True),
|
||
"twilio": lambda: FastAPIWebsocketParams(audio_in_enabled=True),
|
||
"webrtc": lambda: TransportParams(audio_in_enabled=True),
|
||
}
|
||
|
||
|
||
async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
|
||
"""Run example using Speechmatics STT.
|
||
|
||
This example will use diarization within our STT service and output the words spoken by
|
||
each individual speaker and wrap them with XML tags.
|
||
|
||
If you do not wish to use diarization, then set the `enable_speaker_diarization` parameter
|
||
to `False` or omit it altogether. The `text_format` will only be used if diarization is enabled.
|
||
|
||
By default, this example will use our ENHANCED operating point, which is optimized for
|
||
high accuracy. You can change this by setting the `operating_point` parameter to a different
|
||
value.
|
||
|
||
For more information on operating points, see the Speechmatics documentation:
|
||
https://docs.speechmatics.com/rt-api-ref
|
||
"""
|
||
logger.info(f"Starting bot")
|
||
|
||
stt = SpeechmaticsSTTService(
|
||
api_key=os.getenv("SPEECHMATICS_API_KEY"),
|
||
params=SpeechmaticsSTTService.InputParams(
|
||
language=Language.EN,
|
||
enable_diarization=True,
|
||
speaker_active_format="<{speaker_id}>{text}</{speaker_id}>",
|
||
),
|
||
)
|
||
|
||
tl = TranscriptionLogger()
|
||
|
||
pipeline = Pipeline([transport.input(), stt, tl])
|
||
|
||
task = PipelineTask(pipeline)
|
||
|
||
@transport.event_handler("on_client_disconnected")
|
||
async def on_client_disconnected(transport, client):
|
||
logger.info(f"Client disconnected")
|
||
await task.cancel()
|
||
|
||
runner = PipelineRunner(handle_sigint=handle_sigint)
|
||
|
||
await runner.run(task)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
from pipecat.examples.run import main
|
||
|
||
main(run_example, transport_params=transport_params)
|