Руководство по распознаванию речи

Подготовка

В этом руководстве можно посмотреть, как и для каких целей используются методы распознавания речи с примерами.

Эндпоинт для распознавания — api.tinkoff.ai:443.

1. Для авторизации в сервисах получаем ключи и вставляем их в переменные среды.

   export VOICEKIT_API_KEY="PUT_YOUR_API_KEY_HERE"
   export VOICEKIT_SECRET_KEY="PUT_YOUR_SECRET_KEY_HERE"

2. Клонируем репозиторий с примерами.

   git clone --recursive https://github.com/Tinkoff/voicekit-examples.git

3. Устанавливаем зависимости.

   sudo apt-get install python3 python-pyaudio python3-pyaudio
   sudo python3 -m pip install -r requirements/all.txt

В сервисе есть 4 метода для распознавания речи:

• Распознавание речи (Recognize). Работает по принципу «загружаем аудио целиком, получаем ответ». Полезен для распознавания аудиофайлов.

• Потоковое распознавание речи (StreamingRecognize). Нужен для распознавания речи в реальном времени: телефонных звонков, голосовых ассистентов и так далее. Также в методе больше возможностей для распознавания файлов.

• Отложенное распознавание речи (LongRunningRecognize). Работает по принципу «отправляем аудио целиком, а результат — когда он будет готов — забираем из отдельного интерфейса».

• Потоковое распознавание речи с синхронным ответом в конце (StreamingUnaryRecognize). Используется, если:

  • есть большие аудио, которые не укладываются в ограничения Recognize;
  • нужен результат распознавания сразу после окончания аудиопотока, но без промежуточных результатов.

Распознавание речи

Пример 1: метод Recognize() для LINEAR16

$ ./stt_recognize_linear16_raw.py

Самый простой режим запрос-ответ: загружаем аудио целиком, получаем ответ.

В этом примере также загружаем сэмплы из «сырого» формата .s16 — он не содержит метаинформации, в отличие от .wav.

В нашем случае работаем с одноканальным аудио с частотой дискретизации 16 KHz.

1. Импортируем модули.

   from tinkoff.cloud.stt.v1 import stt_pb2_grpc, stt_pb2 # сообщения и стабы gRPC API
   from auth import authorization_metadata # для авторизации по JWT
   import grpc
   import os

2. Получаем конфигурацию.

   # можно получать из переменных среды или заменить в сниппете
   endpoint = os.environ.get("VOICEKIT_ENDPOINT") or "api.tinkoff.ai:443"
   api_key = os.environ["VOICEKIT_API_KEY"]
   secret_key = os.environ["VOICEKIT_SECRET_KEY"]

3. Создаем запрос.

   def build_request():
       request = stt_pb2.RecognizeRequest()
       with open("../audio/sample_3.s16", "rb") as f:
           request.audio.content = f.read()
       request.config.encoding = stt_pb2.AudioEncoding.LINEAR16
       request.config.sample_rate_hertz = 16000 # Значение не содержится в файле `.s16`
       request.config.num_channels = 1 # Значение не содержится в файле `.s16`
       return request

4. Реализуем печать ответа.

   def print_recognition_response(response):
       for result in response.results:
           print("Channel", result.channel)
           print("Phrase start:", result.start_time.ToTimedelta())
           print("Phrase end:  ", result.end_time.ToTimedelta())
           for alternative in result.alternatives:
               print('"' + alternative.transcript + '"')
           print("----------------------------")

5. Отправляем запрос.

   stub = stt_pb2_grpc.SpeechToTextStub(grpc.secure_channel(endpoint, grpc.ssl_channel_credentials()))
   metadata = authorization_metadata(api_key, secret_key, "tinkoff.cloud.stt")
   response = stub.Recognize(build_request(), metadata=metadata)
   print_recognition_response(response)

Исходный код примера

Потоковое распознавание речи

Кодеки и форматы файлов

Пример 2: переходим на метод StreamingRecognize()

$ ./stt_streaming_recognize_linear16_raw.py

В режиме StreamingRecognize() мы можем слать аудио по частям. Используем этот режим для распознавания аудио в реальном времени.

Первый запрос — сообщение streaming_config, все последующие — audio_content.

