谷歌翻译 api 超时

Question

我有大约 20000 篇文本要翻译，每篇平均长度约为 100 个字符。我正在使用多处理库来加速我的 API 调用。如下所示：

from google.cloud.translate_v2 import Client
from time import sleep
from tqdm.notebook import tqdm
import multiprocessing as mp

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = cred_file
translate_client = Client()

def trans(text, MAX_TRIES=5):
    res = None
    sleep_time = 1
    for i in range(MAX_TRIES):
        try:
            res = translate_client.translate(text, target_language="en", model="nmt")
            error = None
        except Exception as error:
            pass

        if res is None:
            sleep(sleep_time)  # wait for 1 seconds before trying to fetch the data again
            sleep_time *= 2
        else:
            break

    return res["translatedText"]

src_text = # eg. ["this is a sentence"]*20000
with mp.Pool(mp.cpu_count()) as pool:
    translated = list(tqdm(pool.imap(trans, src_text), total=len(src_text)))

不幸的是，上面的代码每次在迭代 2828 +/- 5 时都会失败 (HTTP Error 503: Service Unavailable)。我希望具有可变的睡眠时间可以让它重新启动并正常运行。奇怪的是，如果我要立即重新启动循环，它会毫无问题地再次启动，即使自代码完成执行以来已经过去了

1. 我做错了try/except 吗？
2. 进行多处理会以某种方式影响 API。
3. 一般想法？

我需要多处理，否则我将等待大约 3 个小时才能完成整个过程。

Answer 1

一些想法，之前尝试过的google API，只能处理一定数量的并发请求，如果达到限制，服务会返回错误HTTP 503“Service Unavailable”。如果是Daily limit is Exceeded 或User Rate Limit，则为HTTP 403。

尝试使用指数退避实现重试。以指数级增长的等待时间重试操作，已达到最大重试次数。它将提高带宽使用率并最大限度地提高并发环境中的请求吞吐量。

并查看配额和限制page。

• Exponential backoff

Answer 2

503 错误表明此问题出在 Google 一方，这让我相信您可能会受到速率限制。正如 Raphael 提到的，响应中是否有 Retry-After 标头？我建议查看响应标头，因为它可能会更具体地告诉您发生了什么，并可能为您提供有关如何修复它的信息。

Answer 3

Google API 擅长隐藏执行 Google 翻译的复杂性。不幸的是，如果您进入 Google API 代码，它使用的是标准 HTTP 请求。这意味着当您运行 20, 000 多个请求时，无论线程池如何，都会出现巨大的瓶颈。

考虑使用aiohttp（您需要从pip 安装）和asyncio 创建HTTP 请求。这将允许您运行异步 HTTP 请求。 （这意味着您不需要使用 google.cloud.translate_v2、multiprocessing 或 tqdm.notebook）。

只需在asyncio.run()中调用一个await方法，该方法就可以创建一个方法数组来执行aiohttp.session.get()。然后调用 asyncio.gather() 收集所有结果。

在下面的示例中，我使用 API 密钥 https://console.cloud.google.com/apis/credentials（而不是 Google 应用程序凭据/服务帐户）。

将您的示例与 asyncio 和 aiohttp 一起使用，它在 30 秒内运行并且没有任何错误。 （尽管您可能希望将超时延长到会话）。

值得指出的是，Google 的限制是每分钟 600 万 个字符。您的测试正在进行 360,000。因此，如果您在一分钟内运行 17 次测试，您将达到极限！

另外，速度主要由机器决定，而不是 Google API。 （我在一台 3GHz、8 核和 16GB 内存的电脑上进行了测试）。

import asyncio
import aiohttp
from collections import namedtuple
import json
from urllib.parse import quote

TranslateReponseModel = namedtuple('TranslateReponseModel', ['sourceText', 'translatedText', 'detectedSourceLanguage']) # model to store results.

def Logger(json_message):    
    print(json.dumps(json_message)) # Note: logging json is just my personal preference.

async def DownloadString(session, url, index):
    while True: # If client error - this will retry. You may want to limit the amount of attempts
        try:
            r = await session.get(url)
            text = await r.text()
            #Logger({"data": html, "status": r.status}) 
            r.raise_for_status() # This will error if API return 4xx or 5xx status.
            return text
        except aiohttp.ClientConnectionError as e:
            Logger({'Exception': f"Index {index} - connection was dropped before we finished", 'Details': str(e), 'Url': url })
        except aiohttp.ClientError as e:
            Logger({'Exception': f"Index {index} - something went wrong. Not a connection error, that was handled", 'Details': str(e), 'Url': url})


def FormatResponse(sourceText, responseText):
    jsonResponse = json.loads(responseText)
    return TranslateReponseModel(sourceText, jsonResponse["data"]["translations"][0]["translatedText"], jsonResponse["data"]["translations"][0]["detectedSourceLanguage"])

