Асинхронный вызов API Python, дающий ошибку цикла событий ⇐ Python
-
Anonymous
Асинхронный вызов API Python, дающий ошибку цикла событий
Я размечаю файл паркета, вызывая тактильную модель, но это занимает много времени, поскольку файл паркета огромен. Поэтому я пытаюсь использовать асинхронную библиотеку Python asyncio для выполнения нескольких вызовов API и улучшения обработки.
Я пытаюсь запустить приведенный ниже скрипт, чтобы асинхронно выполнять вызовы API к тактильной модели: -
импортировать asyncio импортировать aiohttp импортировать панд как pd импортировать pyarrow.parquet как pq Защитные фрагменты (lst, n): для i в диапазоне (0, len(lst), n): выход lst[i:i + n] класс TaktileModel: def __init__(self, api_url, api_key): self.api_url = api_url self.api_key = api_key self.request_headers = { "Принять": "приложение/json", «X-Api-Key»: api_key, } def _prepare_request_data(self, df, version, text_col): данные = {} данные["message_bodies"] = список(df[text_col].values) req_data = {"данные": данные, "метаданные": {"версия": версия}} вернуть требуемые_данные async def _predict(self, df, concurrency, version, text_col): req_data = self._prepare_request_data(df, версия, text_col) асинхронно с session.post(self.api_url, headers=self.request_headers, json=req_data) в качестве ответа: ответ.raise_for_status() response_json = ожидайте ответа.json() вернуть ответ_json['данные']['метки'] асинхронная защита сбора_predict_tasks(self, df, text_col, версия): concurrency = asyncio.Semaphore(8) # Предполагается, что максимальное количество одновременных запросов равно 8 возвращение ждут asyncio.gather( *( self._predict(кусок, параллелизм, версия, text_col) for chunk in chunks(df, 128) # Предполагая, что размер пакета равен 128 ) ) def предсказать(self, df, text_col, version="v1"): df["метка"] = Нет asyncio.run(self.gather_predict_tasks(df, version, text_col)) df["label"] = df["label"].apply(label.decode) возврат DF защита основной(): api_url = "Masked_api_url" api_key = "Маскированный_api_key" taktile_model = TaktileModel(api_url, api_key) исправленный_hashed_dataset = pq.ParquetDataset('Masked_Path/to/parquet/file') исправленный_хэшед_df = исправленный_хэшированный_dataset.read().to_pandas() версия = "v1" text_col = "тело_сообщения" результаты = taktile_model.predict(corrected_hashed_df, версия, text_col) распечатать(результаты) если __name__ == "__main__": основной() Но я получаю эту ошибку для приведенного выше кода: -
`" -------------------------------------------------- ------------------------- RuntimeError Traceback (самый последний вызов — последний) Ячейка In[1], строка 60 58 # Проверьте, запускается ли скрипт напрямую 59, если __name__ == "__main__": ---> 60 основных() Ячейка In[1], строка 54, в main() 51 text_col = "тело_сообщения" 53 # Прогнозирование с использованием модели Taktile и обновление DataFrame с помощью меток ---> 54 результата = taktile_model.predict(corrected_hashed_df, version, text_col) 56 печать(результаты) Ячейка In[1], строка 32, в TaktileModel.predict(self, df, text_col, version) 30 def предсказать(self, df, text_col, version="v1"): 31 df["метка"] = Нет ---> 32 asyncio.run(self.gather_predict_tasks(df, version, text_col)) 33 df["label"] = df["label"].apply(label.decode) 35 возврат DF Файл ~/SageMaker/.conda/mambaforge/envs/branch/lib/python3.9/asyncio/runners.py:33, в запуске (основной, отладка) 9 """Выполнить сопрограмму и вернуть результат. 10 11 Эта функция запускает переданную сопрограмму, заботясь о (...) 30 asyncio.run(main()) 31 """ 32, если event._get_running_loop() не имеет значения None: ---> 33 поднять RuntimeError( 34 «asyncio.run() нельзя вызвать из работающего цикла событий») 36, если не coroutines.iscoroutine(main): 37 поднять ValueError("ожидалась сопрограмма, получена {!r}".format(main)) RuntimeError: asyncio.run() нельзя вызвать из текущего цикла событий. "` Я также пытался сделать функцию прогнозирования асинхронной, но она выдала ту же ошибку, связанную с циклами событий, говоря, что RuntimeError: asyncio.run() нельзя вызвать из работающего цикла событий
Я размечаю файл паркета, вызывая тактильную модель, но это занимает много времени, поскольку файл паркета огромен. Поэтому я пытаюсь использовать асинхронную библиотеку Python asyncio для выполнения нескольких вызовов API и улучшения обработки.
Я пытаюсь запустить приведенный ниже скрипт, чтобы асинхронно выполнять вызовы API к тактильной модели: -
импортировать asyncio импортировать aiohttp импортировать панд как pd импортировать pyarrow.parquet как pq Защитные фрагменты (lst, n): для i в диапазоне (0, len(lst), n): выход lst[i:i + n] класс TaktileModel: def __init__(self, api_url, api_key): self.api_url = api_url self.api_key = api_key self.request_headers = { "Принять": "приложение/json", «X-Api-Key»: api_key, } def _prepare_request_data(self, df, version, text_col): данные = {} данные["message_bodies"] = список(df[text_col].values) req_data = {"данные": данные, "метаданные": {"версия": версия}} вернуть требуемые_данные async def _predict(self, df, concurrency, version, text_col): req_data = self._prepare_request_data(df, версия, text_col) асинхронно с session.post(self.api_url, headers=self.request_headers, json=req_data) в качестве ответа: ответ.raise_for_status() response_json = ожидайте ответа.json() вернуть ответ_json['данные']['метки'] асинхронная защита сбора_predict_tasks(self, df, text_col, версия): concurrency = asyncio.Semaphore(8) # Предполагается, что максимальное количество одновременных запросов равно 8 возвращение ждут asyncio.gather( *( self._predict(кусок, параллелизм, версия, text_col) for chunk in chunks(df, 128) # Предполагая, что размер пакета равен 128 ) ) def предсказать(self, df, text_col, version="v1"): df["метка"] = Нет asyncio.run(self.gather_predict_tasks(df, version, text_col)) df["label"] = df["label"].apply(label.decode) возврат DF защита основной(): api_url = "Masked_api_url" api_key = "Маскированный_api_key" taktile_model = TaktileModel(api_url, api_key) исправленный_hashed_dataset = pq.ParquetDataset('Masked_Path/to/parquet/file') исправленный_хэшед_df = исправленный_хэшированный_dataset.read().to_pandas() версия = "v1" text_col = "тело_сообщения" результаты = taktile_model.predict(corrected_hashed_df, версия, text_col) распечатать(результаты) если __name__ == "__main__": основной() Но я получаю эту ошибку для приведенного выше кода: -
`" -------------------------------------------------- ------------------------- RuntimeError Traceback (самый последний вызов — последний) Ячейка In[1], строка 60 58 # Проверьте, запускается ли скрипт напрямую 59, если __name__ == "__main__": ---> 60 основных() Ячейка In[1], строка 54, в main() 51 text_col = "тело_сообщения" 53 # Прогнозирование с использованием модели Taktile и обновление DataFrame с помощью меток ---> 54 результата = taktile_model.predict(corrected_hashed_df, version, text_col) 56 печать(результаты) Ячейка In[1], строка 32, в TaktileModel.predict(self, df, text_col, version) 30 def предсказать(self, df, text_col, version="v1"): 31 df["метка"] = Нет ---> 32 asyncio.run(self.gather_predict_tasks(df, version, text_col)) 33 df["label"] = df["label"].apply(label.decode) 35 возврат DF Файл ~/SageMaker/.conda/mambaforge/envs/branch/lib/python3.9/asyncio/runners.py:33, в запуске (основной, отладка) 9 """Выполнить сопрограмму и вернуть результат. 10 11 Эта функция запускает переданную сопрограмму, заботясь о (...) 30 asyncio.run(main()) 31 """ 32, если event._get_running_loop() не имеет значения None: ---> 33 поднять RuntimeError( 34 «asyncio.run() нельзя вызвать из работающего цикла событий») 36, если не coroutines.iscoroutine(main): 37 поднять ValueError("ожидалась сопрограмма, получена {!r}".format(main)) RuntimeError: asyncio.run() нельзя вызвать из текущего цикла событий. "` Я также пытался сделать функцию прогнозирования асинхронной, но она выдала ту же ошибку, связанную с циклами событий, говоря, что RuntimeError: asyncio.run() нельзя вызвать из работающего цикла событий
