??關于LangChain入門，讀者可參考文章NLP（五十六）LangChain入門 。
??本文將會介紹LangChain中的重連機制，并嘗試給出定制化重連方案。
??本文以LangChain中的對話功能（ChatOpenAI）為例。

LangChain中的重連機制

??查看LangChain中對話功能（ChatOpenAI）的重連機制（retry），其源代碼如下：

class ChatOpenAI(BaseChatModel):...def _create_retry_decorator(self) -> Callable[[Any], Any]:import openaimin_seconds = 1max_seconds = 60# Wait 2^x * 1 second between each retry starting with# 4 seconds, then up to 10 seconds, then 10 seconds afterwardsreturn 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_decoratordef _completion_with_retry(**kwargs: Any) -> Any:return self.client.create(**kwargs)return _completion_with_retry(**kwargs)

可以看到，其編碼方式為硬編碼（hardcore），采用tenacity模塊實現重連機制，對于支持的報錯情形，比如openai.error.Timeout, openai.error.APIError等，會嘗試重連，最小等待時間為1s，最大等待時間為60s，每次重連等待時間會乘以2。

簡單重連

??我們嘗試用一個錯誤的OpenAI key進行對話，代碼如下：

from langchain.chat_models import ChatOpenAIdef 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 = '中國的首都是哪里？'print(chat_bot(text))

盡管我們在代碼中設置了重連最大次數（max_retries），代碼運行時會直接報錯，不會重連，原因是LangChain中的對話功能重連機制沒有支持openai.error.AuthenticationError。輸出結果如下：

openai.error.AuthenticationError: Incorrect API key provided: sk-xxx. You can find your API key at https://platform.openai.com/account/api-keys.

??此時，我們嘗試在源代碼的基礎上做簡單的定制，使得其支持openai.error.AuthenticationError錯誤類型，代碼如下：

# -*- 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 logginglogger = logging.getLogger(__name__)class MyChatOpenAI(ChatOpenAI):def _create_retry_decorator(self) -> Callable[[Any], Any]:min_seconds = 1max_seconds = 60# Wait 2^x * 1 second between each retry starting with# 4 seconds, then up to 10 seconds, then 10 seconds after wardsreturn 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_decoratordef _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 = '中國的首都是哪里？'print(chat_bot(text))

分析上述代碼，我們在繼承ChatOpenAI類的基礎上重新創建MyChatOpenAI類，在_create_retry_decorator中的重連錯誤情形中加入了openai.error.AuthenticationError錯誤類型，此時代碼輸出結果如下：

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.

從輸出結果中，我們可以看到，該代碼確實對openai.error.AuthenticationError錯誤類型進行了重連，按照源代碼的方式進行重連，一共嘗試了5次重連，每次重連等待時間是上一次的兩倍。

定制化重連

??LangChain中的重連機制也支持定制化。
??假設我們的使用場景：某個OpenAI key在調用過程中失效了，那么在重連時希望能快速切換至某個能正常使用的OpenAI key，以下為示例代碼（僅需要修改completion_with_retry函數）：

    def completion_with_retry(self, **kwargs: Any) -> Any:"""Use tenacity to retry the completion call."""retry_decorator = self._create_retry_decorator()@retry_decoratordef _completion_with_retry(**kwargs: Any) -> Any:# 重連機制定制化(custom retry)kwargs['api_key'] = 'right openai key'return self.client.create(**kwargs)return _completion_with_retry(**kwargs)

此時就能進行正常的對話功能了。

總結

??本文介紹了LangChain中的重連機制，并嘗試給出定制化重連方案，希望能對讀者有所幫助。
??筆者的個人博客網址為：https://percent4.github.io/ ,歡迎大家訪問~