Используем API Speech2Text для распознавания записей разговоров

В нашей компании анализируются звонки менеджеров отдела продаж для оценки их эффективности, устранения недочётов и улучшения сервиса. На сегодняшний день это составляет немалый массив ручной работы, для облегчения которой мы задумали привлечь технологии искусственного интеллекта. Идея следующая: забираем записи звонков, распознаём речь (преобразовываем в текст), подключаем LLM для анализа текста, знакомимся с выводами, при необходимости (например, возникновении каких-то аномалий) контролируем происходящее вручную.

Распознавание аудио решили делать через сервис Speech2Text, пример использования API которого я и покажу в этой статье.

В черновом варианте получаем примерно следующую схему работы (нас сейчас интересует прямоугольник с подписью Speech2Text connector):

Конвейер обработки аудиозаписей

Реализовывать взаимодействие будем в C# ASP.NET Core приложении. Очевидно, непосредственно обработка данных выполняется в backend, а для удобного управления можно будет использовать пользовательский интерфейс (frontend). Штош, приступим!

Служба взаимодействия с сервисом описывается следующим интерфейсом:

/// <summary>
/// Interface for Speech2Text interaction service.
/// </summary>
public interface ISpeechToText
{
    /// <summary>
    /// Creates a task on the Speech2Text server.
    /// </summary>
    /// <param name="payload">StreamContent containing the audio file.
    /// Will be disposed after use.</param>
    /// <returns>Task ID on the server or NULL in case of error.</returns>
    Task<string?> SendTaskAsync(StreamContent payload);

    /// <summary>
    /// Checks the status of a task on the Speech2Text server.
    /// </summary>
    /// <param name="taskId">Required task ID received from the server when creating the task.</param>
    /// <returns>JobStatus.Decoding if the task is in progress;
    /// JobStatus.Decoded if the task was successfully processed;
    /// JobStatus.FailedToDecode if processing failed;
    /// null if the server response was not received or recognized, or in case of other errors.
    /// </returns>
    Task<JobStatus?> GetTaskStatusAsync(string taskId);

    /// <summary>
    /// Gets the processing result of a task from the Speech2Text server.
    /// </summary>
    /// <param name="taskId">Required task ID received from the server when creating the task.</param>
    /// <returns>A string containing the processing result,
    /// or null in case of an error.</returns>
    Task<string?> GetTaskResultAsync(string taskId);
}

Настройки подключения, такие как адрес сервера и ключ API, будем сохранять в appsettings.json и передавать их, используя инъекцию зависимостей, в чём нам поможет Options pattern. Опишем модель для маппинга настроек:

public class SpeechToTextApiSettings
{
    public static string SectionName { get; } = "Speech2textApi";
    public string BaseUrl { get; set; } = string.Empty;
    public string TaskUrl { get; set; } = string.Empty;
    public string ApiKey { get; set; } = string.Empty;

    /// <summary>
    /// HttpClient timeout in minutes
    /// </summary>
    public int Timeout { get; set; } = 1;
}

и добавим сами настройки в appsettings.json(danger: ключ API здесь лежит на виду, вы знаете, что с этим делать):

"Speech2textApi": {
  "BaseUrl": "https://speech2text.ru/api/recognitions",
  "TaskUrl": "https://speech2text.ru/api/recognitions/task/file",
  "ApiKey": "MY-API-KEY-HERE",
  "Timeout": 1
}

Заготовка реализации службы SpeechToText будет выглядеть следующим образом:

public class SpeechToText : ISpeechToText
{
    private readonly ILogger<SpeechToText> _logger;
    private readonly HttpClient _httpClient;
    private readonly SpeechToTextApiSettings _settings;

    public SpeechToText(
        HttpClient httpClient,
        IOptions<SpeechToTextApiSettings> settings,
        ILogger<SpeechToText> logger)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
        _settings = settings?.Value ?? throw new ArgumentNullException(nameof(settings));

        if (string.IsNullOrWhiteSpace(_settings.BaseUrl))
        {
            throw new ArgumentNullException(nameof(_settings.BaseUrl), 
                "Base URL can't be empty");
        }
        if (string.IsNullOrWhiteSpace(_settings.TaskUrl))
        {
            throw new ArgumentNullException(nameof(_settings.TaskUrl), 
                "Task URL can't be empty");
        }
        _httpClient.BaseAddress = new Uri(_settings.BaseUrl);
        _httpClient.Timeout = TimeSpan.FromMinutes(_settings.Timeout);
        _httpClient.DefaultRequestHeaders.Authorization = 
            new("Bearer", _settings.ApiKey);
        _httpClient.DefaultRequestHeaders.Accept.Add(
            new("application/json"));
    }

    public async Task<string?> SendTaskAsync(StreamContent payload)
    {
        throw new NotImplementedException();
    }

    public async Task<JobStatus?> GetTaskStatusAsync(string taskId)
    {
        throw new NotImplementedException();
    }

    public async Task<string?> GetTaskResultAsync(string taskId)
    {
        throw new NotImplementedException();
    }
}

