"""提示层包装器。"""
import datetime
from typing import Any, Dict, List, Optional
from langchain_core.callbacks import (
AsyncCallbackManagerForLLMRun,
CallbackManagerForLLMRun,
)
from langchain_core.messages import BaseMessage
from langchain_core.outputs import ChatResult
from langchain_community.chat_models import ChatOpenAI
[docs]class PromptLayerChatOpenAI(ChatOpenAI):
"""`PromptLayer` 和 `OpenAI` 聊天大语言模型 API。
要使用，您应该已安装 ``openai`` 和 ``promptlayer`` python
包，并设置环境变量 ``OPENAI_API_KEY``
和 ``PROMPTLAYER_API_KEY`` 分别为您的 openAI API 密钥和
promptlayer 密钥。
所有可以传递给 OpenAI LLM 的参数也可以
在这里传递。PromptLayerChatOpenAI 添加了可选的
参数:
``pl_tags``: 用于标记请求的字符串列表。
``return_pl_id``: 如果为 True，则 PromptLayer 请求 ID 将
在 ``Generation`` 对象的 ``generation_info`` 字段中返回。
示例:
.. code-block:: python
from langchain_community.chat_models import PromptLayerChatOpenAI
openai = PromptLayerChatOpenAI(model="gpt-3.5-turbo")
"""
pl_tags: Optional[List[str]]
return_pl_id: Optional[bool] = False
[docs] @classmethod
def is_lc_serializable(cls) -> bool:
return False
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
stream: Optional[bool] = None,
**kwargs: Any,
) -> ChatResult:
"""调用ChatOpenAI生成，然后调用PromptLayer API记录请求。"""
from promptlayer.utils import get_api_key, promptlayer_api_request
request_start_time = datetime.datetime.now().timestamp()
generated_responses = super()._generate(
messages, stop, run_manager, stream=stream, **kwargs
)
request_end_time = datetime.datetime.now().timestamp()
message_dicts, params = super()._create_message_dicts(messages, stop)
for i, generation in enumerate(generated_responses.generations):
response_dict, params = super()._create_message_dicts(
[generation.message], stop
)
params = {**params, **kwargs}
pl_request_id = promptlayer_api_request(
"langchain.PromptLayerChatOpenAI",
"langchain",
message_dicts,
params,
self.pl_tags,
response_dict,
request_start_time,
request_end_time,
get_api_key(),
return_pl_id=self.return_pl_id,
)
if self.return_pl_id:
if generation.generation_info is None or not isinstance(
generation.generation_info, dict
):
generation.generation_info = {}
generation.generation_info["pl_request_id"] = pl_request_id
return generated_responses
async def _agenerate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
stream: Optional[bool] = None,
**kwargs: Any,
) -> ChatResult:
"""调用ChatOpenAI生成，然后调用PromptLayer记录。"""
from promptlayer.utils import get_api_key, promptlayer_api_request_async
request_start_time = datetime.datetime.now().timestamp()
generated_responses = await super()._agenerate(
messages, stop, run_manager, stream=stream, **kwargs
)
request_end_time = datetime.datetime.now().timestamp()
message_dicts, params = super()._create_message_dicts(messages, stop)
for i, generation in enumerate(generated_responses.generations):
response_dict, params = super()._create_message_dicts(
[generation.message], stop
)
params = {**params, **kwargs}
pl_request_id = await promptlayer_api_request_async(
"langchain.PromptLayerChatOpenAI.async",
"langchain",
message_dicts,
params,
self.pl_tags,
response_dict,
request_start_time,
request_end_time,
get_api_key(),
return_pl_id=self.return_pl_id,
)
if self.return_pl_id:
if generation.generation_info is None or not isinstance(
generation.generation_info, dict
):
generation.generation_info = {}
generation.generation_info["pl_request_id"] = pl_request_id
return generated_responses
@property
def _llm_type(self) -> str:
return "promptlayer-openai-chat"
@property
def _identifying_params(self) -> Dict[str, Any]:
return {
**super()._identifying_params,
"pl_tags": self.pl_tags,
"return_pl_id": self.return_pl_id,
}