1. Создаем первый запрос:

   def build_first_request():
       request = stt_pb2.StreamingRecognizeRequest()
       request.streaming_config.config.encoding = stt_pb2.AudioEncoding.LINEAR16
       request.streaming_config.config.sample_rate_hertz = 16000
       request.streaming_config.config.num_channels = 1
       return request

2. Создаем генератор запросов.

   def generate_requests():
       try:
           yield build_first_request()
           with open("../audio/sample_3.s16", "rb") as f:
               for data in iter(lambda:f.read(3200), b''): # Шлём по 100 миллисекунд (2 байта на сэмпл)
                   request = stt_pb2.StreamingRecognizeRequest()
                   request.audio_content = data
                   yield request
       except Exception as e: # обрабатываем исключение внутри генератора
           print("Got exception in generate_requests", e)
           raise

3. Создаем печать ответов.

   def print_streaming_recognition_responses(responses):
       for response in responses:
           for result in response.results:
               print("Channel", result.recognition_result.channel)
               print("Phrase start:", result.recognition_result.start_time.ToTimedelta())
               print("Phrase end:  ", result.recognition_result.end_time.ToTimedelta())
               for alternative in result.recognition_result.alternatives:
                   print('"' + alternative.transcript + '"')
               print("------------------")

4. Отправляем запрос.

   stub = stt_pb2_grpc.SpeechToTextStub(grpc.secure_channel(endpoint, grpc.ssl_channel_credentials()))
   metadata = authorization_metadata(api_key, secret_key, "tinkoff.cloud.stt")
   responses = stub.StreamingRecognize(generate_requests(), metadata=metadata)
   print_streaming_recognition_responses(responses)

Исходный код примера

Пример 3: загружаем аудио из .wav-файла

$ ./stt_streaming_recognize_linear16_wav.py

Попробуем загрузить аудио из контейнера RIFF WAVE. Пример работает с .wav-файлом с аудио в формате PCM.

1. Импортируем модуль wave.

   import wave

2. Пробрасываем конфигурацию аудио в первый запрос.

   def build_first_request(sample_rate_hertz, num_channels):
       request = stt_pb2.StreamingRecognizeRequest()
       request.streaming_config.config.encoding = stt_pb2.AudioEncoding.LINEAR16
       request.streaming_config.config.sample_rate_hertz = sample_rate_hertz
       request.streaming_config.config.num_channels = num_channels
       return request

3. Читаем сэмплы вместо байтов.

   with wave.open("../audio/sample_3.wav") as f:
       yield build_first_request(f.getframerate(), f.getnchannels())
       frame_samples = f.getframerate()//10 # Шлём по 100 миллисекунд
       for data in iter(lambda:f.readframes(frame_samples), b''):

Исходный код примера

Пример 4: используем кодек A-Law

$ ./stt_streaming_recognize_alaw_raw.py

A-Law — кодек с квази-экспоненциальным кодированием амплитуды аудиоимпульса.

Он позволяет кодировать аудио по 1 байту за сэмпл с сохранением приемлемого качества — в отличие от линейного кодирования. Считается очень производительным форматом сжатия с потерями с точки зрения нагрузки на CPU, так как сжатие сводится к независимому преобразованию сэмплов.

В основном применяется в телефонии. В нашем примере, как и в типичной конфигурации VoIP, используется частота в 8 KHz.

Мы также поддерживаем формат Mu-Law, но поскольку он является устаревшим аналогом A-Law с менее эффективным распределением кодируемых значений амплитуды аудиоимпульса, его использование оправданно только поддержкой совместимости.

Логика похожа на пример 2.

1. Задаем кодек и частоту дискретизации.

   request.streaming_config.config.encoding = stt_pb2.AudioEncoding.ALAW
   request.streaming_config.config.sample_rate_hertz = 8000

2. Адаптируем размер фрейма в байтах.

   for data in iter(lambda:f.read(800), b''): # Шлём по 100 миллисекунд (1 байт на сэмпл)

Исходный код примера

Пример 5: распознаём MP3

$ ./stt_streaming_recognize_mp3.py

MP3 поддерживается в формате контейнера с фиксированными количеством каналов и частотой дискретизации.

Меняем пример 3.

1. Импортируем модуль для чтения метаинформации из MP3.

   from mutagen.mp3 import MP3

2. Задаем формат кодирования.

   request.streaming_config.config.encoding = stt_pb2.AudioEncoding.MPEG_AUDIO