Здесь мы получаем параметры конфигурации и соответствующим образом настраиваем HttpClient. Используем Bearer Authentication, добавив соответствующий заголовок (строка 28). Кроме того, мы хотим ответ в формате JSON, поэтому добавляем и такой заголовок (строка 30).

Для логирования я использую nLog, впрочем, это сейчас несущественно, поэтому я уберу все вызовы логгера из приводимого здесь кода, чтобы не захламлять его.

Отправка файла на сервер производится POST запросом. В ответ, сервер вернёт идентификатор задания, который нам нужно сохранить – именно по этому идентификатору мы впоследствии найдём задание и узнаем, выполнена ли транскрибация записи. Как именно узнаем? Всё просто, ответ сервера будет содержать статус. Сразу опишем возможные статусы:

/// <summary>
/// Speech2Text server status codes.
/// </summary>
public enum SpeechToTextStatuses
{
    /// <summary>
    /// Content is received.
    /// </summary>
    Received = 80,
    /// <summary>
    /// Task in progress.
    /// </summary>
    Processing = 100,
    /// <summary>
    /// Completed successfully.
    /// </summary>
    Completed = 200,
    /// <summary>
    /// Error while processing.
    /// </summary>
    Error = 501
}

Теперь мы готовы написать реализацию метода отправки задания:

public async Task<string?> SendTaskAsync(StreamContent payload)
{
    try
    {
        using var content = new MultipartFormDataContent();
        content.Add(payload, "file", "audio.mp3");
        content.Add(new StringContent("ru"), "lang");
        content.Add(new StringContent("2"), "speakers");

        using var response = await _httpClient.PostAsync(_settings.TaskUrl, content);
        var responseText = await response.Content.ReadAsStringAsync();
        if (!response.IsSuccessStatusCode)
        {
            return null;
        }

        using JsonDocument doc = JsonDocument.Parse(responseText);
        var root = doc.RootElement;
        if (root.TryGetProperty("id", out var taskId)
            && taskId.GetString() is string taskIdValue
            && root.TryGetProperty("status", out var taskStatus)
            && taskStatus.TryGetProperty("code", out var taskCode)
            && taskCode.TryGetInt32(out int taskCodeValue))
        {
            if (taskCodeValue == (int)SpeechToTextStatuses.Received
                || taskCodeValue == (int)SpeechToTextStatuses.Processing)
            {
                return taskIdValue;
            }
        }
        return null;
    }
    catch (Exception ex)
    {
        return null;
    }
    finally
    {
        payload?.Dispose();
    }
}

Сервер принимает дополнительные параметры распознавания: lang для указания языка и speakers для количества говорящих (либо max_speakers и min_speakers, если участников диалога может быть переменное количество). Эти параметры необязательны, но я их установил, потому что у меня все записи однотипные. Разумеется, лучше избегать hardcoded values и следовало бы передавать аргументом некий DTO, содержащий не только само содержимое файла, но и эти дополнительные параметры.

Ещё есть интересный параметр multi_channel, который я не использовал. Устанавливается в 1, если файл содержит стереозвук, в котором один собеседник в одном канале, а другой – во втором.

Я не стал описывать модель для парсинга ответа сервера, потому что мне нужен только один параметр. Вообще, полный ответ выглядит так:

{
  "id": "EUmFNuJzxuc0fAf8pjHaq29RDwF3Wuj0",
  "created": null,
  "options": {
    "lang": "ru",
    "speakers": 2,
    "multi_channel": null
  },
  "file_meta": {
    "mime": "audio/mpeg",
    "format": "MPEG Audio",
    "audio_format": "MPEG Audio",
    "channels": 1,
    "duration": "00:01:14"
  },
  "resource": {
    "type": "file",
    "name": "audio.mp3"
  },
  "status": {
    "code": 100,
    "description": "В очереди на распознание"
  },
  "payment": {
    "source": 1,
    "price": 0
  },
  "result": null
}

Поставленное в очередь задание будет обрабатываться в течение некоторого времени. По завершению обработки, код статуса изменится на 200 – успешно завершено или 501 – ошибка транскрибации. Периодически опрашиваем сервер, вызывая соответствующий метод, код которого будет таким:

