Public Access
507 lines
19 KiB
C#
507 lines
19 KiB
C#
using System.Threading.Channels;
|
|
using MeetingAssistant.Recording;
|
|
using Microsoft.CognitiveServices.Speech;
|
|
using Microsoft.CognitiveServices.Speech.Audio;
|
|
using Microsoft.CognitiveServices.Speech.Transcription;
|
|
using Microsoft.Extensions.Options;
|
|
|
|
namespace MeetingAssistant.Transcription;
|
|
|
|
public sealed class AzureSpeechStreamingTranscriptionProvider : IStreamingTranscriptionProvider
|
|
{
|
|
private readonly MeetingAssistantOptions options;
|
|
private readonly ILogger<AzureSpeechStreamingTranscriptionProvider> logger;
|
|
|
|
public AzureSpeechStreamingTranscriptionProvider(
|
|
IOptions<MeetingAssistantOptions> options,
|
|
ILogger<AzureSpeechStreamingTranscriptionProvider> logger)
|
|
{
|
|
this.options = options.Value;
|
|
this.logger = logger;
|
|
}
|
|
|
|
public async IAsyncEnumerable<TranscriptionSegment> TranscribeAsync(
|
|
IAsyncEnumerable<AudioChunk> audio,
|
|
SpeechRecognitionPipelineOptions pipelineOptions,
|
|
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
var enumerator = audio.GetAsyncEnumerator(cancellationToken);
|
|
try
|
|
{
|
|
if (!await enumerator.MoveNextAsync())
|
|
{
|
|
yield break;
|
|
}
|
|
|
|
var nextChunk = enumerator.Current;
|
|
var reconnectAttempt = 0;
|
|
var markerId = "";
|
|
var disconnectedMarkerWritten = false;
|
|
|
|
while (true)
|
|
{
|
|
var session = StartSession(
|
|
nextChunk,
|
|
enumerator,
|
|
pipelineOptions,
|
|
markerId,
|
|
cancellationToken);
|
|
|
|
await foreach (var segment in session.Segments.ReadAllAsync(cancellationToken))
|
|
{
|
|
if (segment.ReplacesMarkerId is not null)
|
|
{
|
|
reconnectAttempt = 0;
|
|
disconnectedMarkerWritten = false;
|
|
}
|
|
|
|
yield return segment;
|
|
}
|
|
|
|
var result = await session.Completion.WaitAsync(cancellationToken);
|
|
if (result.IsCompleted)
|
|
{
|
|
yield break;
|
|
}
|
|
|
|
reconnectAttempt++;
|
|
markerId = string.IsNullOrWhiteSpace(markerId)
|
|
? $"azure-reconnect-{Guid.NewGuid():N}"
|
|
: markerId;
|
|
nextChunk = result.NextChunk ?? nextChunk;
|
|
if (reconnectAttempt <= Math.Max(1, options.AzureSpeech.ReconnectAttemptsBeforeDisconnectedMarker))
|
|
{
|
|
logger.LogWarning(
|
|
"Azure Speech reconnecting after connection interruption, attempt {Attempt}/{MaxAttempts}: {ErrorDetails}",
|
|
reconnectAttempt,
|
|
Math.Max(1, options.AzureSpeech.ReconnectAttemptsBeforeDisconnectedMarker),
|
|
result.ErrorDetails);
|
|
yield return new TranscriptionSegment(
|
|
TimeSpan.Zero,
|
|
TimeSpan.Zero,
|
|
"System",
|
|
$"<Reconnecting... {reconnectAttempt}/{Math.Max(1, options.AzureSpeech.ReconnectAttemptsBeforeDisconnectedMarker)}>",
|
|
TranscriptionSegmentKind.Marker,
|
|
MarkerId: markerId);
|
|
}
|
|
else if (!disconnectedMarkerWritten)
|
|
{
|
|
disconnectedMarkerWritten = true;
|
|
logger.LogWarning(
|
|
"Azure Speech remains disconnected after {AttemptCount} reconnect attempt(s); recording continues and transcription will retry: {ErrorDetails}",
|
|
reconnectAttempt - 1,
|
|
result.ErrorDetails);
|
|
yield return new TranscriptionSegment(
|
|
TimeSpan.Zero,
|
|
TimeSpan.Zero,
|
|
"System",
|
|
"<Azure Speech disconnected. The meeting can continue; transcription and summarization will continue once Azure Speech reconnects.>",
|
|
TranscriptionSegmentKind.Marker,
|
|
MarkerId: markerId);
|
|
}
|
|
|
|
if (options.AzureSpeech.ReconnectDelay > TimeSpan.Zero)
|
|
{
|
|
await Task.Delay(options.AzureSpeech.ReconnectDelay, cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
try
|
|
{
|
|
await enumerator.DisposeAsync();
|
|
}
|
|
catch (NotSupportedException)
|
|
{
|
|
// Some diagnostic async enumerable sources expose DisposeAsync but throw after the stream has been read.
|
|
}
|
|
}
|
|
}
|
|
|
|
private AzureSpeechSession StartSession(
|
|
AudioChunk firstChunk,
|
|
IAsyncEnumerator<AudioChunk> audio,
|
|
SpeechRecognitionPipelineOptions pipelineOptions,
|
|
string reconnectMarkerId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var segments = Channel.CreateUnbounded<TranscriptionSegment>();
|
|
var completion = Task.Run(
|
|
() => RunSessionAsync(
|
|
firstChunk,
|
|
audio,
|
|
pipelineOptions,
|
|
reconnectMarkerId,
|
|
segments.Writer,
|
|
cancellationToken),
|
|
CancellationToken.None);
|
|
return new AzureSpeechSession(segments.Reader, completion);
|
|
}
|
|
|
|
private async Task<AzureSpeechSessionResult> RunSessionAsync(
|
|
AudioChunk firstChunk,
|
|
IAsyncEnumerator<AudioChunk> audio,
|
|
SpeechRecognitionPipelineOptions pipelineOptions,
|
|
string reconnectMarkerId,
|
|
ChannelWriter<TranscriptionSegment> segments,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
using var pushStream = AudioInputStream.CreatePushStream(
|
|
AudioStreamFormat.GetWaveFormatPCM(
|
|
(uint)firstChunk.SampleRate,
|
|
16,
|
|
(byte)firstChunk.Channels));
|
|
using var audioConfig = AudioConfig.FromStreamInput(pushStream);
|
|
var speechConfig = CreateSpeechConfig();
|
|
using var recognizer = CreateTranscriber(speechConfig, audioConfig);
|
|
AddPhraseList(recognizer, pipelineOptions.DictationWords);
|
|
var sessionStop = new TaskCompletionSource<AzureSpeechSessionResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var pendingReconnectMarkerId = reconnectMarkerId;
|
|
|
|
recognizer.Transcribed += (_, args) =>
|
|
{
|
|
if (args.Result.Reason != ResultReason.RecognizedSpeech
|
|
|| string.IsNullOrWhiteSpace(args.Result.Text))
|
|
{
|
|
return;
|
|
}
|
|
|
|
var start = TimeSpan.FromTicks(args.Result.OffsetInTicks);
|
|
var replacesMarkerId = string.IsNullOrWhiteSpace(pendingReconnectMarkerId)
|
|
? null
|
|
: Interlocked.Exchange(ref pendingReconnectMarkerId, null);
|
|
var segment = new TranscriptionSegment(
|
|
start,
|
|
start + args.Result.Duration,
|
|
NormalizeSpeakerId(args.Result.SpeakerId),
|
|
args.Result.Text.Trim(),
|
|
ReplacesMarkerId: replacesMarkerId);
|
|
logger.LogInformation(
|
|
"Azure Speech emitted segment at {Start} from {Speaker}: {Text}",
|
|
segment.Start,
|
|
segment.Speaker,
|
|
segment.Text);
|
|
segments.TryWrite(segment);
|
|
};
|
|
recognizer.Canceled += (_, args) =>
|
|
{
|
|
if (args.Reason == CancellationReason.Error)
|
|
{
|
|
logger.LogWarning(
|
|
"Azure Speech recognition canceled with {ErrorCode}: {ErrorDetails}",
|
|
args.ErrorCode,
|
|
args.ErrorDetails);
|
|
sessionStop.TrySetResult(AzureSpeechSessionResult.Reconnect(
|
|
$"Azure Speech recognition canceled: {args.ErrorCode} {args.ErrorDetails}"));
|
|
return;
|
|
}
|
|
|
|
sessionStop.TrySetResult(AzureSpeechSessionResult.Completed());
|
|
};
|
|
recognizer.SessionStopped += (_, _) => sessionStop.TrySetResult(AzureSpeechSessionResult.Completed());
|
|
|
|
await recognizer.StartTranscribingAsync();
|
|
return await PumpAudioAsync(
|
|
firstChunk,
|
|
audio,
|
|
pushStream,
|
|
recognizer,
|
|
sessionStop.Task,
|
|
options.AzureSpeech.RecognitionStopTimeout,
|
|
cancellationToken);
|
|
}
|
|
catch (Exception exception) when (IsTransientConnectionException(exception))
|
|
{
|
|
logger.LogWarning(exception, "Azure Speech session failed transiently; reconnecting.");
|
|
return AzureSpeechSessionResult.Reconnect(firstChunk, exception.Message);
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
segments.TryComplete(exception);
|
|
throw;
|
|
}
|
|
finally
|
|
{
|
|
segments.TryComplete();
|
|
}
|
|
}
|
|
|
|
internal ConversationTranscriber CreateTranscriber(SpeechConfig speechConfig, AudioConfig audioConfig)
|
|
{
|
|
var languages = GetAutoDetectLanguages(options.AzureSpeech);
|
|
if (languages.Count == 0)
|
|
{
|
|
return new ConversationTranscriber(speechConfig, audioConfig);
|
|
}
|
|
|
|
logger.LogInformation(
|
|
"Azure Speech auto language detection enabled for {Languages}",
|
|
string.Join(", ", languages));
|
|
return new ConversationTranscriber(
|
|
speechConfig,
|
|
AutoDetectSourceLanguageConfig.FromLanguages(languages.ToArray()),
|
|
audioConfig);
|
|
}
|
|
|
|
internal void AddPhraseList(ConversationTranscriber recognizer, IReadOnlyList<string>? words)
|
|
{
|
|
var phrases = NormalizeDictationWords(words);
|
|
if (phrases.Count == 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var phraseList = PhraseListGrammar.FromRecognizer(recognizer);
|
|
foreach (var phrase in phrases)
|
|
{
|
|
phraseList.AddPhrase(phrase);
|
|
}
|
|
|
|
phraseList.SetWeight(options.AzureSpeech.PhraseListWeight);
|
|
logger.LogInformation("Configured Azure Speech phrase list with {PhraseCount} phrase(s)", phrases.Count);
|
|
}
|
|
|
|
internal static IReadOnlyList<string> NormalizeDictationWords(IReadOnlyList<string>? words)
|
|
{
|
|
return (words ?? [])
|
|
.Select(word => word.Trim())
|
|
.Where(word => !string.IsNullOrWhiteSpace(word))
|
|
.Distinct(StringComparer.OrdinalIgnoreCase)
|
|
.ToList();
|
|
}
|
|
|
|
internal SpeechConfig CreateSpeechConfig()
|
|
{
|
|
var azure = options.AzureSpeech;
|
|
var key = ResolveKey(azure);
|
|
if (string.IsNullOrWhiteSpace(key))
|
|
{
|
|
throw new InvalidOperationException(
|
|
$"Azure Speech key is not configured. Set AzureSpeech:Key or environment variable {azure.KeyEnv}.");
|
|
}
|
|
|
|
var speechConfig = CreateSpeechConfig(azure, key);
|
|
var autoDetectLanguages = GetAutoDetectLanguages(azure);
|
|
if (autoDetectLanguages.Count == 0)
|
|
{
|
|
speechConfig.SpeechRecognitionLanguage = azure.Language;
|
|
}
|
|
else
|
|
{
|
|
speechConfig.SetProperty(
|
|
PropertyId.SpeechServiceConnection_LanguageIdMode,
|
|
NormalizeLanguageIdMode(azure.LanguageIdMode));
|
|
}
|
|
|
|
speechConfig.OutputFormat = OutputFormat.Detailed;
|
|
speechConfig.SetProperty(
|
|
PropertyId.SpeechServiceResponse_DiarizeIntermediateResults,
|
|
azure.DiarizeIntermediateResults ? "true" : "false");
|
|
if (!string.IsNullOrWhiteSpace(azure.PostProcessingOption))
|
|
{
|
|
logger.LogWarning(
|
|
"Azure Speech post-processing option {PostProcessingOption} is configured but skipped because the live Azure backend uses ConversationTranscriber; Speech SDK post-processing is documented for SpeechRecognizer.",
|
|
azure.PostProcessingOption.Trim());
|
|
}
|
|
|
|
return speechConfig;
|
|
}
|
|
|
|
internal static IReadOnlyList<string> GetAutoDetectLanguages(AzureSpeechOptions options)
|
|
{
|
|
if (!options.Language.Equals("auto", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return [];
|
|
}
|
|
|
|
return options.AutoDetectLanguages
|
|
.Select(language => language.Trim())
|
|
.Where(language => !string.IsNullOrWhiteSpace(language))
|
|
.Distinct(StringComparer.OrdinalIgnoreCase)
|
|
.ToList();
|
|
}
|
|
|
|
internal static Uri GetEndpoint(AzureSpeechOptions options)
|
|
{
|
|
return string.IsNullOrWhiteSpace(options.Endpoint)
|
|
? throw new InvalidOperationException("Azure Speech endpoint must be configured.")
|
|
: new Uri(options.Endpoint);
|
|
}
|
|
|
|
internal static SpeechConfig CreateSpeechConfig(AzureSpeechOptions options, string key)
|
|
{
|
|
if (!string.IsNullOrWhiteSpace(options.Endpoint))
|
|
{
|
|
return SpeechConfig.FromEndpoint(GetEndpoint(options), key);
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(options.Region))
|
|
{
|
|
throw new InvalidOperationException("Azure Speech region or endpoint must be configured.");
|
|
}
|
|
|
|
return SpeechConfig.FromSubscription(key, options.Region.Trim());
|
|
}
|
|
|
|
internal static string NormalizeLanguageIdMode(string? value)
|
|
{
|
|
return value?.Trim().Equals("AtStart", StringComparison.OrdinalIgnoreCase) == true
|
|
? "AtStart"
|
|
: "Continuous";
|
|
}
|
|
|
|
private static async Task<AzureSpeechSessionResult> PumpAudioAsync(
|
|
AudioChunk firstChunk,
|
|
IAsyncEnumerator<AudioChunk> audio,
|
|
PushAudioInputStream pushStream,
|
|
ConversationTranscriber recognizer,
|
|
Task<AzureSpeechSessionResult> sessionStop,
|
|
TimeSpan recognitionStopTimeout,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
WriteChunk(pushStream, firstChunk, firstChunk);
|
|
while (true)
|
|
{
|
|
var moveNextTask = audio.MoveNextAsync().AsTask();
|
|
var completedTask = await Task.WhenAny(moveNextTask, sessionStop);
|
|
if (completedTask == sessionStop)
|
|
{
|
|
var stop = await sessionStop;
|
|
if (stop.IsCompleted)
|
|
{
|
|
return AzureSpeechSessionResult.Completed();
|
|
}
|
|
|
|
if (!await moveNextTask)
|
|
{
|
|
return AzureSpeechSessionResult.Completed();
|
|
}
|
|
|
|
return AzureSpeechSessionResult.Reconnect(audio.Current, stop.ErrorDetails);
|
|
}
|
|
|
|
if (!await moveNextTask)
|
|
{
|
|
break;
|
|
}
|
|
|
|
if (sessionStop.IsCompleted)
|
|
{
|
|
var stop = await sessionStop;
|
|
if (!stop.IsCompleted)
|
|
{
|
|
return AzureSpeechSessionResult.Reconnect(audio.Current, stop.ErrorDetails);
|
|
}
|
|
}
|
|
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
WriteChunk(pushStream, firstChunk, audio.Current);
|
|
}
|
|
|
|
pushStream.Close();
|
|
try
|
|
{
|
|
var stopTask = recognizer.StopTranscribingAsync();
|
|
if (recognitionStopTimeout > TimeSpan.Zero)
|
|
{
|
|
await stopTask.WaitAsync(recognitionStopTimeout, cancellationToken);
|
|
}
|
|
else
|
|
{
|
|
await stopTask.WaitAsync(cancellationToken);
|
|
}
|
|
}
|
|
catch (TimeoutException)
|
|
{
|
|
// The SDK can outlive short diagnostic streams while waiting for final service events.
|
|
}
|
|
catch (Exception exception) when (IsTransientConnectionException(exception))
|
|
{
|
|
return AzureSpeechSessionResult.Reconnect(firstChunk, exception.Message);
|
|
}
|
|
|
|
return AzureSpeechSessionResult.Completed();
|
|
}
|
|
|
|
private static bool IsTransientConnectionException(Exception exception)
|
|
{
|
|
if (exception is OperationCanceledException)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
return exception is TimeoutException
|
|
or IOException
|
|
or HttpRequestException
|
|
or System.Net.Sockets.SocketException
|
|
|| exception.Message.Contains("connection", StringComparison.OrdinalIgnoreCase)
|
|
|| exception.Message.Contains("websocket", StringComparison.OrdinalIgnoreCase)
|
|
|| exception.Message.Contains("network", StringComparison.OrdinalIgnoreCase)
|
|
|| exception.Message.Contains("transport", StringComparison.OrdinalIgnoreCase);
|
|
}
|
|
|
|
private sealed record AzureSpeechSession(
|
|
ChannelReader<TranscriptionSegment> Segments,
|
|
Task<AzureSpeechSessionResult> Completion);
|
|
|
|
private sealed record AzureSpeechSessionResult(
|
|
bool IsCompleted,
|
|
AudioChunk? NextChunk,
|
|
string? ErrorDetails)
|
|
{
|
|
public static AzureSpeechSessionResult Completed()
|
|
{
|
|
return new AzureSpeechSessionResult(true, null, null);
|
|
}
|
|
|
|
public static AzureSpeechSessionResult Reconnect(AudioChunk nextChunk, string? errorDetails)
|
|
{
|
|
return new AzureSpeechSessionResult(false, nextChunk, errorDetails);
|
|
}
|
|
|
|
public static AzureSpeechSessionResult Reconnect(string? errorDetails)
|
|
{
|
|
return new AzureSpeechSessionResult(false, null, errorDetails);
|
|
}
|
|
}
|
|
|
|
private static void WriteChunk(PushAudioInputStream pushStream, AudioChunk expectedFormat, AudioChunk chunk)
|
|
{
|
|
if (chunk.SampleRate != expectedFormat.SampleRate || chunk.Channels != expectedFormat.Channels)
|
|
{
|
|
return;
|
|
}
|
|
|
|
pushStream.Write(chunk.Pcm);
|
|
}
|
|
|
|
private static string NormalizeSpeakerId(string? speakerId)
|
|
{
|
|
return string.IsNullOrWhiteSpace(speakerId) || speakerId.Equals("Unknown", StringComparison.OrdinalIgnoreCase)
|
|
? "Unknown"
|
|
: speakerId;
|
|
}
|
|
|
|
internal static string ResolveKey(AzureSpeechOptions options)
|
|
{
|
|
if (!string.IsNullOrWhiteSpace(options.Key))
|
|
{
|
|
return options.Key;
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(options.KeyEnv))
|
|
{
|
|
return "";
|
|
}
|
|
|
|
return Environment.GetEnvironmentVariable(options.KeyEnv) ??
|
|
Environment.GetEnvironmentVariable(options.KeyEnv, EnvironmentVariableTarget.User) ??
|
|
Environment.GetEnvironmentVariable(options.KeyEnv, EnvironmentVariableTarget.Machine) ??
|
|
"";
|
|
}
|
|
}
|