NLP(六十五)LangChain中的重連(retry)機(jī)制
??關(guān)于LangChain入門(mén)，讀者可參考文章NLP（五十六）LangChain入門(mén) 。
??本文將會(huì)介紹LangChain中的重連機(jī)制，并嘗試給出定制化重連方案。
??本文以LangChain中的對(duì)話功能（ChatOpenAI）為例。
LangChain中的重連機(jī)制
??查看LangChain中對(duì)話功能（ChatOpenAI）的重連機(jī)制（retry），其源代碼如下：
class ChatOpenAI(BaseChatModel):
...
def _create_retry_decorator(self) -> Callable[[Any], Any]:
import openai
min_seconds = 1
max_seconds = 60
# Wait 2^x * 1 second between each retry starting with
# 4 seconds, then up to 10 seconds, then 10 seconds afterwards
return retry(
reraise=True,
stop=stop_after_attempt(self.max_retries),
wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds),
retry=(
retry_if_exception_type(openai.error.Timeout)
| retry_if_exception_type(openai.error.APIError)
| retry_if_exception_type(openai.error.APIConnectionError)
| retry_if_exception_type(openai.error.RateLimitError)
| retry_if_exception_type(openai.error.ServiceUnavailableError)
),
before_sleep=before_sleep_log(logger, logging.WARNING),
)
def completion_with_retry(self, **kwargs: Any) -> Any:
"""Use tenacity to retry the completion call."""
retry_decorator = self._create_retry_decorator()
@retry_decorator
def _completion_with_retry(**kwargs: Any) -> Any:
return self.client.create(**kwargs)
return _completion_with_retry(**kwargs)
可以看到，其編碼方式為硬編碼（hardcore），采用tenacity模塊實(shí)現(xiàn)重連機(jī)制，對(duì)于支持的報(bào)錯(cuò)情形，比如openai.error.Timeout, openai.error.APIError等，會(huì)嘗試重連，最小等待時(shí)間為1s，最大等待時(shí)間為60s，每次重連等待時(shí)間會(huì)乘以2。
簡(jiǎn)單重連
??我們嘗試用一個(gè)錯(cuò)誤的OpenAI key進(jìn)行對(duì)話，代碼如下：
from langchain.chat_models import ChatOpenAI
def chat_bot(input_text: str):
llm = ChatOpenAI(temperature=0,
model_name="gpt-3.5-turbo",
openai_api_key="sk-xxx",
max_retries=5)
return llm.predict(input_text)
if __name__ == '__main__':
text = '中國(guó)的首都是哪里？'
print(chat_bot(text))
盡管我們?cè)诖a中設(shè)置了重連最大次數(shù)（max_retries），代碼運(yùn)行時(shí)會(huì)直接報(bào)錯(cuò)，不會(huì)重連，原因是LangChain中的對(duì)話功能重連機(jī)制沒(méi)有支持openai.error.AuthenticationError。輸出結(jié)果如下：
openai.error.AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys.
??此時(shí)，我們嘗試在源代碼的基礎(chǔ)上做簡(jiǎn)單的定制，使得其支持openai.error.AuthenticationError錯(cuò)誤類型，代碼如下：
# -*- coding: utf-8 -*-
import openai
from typing import Callable, Any
from tenacity import (
before_sleep_log,
retry,
retry_if_exception_type,
stop_after_attempt,
wait_exponential,
)
from langchain.chat_models import ChatOpenAI
import logging
logger = logging.getLogger(__name__)
class MyChatOpenAI(ChatOpenAI):
def _create_retry_decorator(self) -> Callable[[Any], Any]:
min_seconds = 1
max_seconds = 60
# Wait 2^x * 1 second between each retry starting with
# 4 seconds, then up to 10 seconds, then 10 seconds after wards
return retry(
reraise=True,
stop=stop_after_attempt(self.max_retries),
wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds),
retry=(
retry_if_exception_type(openai.error.Timeout)
| retry_if_exception_type(openai.error.APIError)
| retry_if_exception_type(openai.error.APIConnectionError)
| retry_if_exception_type(openai.error.RateLimitError)
| retry_if_exception_type(openai.error.ServiceUnavailableError)
# add new error
| retry_if_exception_type(openai.error.AuthenticationError)
),
before_sleep=before_sleep_log(logger, logging.WARNING),
)
def completion_with_retry(self, **kwargs: Any) -> Any:
"""Use tenacity to retry the completion call."""
retry_decorator = self._create_retry_decorator()
@retry_decorator
def _completion_with_retry(**kwargs: Any) -> Any:
return self.client.create(**kwargs)
return _completion_with_retry(**kwargs)
def chat_bot(input_text: str):
llm = MyChatOpenAI(temperature=0,
model_name="gpt-3.5-turbo",
openai_api_key="sk-xxx",
max_retries=5)
return llm.predict(input_text)
if __name__ == '__main__':
text = '中國(guó)的首都是哪里？'
print(chat_bot(text))
分析上述代碼，我們?cè)诶^承ChatOpenAI類的基礎(chǔ)上重新創(chuàng)建MyChatOpenAI類，在_create_retry_decorator中的重連錯(cuò)誤情形中加入了openai.error.AuthenticationError錯(cuò)誤類型，此時(shí)代碼輸出結(jié)果如下：
Retrying __main__.MyChatOpenAI.completion_with_retry.<locals>._completion_with_retry in 1.0 seconds as it raised AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys..
Retrying __main__.MyChatOpenAI.completion_with_retry.<locals>._completion_with_retry in 2.0 seconds as it raised AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys..
Retrying __main__.MyChatOpenAI.completion_with_retry.<locals>._completion_with_retry in 4.0 seconds as it raised AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys..
Retrying __main__.MyChatOpenAI.completion_with_retry.<locals>._completion_with_retry in 8.0 seconds as it raised AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys..
Traceback (most recent call last):
......
openai.error.AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys.
從輸出結(jié)果中，我們可以看到，該代碼確實(shí)對(duì)openai.error.AuthenticationError錯(cuò)誤類型進(jìn)行了重連，按照源代碼的方式進(jìn)行重連，一共嘗試了5次重連，每次重連等待時(shí)間是上一次的兩倍。
定制化重連
??LangChain中的重連機(jī)制也支持定制化。
??假設(shè)我們的使用場(chǎng)景：某個(gè)OpenAI key在調(diào)用過(guò)程中失效了，那么在重連時(shí)希望能快速切換至某個(gè)能正常使用的OpenAI key，以下為示例代碼（僅需要修改completion_with_retry函數(shù)）：
def completion_with_retry(self, **kwargs: Any) -> Any:
"""Use tenacity to retry the completion call."""
retry_decorator = self._create_retry_decorator()
@retry_decorator
def _completion_with_retry(**kwargs: Any) -> Any:
# 重連機(jī)制定制化(custom retry)
kwargs['api_key'] = 'right openai key'
return self.client.create(**kwargs)
return _completion_with_retry(**kwargs)
此時(shí)就能進(jìn)行正常的對(duì)話功能了。
總結(jié)
??本文介紹了LangChain中的重連機(jī)制，并嘗試給出定制化重連方案，希望能對(duì)讀者有所幫助。
??筆者的個(gè)人博客網(wǎng)址為：https://percent4.github.io/ ,歡迎大家訪問(wèn)~