3. Заполняем метаинформацию и отправляем аудио — шлём по 4096 байт за раз. В реальности значение выбирается исходя из условий использования.

   fname = "../audio/sample_3.mp3"
    info = MP3(fname).info
    yield build_first_request(info.sample_rate, info.channels)
    with open(fname, "rb") as f:
        for data in iter(lambda:f.read(4096), b''): # Шлём по 4096 байт за раз

Исходный код примера

Пример 6: кодируем в Opus на лету

$ ./stt_streaming_recognize_raw_opus_from_wav.py

В этом примере читаем .wav-файл и кодируем в сырые Opus-пакеты.

На практике сжатие на лету удобно применять при работе с аудио-разговорами в реальном времени, при распознавании с микрофона и так далее.

На сегодняшний день Opus — наиболее эффективный кодек сжатия с потерями: по своим характеристикам он обходит как кодеки для разговоров — с небольшой задержкой декодирования, так и кодеки для хранения аудио-трэков — сжатые большими блоками для большего сжатия.

Задаем формат кодирования:

request.streaming_config.config.encoding = stt_pb2.AudioEncoding.RAW_OPUS

В режиме RAW_OPUS важно отправлять ровно по одному Opus-фрейму в запрос, так как сами фреймы не содержат информацию о длине фрейма.

Например, если сконкатенировать два фрейма в один, невозможно будет понять, где заканчивается первый фрейм и начинается второй.

Для хранения Opus-фреймов в файле используются контейнерные форматы, например Ogg — не путать с аудиокодеком Vorbis.

1. Реализуем проверку частоты дискретизации на допустимое значение.

   def frame_rate_is_valid(frame_rate):
       return frame_rate in [8000, 12000, 16000, 24000, 48000]

2. Реализуем вычисление оптимального размера выровненного фрейма для последнего фрагмента аудио.

   def get_padded_frame_size(frame_samples, frame_rate):
       for duration_half_msec in [5, 10, 20, 40, 80, 120]: # Допустимые размеры фреймов: 2.5, 5, 10, 20, 40 или 60 мсек.
           padded_samples = frame_rate//2000*duration_half_msec
           if frame_samples <= padded_samples:
               return padded_samples
       raise("Unexpected frame samples")

3. Инициализируем состояние энкодера в Opus и кодируем фреймами фиксированного размера — кроме последнего.

   with wave.open("../audio/sample_3.wav") as f:
       frame_rate_is_valid(f.getframerate())
       yield build_first_request(f.getframerate(), f.getnchannels())
       frame_samples = f.getframerate()//1000*60 # 60 мсек.
       opus_encoder = opuslib.Encoder(f.getframerate(), f.getnchannels(), opuslib.APPLICATION_AUDIO)
       for data in iter(lambda:f.readframes(frame_samples), b''): # Отправляем 60 мсек. за раз
           if len(data) < frame_samples*2: # Расширяем фрейм до ближайшего допустимого размера
               data = data.ljust(get_padded_frame_size(frame_samples, f.getframerate())*2, b'\0')
           request = stt_pb2.StreamingRecognizeRequest()
           request.audio_content = opus_encoder.encode(data, len(data) >> 1)
           yield request

Исходный код примера

Ввод с микрофона

Пример 7: распознаём с микрофона

$ ./stt_streaming_recognize_linear16_from_microphone.py

Для удобства тестирования сервиса можно делать запросы непосредственно с микрофона.

1. Импортируем модуль.

   import pyaudio

2. Инициализируем запись с микрофона и отправляем запросы.

   sample_rate_hertz, num_channels = 16000, 1
   pyaudio_lib = pyaudio.PyAudio()
   f = pyaudio_lib.open(input=True, channels=num_channels, format=pyaudio.paInt16, rate=sample_rate_hertz)
   yield build_first_request(sample_rate_hertz, num_channels)
   for data in iter(lambda:f.read(800), b''): # Шлём по 50ms за раз
       request = stt_pb2.StreamingRecognizeRequest()
       request.audio_content = data
       yield request

Исходный код примера

Режимы и опции распознавания

Пример 8: получаем несколько гипотез

$ ./stt_streaming_recognize_max_alternatives.py

Сервис может генерировать несколько предположений о том, что хотел сказать автор. Мы можем их получить.

1. Задаем количество желаемых гипотез.

   request.streaming_config.config.max_alternatives = 3