public async Task<JobStatus?> GetTaskStatusAsync(string taskId)
{
    if (string.IsNullOrWhiteSpace(taskId))
    {
        return null;
    }

    try
    {
        string uriString = $"{_settings.BaseUrl.TrimEnd('/')}/{taskId}";
        using var response = await _httpClient.GetAsync(uriString);
        var responseText = await response.Content.ReadAsStringAsync();
        if (!response.IsSuccessStatusCode)
        {
            return null;
        }
        using JsonDocument doc = JsonDocument.Parse(responseText);
        var root = doc.RootElement;
        if (root.TryGetProperty("status", out var taskStatus)
            && taskStatus.TryGetProperty("code", out var taskCode)
            && taskCode.TryGetInt32(out int taskCodeValue))
        {
            switch (taskCodeValue)
            {
                case (int)SpeechToTextStatuses.Received:
                case (int)SpeechToTextStatuses.Processing:
                    return JobStatus.Decoding;
                case (int)SpeechToTextStatuses.Completed:
                    return JobStatus.Decoded;
                case (int)SpeechToTextStatuses.Error:
                    return JobStatus.FailedToDecode;
                default:
                    return JobStatus.FailedToDecode;
            }
        }
        return null;
    }
    catch (Exception ex)
    {
        return null;
    }
}

Здесь мы проверяем ответ сервера и соответствующим образом устанавливаем статус задания в нашей локальной БД.

Дождавшись завершения задачи, заберём результат с сервера:

public async Task<string?> GetTaskResultAsync(string taskId)
{
    if (string.IsNullOrWhiteSpace(taskId))
    {
        return null;
    }
    try
    {
        string uriString = $"{_settings.BaseUrl.TrimEnd('/')}/{taskId}/result/txt";
        using var response = await _httpClient.GetAsync(uriString);
        var responseText = await response.Content.ReadAsStringAsync();
        if (!response.IsSuccessStatusCode)
        {
            return null;
        }
        return string.IsNullOrWhiteSpace(responseText) ? null : responseText;
    }
    catch (Exception ex)
    {
        return null;
    }
}

Здесь у нас имеет значение младший сегмент пути URL (кстати, как правильно это называется?), определяющий формат возвращаемого результата. Возможные варианты: raw, txt, srt, vtt, json и xml. Как видно из кода, я использую текстовое представление и получаю результат в следующем виде:

Спикер 1:

0:00:00 - Да, Алексей, здравствуйте.

Спикер 2:

0:00:03 - Я там вам письмо написал, вы видели?

Спикер 1:

0:00:06 - ...

Взаимодействие со службой происходит в цикле, находящемся внутри фоновой задачи. Между итерациями цикла обязательно делаем задержку, чтоб не DoSить сервер запросами. Упрощённо и сокращённо это выглядит примерно так, как в коде ниже. Здесь мы выбираем задания из локальной БД, основываясь на их статусах, делаем запросы к серверу и соответствующим образом меняем статусы, тем самым обеспечивая продвижение задания по конвейеру обработки.

public class ProcessingPipeline : BackgroundService
{
    // ...

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // enumerate file names
                foreach (var file in _filesProcessor.GetMp3Files())
                {
                    // create jobs for newly added files
                }

                // enumerate executing jobs
                foreach (var job in await _jobsRepository.GetDecodingJobs())
                {
                    if (await _speechToText.GetTaskStatusAsync(job.TaskId) is JobStatus status)
                    {
                        // change statuses of completed tasks
                    }
                }

                // enumerate new jobs
                foreach (var job in await _jobsRepository.GetNewJobs())
                {
                    // create server task and change job status
                    if (_filesProcessor.ReadMp3FileToHttpStream(job.FileName) 
                        is StreamContent stream)
                    {
                        var result = await _speechToText.SendTaskAsync(stream);
                        if (result is null)
                        {
                            job.Status = JobStatus.FailedToDecode;
                        }
                        else
                        {
                            job.Status = JobStatus.Decoding;
                            job.TaskId = result;
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                break;
            }
            catch (Exception ex)
            {
                // _logger
            }
            await Task.Delay(_settings.Interval_ms, stoppingToken);
        }
    }
}

Я не рассматриваю в подробностях конвейер обработки, взаимодействие с БД и прочее, поскольку это за рамками статьи (тем не менее, можем поговорить и об этом, если будет интересно).

Вот таким нехитрым способом можно производить распознавание аудиозаписей. Со своей стороны хочу поблагодарить сервис за предоставленный тестовый доступ. На данный момент, я столкнулся с определёнными особенностями (аудиозаписи не очень хорошего качества, есть много терминов, аббревиатур и технического жаргона), из-за которых транскрибация не всегда так хороша, как того хотелось бы. Но, надеюсь, мы сможем это решить.

Задавайте вопросы, указывайте на ошибки, спасибо за внимание)