def TranslatorUriBuilder(targetLanguage, sourceText):
    apiKey = 'ABCDED1234' # TODO This is a 41 characters API Key. You'll need to generate one (it's not part of the json certificate)
    return f"https://translation.googleapis.com/language/translate/v2?key={apiKey}={quote(sourceText)}&target={targetLanguage}"

async def Process(session, sourceText, lineNumber):
    translateUri = TranslatorUriBuilder('en', sourceText) # Country code is set to en (English)
    translatedResponseText = await DownloadString(session, translateUri, lineNumber)
    response = FormatResponse(sourceText, translatedResponseText)
    return response

async def main():       
    statements = ["this is another sentence"]*20000

    Logger({'Message': f'Start running Google Translate API for {len(statements)}'})
    results = []
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[Process(session, val, idx) for idx, val in enumerate(statements)]  )  

    Logger({'Message': f'Results are: {", ".join(map(str, [x.translatedText for x in results]))}'})
    Logger({'Message': f'Finished running Google Translate API for {str(len(statements))} and got {str(len(results))} results'})

if __name__ == '__main__':
    asyncio.run(main())

附加测试

初始测试运行相同的翻译。因此，我创建了一个测试来检查结果是否未缓存在 Google 上。我手动将电子书复制到文本文件中。然后在 Python 中，代码打开文件并将文本分组为 100 个字符的数组，然后从数组中取出前 20,000 个项目并翻译每一行。有趣的是，它仍然用了不到 30 秒。

import asyncio
import aiohttp
from collections import namedtuple
import json
from urllib.parse import quote

TranslateReponseModel = namedtuple('TranslateReponseModel', ['sourceText', 'translatedText', 'detectedSourceLanguage']) # model to store results.

def Logger(json_message):    
    print(json.dumps(json_message)) # Note: logging json is just my personal preference.

async def DownloadString(session, url, index):
    while True: # If client error - this will retry. You may want to limit the amount of attempts
        try:
            r = await aiohttp.session.get(url)
            text = await r.text()
            #Logger({"data": html, "status": r.status}) 
            r.raise_for_status() # This will error if API return 4xx or 5xx status.
            return text
        except aiohttp.ClientConnectionError as e:
            Logger({'Exception': f"Index {index} - connection was dropped before we finished", 'Details': str(e), 'Url': url })
        except aiohttp.ClientError as e:
            Logger({'Exception': f"Index {index} - something went wrong. Not a connection error, that was handled", 'Details': str(e), 'Url': url})


def FormatResponse(sourceText, responseText):
    jsonResponse = json.loads(responseText)
    return TranslateReponseModel(sourceText, jsonResponse["data"]["translations"][0]["translatedText"], jsonResponse["data"]["translations"][0]["detectedSourceLanguage"])

def TranslatorUriBuilder(targetLanguage, sourceText):
    apiKey = 'ABCDED1234' # TODO This is a 41 characters API Key. You'll need to generate one (it's not part of the json certificate)
    return f"https://translation.googleapis.com/language/translate/v2?key={apiKey}={quote(sourceText)}&target={targetLanguage}"

async def Process(session, sourceText, lineNumber):
    translateUri = TranslatorUriBuilder('en', sourceText) # Country code is set to en (English)
    translatedResponseText = await DownloadString(session, translateUri, lineNumber)
    response = FormatResponse(sourceText, translatedResponseText)
    return response

def readEbook():
    # This is a simple test to make sure response is not cached.
    # I grabbed a random online pdf (http://sd.blackball.lv/library/Beginning_Software_Engineering_(2015).pdf) and copied text into notepad.
    with open("C:\\Dev\\ebook.txt", "r", encoding="utf8") as f:
        return f.read()

def chunkText(text):
    chunk_size = 100
    chunks= len(text)
    chunk_array = [text[i:i+chunk_size] for i in range(0, chunks, chunk_size)]
    formatResults = [x for x in chunk_array if len(x) > 10]
    return formatResults[:20000]

async def main():  
    data = readEbook()
    chunk_data = chunkText(data)
    
    Logger({'Message': f'Start running Google Translate API for {len(chunk_data)}'})
    results = []
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[Process(session, val, idx) for idx, val in enumerate(chunk_data)]  )  

    Logger({'Message': f'Results are: {", ".join(map(str, [x.translatedText for x in results]))}'})
    Logger({'Message': f'Finished running Google Translate API for {str(len(chunk_data))} and got {str(len(results))} results'})

if __name__ == '__main__':
    asyncio.run(main())

最后，您可以找到有关 Google Translate API HTTP 请求 https://cloud.google.com/translate/docs/reference/rest/v2/translate 的更多信息，您可以通过 Postman 运行该请求。