2. Адаптируем вывод результата.

   print("Alternatives: [")
   for alternative in result.recognition_result.alternatives:
       print('    "' + alternative.transcript + '",')
   print("]")
   print("------------------")

Исходный код примера

Пример 9: получаем промежуточные гипотезы

$ ./stt_streaming_recognize_interim_results.py

Так как сервис распознаёт текст на лету, можно получать промежуточные версии на лету. Чаще всего это применяется при надиктовывании текста в чат.

1. Включаем промежуточные гипотезы.

   request.streaming_config.interim_results_config.enable_interim_results = True

2. Выравниваем таймштампы.

   def format_time_stamp(timestamp):
       return "{:>02d}:{:>02d}:{:>02d}.{:>03d}".format(timestamp.seconds//(60*60), (timestamp.seconds//60)%60, timestamp.seconds%60, timestamp.microseconds//1000)
   
   def time_range(recognition_result):
       return "[" + format_time_stamp(recognition_result.start_time.ToTimedelta()) + " .. " + format_time_stamp(recognition_result.end_time.ToTimedelta()) + "]"

3. Теперь нам нужно различать финальные и промежуточные гипотезы — эту информацию мы возьмем из флага is_final. Также важно понимать, к какой фразе относится промежуточная гипотеза. Изобразим наглядно:

   if not inside_phrase:
        print("[Phrase begin]")
        inside_phrase = True
   assert(len(result.recognition_result.alternatives) == 1) # Handle carefully at real service
   if result.is_final:
        print("Final result:   " + time_range(result.recognition_result) + " \"" + result.recognition_result.alternatives[0].transcript + "\"")
        print("[Phrase end]")
        inside_phrase = False
   else:
        print("Interim result: " + time_range(result.recognition_result) + " \"" + result.recognition_result.alternatives[0].transcript + "\"")

Исходный код примера

Пример 10: кастомизируем VAD (Voice Activity Detection) для streaming_recognize

$ ./stt_streaming_recognize_vad_customization.py

Иногда нужно переопределить настройки определения завершения фразы — например, для изменения отзывчивости телефонного робота-опросника.

В реальном поведении также участвуют характеристики модели и внутреннее устройство сервиса.

Пример использования silence_min и silence_max:

request.streaming_config.config.vad_config.silence_prob_threshold = 0.2
request.streaming_config.config.vad_config.silence_min = 3.
request.streaming_config.config.vad_config.silence_max = 3.2

Исходный код примера

Пример 11: кастомизируем VAD (Voice Activity Detection) для recognize

$ ./stt_recognize_vad_customization.py

Иногда нужно переопределить настройки определения завершения фразы — например, для изменения отзывчивости телефонного робота-опросника.

В реальном поведении также участвуют характеристики модели и внутреннее устройство сервиса.

Пример использования silence_duration_threshold и silence_prob_threshold:

request.config.vad_config.silence_duration_threshold = 3.0
request.config.vad_config.silence_prob_threshold = 0.2

Исходный код примера

Пример 12: режим завершения после первой фразы

$ ./stt_streaming_recognize_single_utterance.py

Этот режим типичен в ситуациях, когда нет желания анализировать диалог целиком — хочется упростить логику сервиса. Например, в голосовом ассистенте: чтобы не следить за тем, что пользователь наговорил между запросами, а начать распознавать новый запрос с начала.

Выставляем соответствующий флаг:

request.streaming_config.single_utterance = True

Исходный код примера

Пример 13: выключаем VAD

$ ./stt_streaming_recognize_vad_disabled.py

Если выключить VAD, мы распознаем аудиофрагмент как одну фразу.

Выставляем соответствующий флаг:

request.streaming_config.config.do_not_perform_vad = True

Исходный код примера

Пример 14: включаем пунктуацию

$ ./stt_streaming_recognize_enable_automatic_punctuation.py

Воспользуемся автоматической пунктуацией сервиса.

Выставляем соответствующий флаг:

request.streaming_config.config.enable_automatic_punctuation = True

Исходный код примера

Пример 15: включаем фильтр ненормативной лексики

$ ./stt_streaming_recognize_profanity_filter.py

Чтобы закрыть нецезурные слова, воспользуемся фильтром ненормативной лексики.

Выставляем соответствующий флаг:

request.streaming_config.config.profanity_filter = True

Исходный код примера

Пример 16: задаём контекст

$ ./stt_streaming_recognize_context.py

Для задания контекста передаём соответствующие фразы.

Выставляем соответствующий флаг:

request.streaming_config.config.speech_contexts.append(stt_pb2.SpeechContext(phrases = [
    stt_pb2.SpeechContextPhrase(text = "мюллер", score = 10.0),
]))

В стриминговой сессии конфиг может быть передан повторно с целью изменения контекста — подробнее в исходном коде примера.

Пример 17: включаем определения пола спикера

$ ./stt_streaming_recognize_gender_identification.py

Выставляем соответствующий флаг:

request.streaming_config.config.enable_gender_identification = True

Исходный код примера

Пример 18: включаем определение эмоций

$ ./stt_recognize_sentiment_analysis.py

Выставляем соответствующий флаг:

request.config.enable_sentiment_analysis = True

Исходный код примера

Отложенное распознавание речи

Пример 19: загрузка аудио для отложенной обработки и получение результата через циклический опрос

$ ../stt_long_running_recognize_single_audio.py

Метод LongRunningRecognize принимает на вход параметры, аналогичные Recognize — но возвращает не результат распознавания, а идентификатор операции, за ходом обработки которой можно следить.

Этот режим может быть полезен для распознавания, когда результаты не нужны сразу.

Генерация запроса аналогична генерации RecognizeRequest — за исключением названия метода:

def build_recognize_request():
    request = stt_pb2.LongRunningRecognizeRequest()
    with open("../../audio/sample_3.s16", "rb") as f:
        request.audio.content = f.read()
    request.config.encoding = stt_pb2.AudioEncoding.LINEAR16
    request.config.sample_rate_hertz = 16000  # Not stored at raw ".s16" file
    request.config.num_channels = 1  # Not stored at raw ".s16" file
    return request

Метод LongRunningRecognize возвращает объект операции — это объект, который содержит информацию о текущем статусе обработки задачи.

Функция, которая показывает, как получить разную информацию о текущем состоянии операции:

def print_longrunning_operation(operation):
    print("Operation id:", operation.id)
    print("State:", OperationState.Name(operation.state))
    if operation.state == DONE:
        response = stt_pb2.RecognizeResponse()
        operation.response.Unpack(response)
        print_recognition_response(response)
    if operation.state == FAILED:
        print("Error:", operation.error)
    print("============================")

Для получения информации по статусам операций используется отдельный сервис — Operations. В метаданных указывается поле AUD, соответствующее этому сервису:

operations_stub = longrunning_pb2_grpc.OperationsStub(grpc.secure_channel(endpoint, grpc.ssl_channel_credentials()))
operations_metadata = authorization_metadata(api_key, secret_key, "tinkoff.cloud.longrunning")

Наивный способ дождаться завершения операции — запрашивать ее статус в цикле. Хотя этот метод не самый эффективный, он удобен для демонстрации работы системы:

1. Сгенерируем запрос.

   def build_get_operation_request(id):
       request = longrunning_pb2.GetOperationRequest()
       request.id = id
       return request

2. Отправим запрос.

   while operation.state != FAILED and operation.state != DONE:
       time.sleep(1)
       operation = operations_stub.GetOperation(build_get_operation_request(operation.id), metadata=operations_metadata)
       print_longrunning_operation(operation)

Исходный код примера

Пример 20: загрузка пачки аудио для отложенной обработки и получение результата через нотификации

$ ./stt_long_running_recognize_audio_group.py

Вместо того чтобы постоянно опрашивать API в цикле и запоминать идентификаторы каждой операции, можно объединить их в группы и получать результаты обработки через метод WatchOperations.

Чтобы избежать путаницы между наборами аудио, загружаемыми при повторных запусках, организуем их в группы.

1. В качестве названия группы можно использовать любую строку — например, текущую дату.

   group_name = datetime.now().strftime("test-group-%Y-%m-%d, %H:%M:%S")

2. При формировании запроса на распознавание нужно указать группу, в которую должна быть помещена операция.

   def build_recognize_request(file_path, group_name):
       request = stt_pb2.LongRunningRecognizeRequest()
       # Note: setting the group name here allows us filtering by group name in WatchOperations
       request.group = group_name
       with open(file_path, "rb") as f:
           request.audio.content = f.read()
       request.config.encoding = stt_pb2.AudioEncoding.LINEAR16
       request.config.sample_rate_hertz = 48000  # Not stored at raw ".s16" file
       request.config.num_channels = 1  # Not stored at raw ".s16" file
       return request

3. В качестве набора аудио будем использовать все файлы из каталога.

   audio_folder = "../../audio/sample_group"
   
   # Send audio files for recognition
   stt_stub = stt_pb2_grpc.SpeechToTextStub(grpc.secure_channel(endpoint, grpc.ssl_channel_credentials()))
   stt_metadata = authorization_metadata(api_key, secret_key, "tinkoff.cloud.stt")
   created_operations = 0
   for test_file in os.listdir(audio_folder):
       file_path = join(audio_folder, test_file)
       stt_stub.LongRunningRecognize(build_recognize_request(file_path, group_name), metadata=stt_metadata)
       created_operations += 1

4. Реализуем вспомогательные методы для вывода состояния операций в консоль.

   def print_longrunning_operations(operations):
       for operation in operations:
           print(f"[{operation.id}] {get_recognition_state_description(operation)}")
       print("============================")
   
   
   def get_recognition_state_description(operation):
       if operation.state == DONE:
           response = stt_pb2.RecognizeResponse()
           operation.response.Unpack(response)
   
           return " ".join([result.alternatives[0].transcript for result in response.results])
       if operation.state == FAILED:
           return operation.error
   
       return OperationState.Name(operation.state)

5. Подготовим запрос для вызова WatchOperations. Он содержит информацию о том, какие операции должны возвращаться, а также признак того, нужно ли после отправки текущего состояния операций переходить в режим уведомлений об изменениях.

   def build_watch_operations_request(group_name):
        request = longrunning_pb2.WatchOperationsRequest()
        request.filter.exact_group = group_name
        # Note: listen_for_updates is set to False by default.
        # Setting it to True here is required for update notifications to be sent.
        request.listen_for_updates = True
        return request

6. Вызываем WatchOperations, который позволяет осуществлять фильтрацию операций, входящих в определенную группу. Используем такой способ ожидания завершения обработки операций и обработки статусов:

    def count_finished_operations(operations):
        return sum([int(operation.state == DONE) for operation in operations])
   
    print(f"Watching operations in group '{group_name}'")
    responses = operations_stub.WatchOperations(
        build_watch_operations_request(group_name), metadata=operations_metadata
    )
    finished_operations = 0
    for response in responses:
        if response.HasField("initial_state"):
            finished_operations += count_finished_operations(
                response.initial_state.operations
            )
            print("WatchOperations. Initial state:")
            print_longrunning_operations(response.initial_state.operations)
        elif response.HasField("init_finished"):
            print("WatchOperations. Init finished.")
        else:
            assert response.HasField("update")
            finished_operations += count_finished_operations(response.update.operations)
            print("WatchOperations. Update:")
            print_longrunning_operations(response.update.operations)
    
        if finished_operations == created_operations:
            break
    print("Done.")

Исходный код примера

Пример 21: загрузка пачки аудио для отложенной обработки с идентификацией задания через x-client-request-id

$ ./stt_long_running_recognize_audio_group_x_client_request_id.py

В этом примере нас интересует возможность передавать и затем получать локальный идентификатор запроса. В нашем случае это имя файла без расширения, но также это могут быть ID записи в базе данных и другие варианты.

1. В запросе передаем имя файла без расширения в метаполе x-client-request-id.

   metadata = []
   for entry in stt_metadata:
       metadata.append(entry)
   # Passing filename without extension into "x-client-request-id":
   metadata.append(("x-client-request-id", os.path.basename(test_file)))
   stt_stub.LongRunningRecognize(build_recognize_request(file_path, group_name), metadata=metadata)

2. Получаем его через поле x_client_request_id операции и записываем в файл с соответствующим названием.

   def store_longrunning_operations(dname, operations):
       os.makedirs(dname, exist_ok=True)
       for operation in operations:
           description = get_recognition_state_description(operation)
           if description != None:
               # "x-client-request-id" specified at operation creation is passed here as is
               x_client_request_id = operation.x_client_request_id
               with open(dname + "/" + x_client_request_id + ".txt", "wb") as f:
                   f.write(bytes(f"[{operation.id}] {description}", 'utf-8'))

Исходный код примера