mirror of
https://github.com/hwchase17/langchain.git
synced 2025-06-24 15:43:54 +00:00
community: fix some features on Naver ChatModel & embedding model 2 (#29243)
## Description - Responding to `NCP API Key` changes. - To fix `ChatClovaX` `astream` function to raise `SSEError` when an error event occurs. - To add `token length` and `ai_filter` to ChatClovaX's `response_metadata`. - To update document for apply NCP API Key changes. cc. @efriis @vbarda
This commit is contained in:
CLOVA Studio 개발
GitHub
parent
5d64597490
commit
7a95ffc775
@ -35,25 +35,24 @@
|
||||
"\n",
|
||||
"## Setup\n",
|
||||
"\n",
|
||||
"Before using the chat model, you must go through the three steps below.\n",
|
||||
"Before using the chat model, you must go through the four steps below.\n",
|
||||
"\n",
|
||||
"1. Creating [NAVER Cloud Platform](https://www.ncloud.com/) account \n",
|
||||
"2. Apply to use [CLOVA Studio](https://www.ncloud.com/product/aiService/clovaStudio)\n",
|
||||
"3. Find API Keys after creating CLOVA Studio Test App or Service App (See [here](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#테스트앱생성).)\n",
|
||||
"3. Create a CLOVA Studio Test App or Service App of a model to use (See [here](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#테스트앱생성).)\n",
|
||||
"4. Issue a Test or Service API key (See [here](https://api.ncloud-docs.com/docs/ai-naver-clovastudio-summary#API%ED%82%A4).)\n",
|
||||
"\n",
|
||||
"### Credentials\n",
|
||||
"\n",
|
||||
"CLOVA Studio requires 2 keys (`NCP_CLOVASTUDIO_API_KEY` and `NCP_APIGW_API_KEY`).\n",
|
||||
" - `NCP_CLOVASTUDIO_API_KEY` is issued per Test App or Service App\n",
|
||||
" - `NCP_APIGW_API_KEY` is issued per account, could be optional depending on the region you are using\n",
|
||||
"\n",
|
||||
"The two API Keys could be found by clicking `App Request Status` > `Service App, Test App List` > `‘Details’ button for each app` in [CLOVA Studio](https://clovastudio.ncloud.com/studio-application/service-app)\n",
|
||||
"Set the `NCP_CLOVASTUDIO_API_KEY` environment variable with your API key.\n",
|
||||
" - Note that if you are using a legacy API Key (that doesn't start with `nv-*` prefix), you might need to get an additional API Key by clicking `App Request Status` > `Service App, Test App List` > `‘Details’ button for each app` in [CLOVA Studio](https://clovastudio.ncloud.com/studio-application/service-app) and set it as `NCP_APIGW_API_KEY`.\n",
|
||||
"\n",
|
||||
"You can add them to your environment variables as below:\n",
|
||||
"\n",
|
||||
"``` bash\n",
|
||||
"export NCP_CLOVASTUDIO_API_KEY=\"your-api-key-here\"\n",
|
||||
"export NCP_APIGW_API_KEY=\"your-api-key-here\"\n",
|
||||
"# Uncomment below to use a legacy API key\n",
|
||||
"# export NCP_APIGW_API_KEY=\"your-api-key-here\"\n",
|
||||
"```"
|
||||
]
|
||||
},
|
||||
@ -71,10 +70,11 @@
|
||||
" os.environ[\"NCP_CLOVASTUDIO_API_KEY\"] = getpass.getpass(\n",
|
||||
" \"Enter your NCP CLOVA Studio API Key: \"\n",
|
||||
" )\n",
|
||||
"if not os.getenv(\"NCP_APIGW_API_KEY\"):\n",
|
||||
" os.environ[\"NCP_APIGW_API_KEY\"] = getpass.getpass(\n",
|
||||
" \"Enter your NCP API Gateway API key: \"\n",
|
||||
" )"
|
||||
"# Uncomment below to use a legacy API key\n",
|
||||
"# if not os.getenv(\"NCP_APIGW_API_KEY\"):\n",
|
||||
"# os.environ[\"NCP_APIGW_API_KEY\"] = getpass.getpass(\n",
|
||||
"# \"Enter your NCP API Gateway API key: \"\n",
|
||||
"# )"
|
||||
]
|
||||
},
|
||||
{
|
||||
@ -340,7 +340,7 @@
|
||||
"\n",
|
||||
"When going live with production-level application using CLOVA Studio, you should apply for and use Service App. (See [here](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#서비스앱신청).)\n",
|
||||
"\n",
|
||||
"For a Service App, a corresponding `NCP_CLOVASTUDIO_API_KEY` is issued and can only be called with it."
|
||||
"For a Service App, you should use a corresponding Service API key and can only be called with it."
|
||||
]
|
||||
},
|
||||
{
|
||||
@ -353,7 +353,7 @@
|
||||
"# Update environment variables\n",
|
||||
"\n",
|
||||
"os.environ[\"NCP_CLOVASTUDIO_API_KEY\"] = getpass.getpass(\n",
|
||||
" \"Enter NCP CLOVA Studio API Key for Service App: \"\n",
|
||||
" \"Enter NCP CLOVA Studio Service API Key: \"\n",
|
||||
")"
|
||||
]
|
||||
},
|
||||
|
||||
@ -10,7 +10,8 @@ Please refer to [NCP User Guide](https://guide.ncloud-docs.com/docs/clovastudio-
|
||||
|
||||
## Installation and Setup
|
||||
|
||||
- Get both CLOVA Studio API Key and API Gateway Key by [creating your app](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#create-test-app) and set them as environment variables respectively (`NCP_CLOVASTUDIO_API_KEY`, `NCP_APIGW_API_KEY`).
|
||||
- Get a CLOVA Studio API Key by [issuing it](https://api.ncloud-docs.com/docs/ai-naver-clovastudio-summary#API%ED%82%A4) and set it as an environment variable (`NCP_CLOVASTUDIO_API_KEY`).
|
||||
- If you are using a legacy API Key (that doesn't start with `nv-*` prefix), you might need to get an additional API Key by [creating your app](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#create-test-app) and set it as `NCP_APIGW_API_KEY`.
|
||||
- Install the integration Python package with:
|
||||
|
||||
```bash
|
||||
|
||||
@ -32,15 +32,13 @@
|
||||
"\n",
|
||||
"1. Creating [NAVER Cloud Platform](https://www.ncloud.com/) account \n",
|
||||
"2. Apply to use [CLOVA Studio](https://www.ncloud.com/product/aiService/clovaStudio)\n",
|
||||
"3. Find API Keys after creating CLOVA Studio Test App or Service App (See [here](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#테스트앱생성).)\n",
|
||||
"3. Create a CLOVA Studio Test App or Service App of a model to use (See [here](https://guide.ncloud-docs.com/docs/clovastudio-explorer03#%ED%85%8C%EC%8A%A4%ED%8A%B8%EC%95%B1%EC%83%9D%EC%84%B1).)\n",
|
||||
"4. Issue a Test or Service API key (See [here](https://api.ncloud-docs.com/docs/ai-naver-clovastudio-summary#API%ED%82%A4).)\n",
|
||||
"\n",
|
||||
"### Credentials\n",
|
||||
"\n",
|
||||
"CLOVA Studio requires 3 keys (`NCP_CLOVASTUDIO_API_KEY`, `NCP_APIGW_API_KEY` and `NCP_CLOVASTUDIO_APP_ID`) for embeddings.\n",
|
||||
"- `NCP_CLOVASTUDIO_API_KEY` and `NCP_CLOVASTUDIO_APP_ID` is issued per serviceApp or testApp\n",
|
||||
"- `NCP_APIGW_API_KEY` is issued per account\n",
|
||||
"\n",
|
||||
"The two API Keys could be found by clicking `App Request Status` > `Service App, Test App List` > `‘Details’ button for each app` in [CLOVA Studio](https://clovastudio.ncloud.com/studio-application/service-app)."
|
||||
"Set the `NCP_CLOVASTUDIO_API_KEY` environment variable with your API key.\n",
|
||||
" - Note that if you are using a legacy API Key (that doesn't start with `nv-*` prefix), you might need two additional keys to be set as environment variables (`NCP_APIGW_API_KEY` and `NCP_CLOVASTUDIO_APP_ID`. They could be found by clicking `App Request Status` > `Service App, Test App List` > `Details` button for each app in [CLOVA Studio](https://clovastudio.ncloud.com/studio-application/service-app)."
|
||||
]
|
||||
},
|
||||
{
|
||||
@ -56,9 +54,15 @@
|
||||
"if not os.getenv(\"NCP_CLOVASTUDIO_API_KEY\"):\n",
|
||||
" os.environ[\"NCP_CLOVASTUDIO_API_KEY\"] = getpass.getpass(\n",
|
||||
" \"Enter NCP CLOVA Studio API Key: \"\n",
|
||||
" )\n",
|
||||
"if not os.getenv(\"NCP_APIGW_API_KEY\"):\n",
|
||||
" os.environ[\"NCP_APIGW_API_KEY\"] = getpass.getpass(\"Enter NCP API Gateway API Key: \")"
|
||||
" )"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "b31fc062",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Uncomment below to use a legacy API key:"
|
||||
]
|
||||
},
|
||||
{
|
||||
@ -68,7 +72,9 @@
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"os.environ[\"NCP_CLOVASTUDIO_APP_ID\"] = input(\"Enter NCP CLOVA Studio App ID: \")"
|
||||
"# if not os.getenv(\"NCP_APIGW_API_KEY\"):\n",
|
||||
"# os.environ[\"NCP_APIGW_API_KEY\"] = getpass.getpass(\"Enter NCP API Gateway API Key: \")\n",
|
||||
"# os.environ[\"NCP_CLOVASTUDIO_APP_ID\"] = input(\"Enter NCP CLOVA Studio App ID: \")"
|
||||
]
|
||||
},
|
||||
{
|
||||
@ -118,8 +124,7 @@
|
||||
"from langchain_community.embeddings import ClovaXEmbeddings\n",
|
||||
"\n",
|
||||
"embeddings = ClovaXEmbeddings(\n",
|
||||
" model=\"clir-emb-dolphin\", # set with the model name of corresponding app id. Default is `clir-emb-dolphin`\n",
|
||||
" # app_id=\"...\" # set if you prefer to pass app id directly instead of using environment variables\n",
|
||||
" model=\"clir-emb-dolphin\" # set with the model name of corresponding app id. Default is `clir-emb-dolphin`\n",
|
||||
")"
|
||||
]
|
||||
},
|
||||
@ -251,7 +256,7 @@
|
||||
"\n",
|
||||
"When going live with production-level application using CLOVA Studio, you should apply for and use Service App. (See [here](https://guide.ncloud-docs.com/docs/en/clovastudio-playground01#서비스앱신청).)\n",
|
||||
"\n",
|
||||
"For a Service App, corresponding `NCP_CLOVASTUDIO_API_KEY` and `NCP_CLOVASTUDIO_APP_ID` are issued and can only be called with them."
|
||||
"For a Service App, you should use a corresponding Service API key and can only be called with it."
|
||||
]
|
||||
},
|
||||
{
|
||||
@ -266,6 +271,7 @@
|
||||
"os.environ[\"NCP_CLOVASTUDIO_API_KEY\"] = getpass.getpass(\n",
|
||||
" \"Enter NCP CLOVA Studio API Key for Service App: \"\n",
|
||||
")\n",
|
||||
"# Uncomment below to use a legacy API key:\n",
|
||||
"os.environ[\"NCP_CLOVASTUDIO_APP_ID\"] = input(\"Enter NCP CLOVA Studio Service App ID: \")"
|
||||
]
|
||||
},
|
||||
@ -279,7 +285,6 @@
|
||||
"embeddings = ClovaXEmbeddings(\n",
|
||||
" service_app=True,\n",
|
||||
" model=\"clir-emb-dolphin\", # set with the model name of corresponding app id of your Service App\n",
|
||||
" # app_id=\"...\" # set if you prefer to pass app id directly instead of using environment variables\n",
|
||||
")"
|
||||
]
|
||||
},
|
||||
|
||||
@ -15,6 +15,7 @@ from typing import (
|
||||
)
|
||||
|
||||
import httpx
|
||||
from httpx_sse import SSEError
|
||||
from langchain_core.callbacks import (
|
||||
AsyncCallbackManagerForLLMRun,
|
||||
CallbackManagerForLLMRun,
|
||||
@ -35,7 +36,13 @@ from langchain_core.messages import (
|
||||
)
|
||||
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
|
||||
from langchain_core.utils import convert_to_secret_str, get_from_env
|
||||
from pydantic import AliasChoices, ConfigDict, Field, SecretStr, model_validator
|
||||
from pydantic import (
|
||||
AliasChoices,
|
||||
ConfigDict,
|
||||
Field,
|
||||
SecretStr,
|
||||
model_validator,
|
||||
)
|
||||
from typing_extensions import Self
|
||||
|
||||
_DEFAULT_BASE_URL = "https://clovastudio.stream.ntruss.com"
|
||||
@ -47,16 +54,13 @@ def _convert_chunk_to_message_chunk(
|
||||
sse: Any, default_class: Type[BaseMessageChunk]
|
||||
) -> BaseMessageChunk:
|
||||
sse_data = sse.json()
|
||||
if sse.event == "result":
|
||||
response_metadata = _sse_data_to_response_metadata(sse_data)
|
||||
return AIMessageChunk(content="", response_metadata=response_metadata)
|
||||
|
||||
message = sse_data.get("message")
|
||||
role = message.get("role")
|
||||
content = message.get("content") or ""
|
||||
|
||||
if sse.event == "result":
|
||||
response_metadata = {}
|
||||
if "stopReason" in sse_data:
|
||||
response_metadata["stopReason"] = sse_data["stopReason"]
|
||||
return AIMessageChunk(content="", response_metadata=response_metadata)
|
||||
|
||||
if role == "user" or default_class == HumanMessageChunk:
|
||||
return HumanMessageChunk(content=content)
|
||||
elif role == "assistant" or default_class == AIMessageChunk:
|
||||
@ -69,6 +73,21 @@ def _convert_chunk_to_message_chunk(
|
||||
return default_class(content=content) # type: ignore[call-arg]
|
||||
|
||||
|
||||
def _sse_data_to_response_metadata(sse_data: Dict) -> Dict[str, Any]:
|
||||
response_metadata = {}
|
||||
if "stopReason" in sse_data:
|
||||
response_metadata["stop_reason"] = sse_data["stopReason"]
|
||||
if "inputLength" in sse_data:
|
||||
response_metadata["input_length"] = sse_data["inputLength"]
|
||||
if "outputLength" in sse_data:
|
||||
response_metadata["output_length"] = sse_data["outputLength"]
|
||||
if "seed" in sse_data:
|
||||
response_metadata["seed"] = sse_data["seed"]
|
||||
if "aiFilter" in sse_data:
|
||||
response_metadata["ai_filter"] = sse_data["aiFilter"]
|
||||
return response_metadata
|
||||
|
||||
|
||||
def _convert_message_to_naver_chat_message(
|
||||
message: BaseMessage,
|
||||
) -> Dict:
|
||||
@ -130,6 +149,8 @@ async def _aiter_sse(
|
||||
event_data = sse.json()
|
||||
if sse.event == "signal" and event_data.get("data", {}) == "[DONE]":
|
||||
return
|
||||
if sse.event == "error":
|
||||
raise SSEError(message=sse.data)
|
||||
yield sse
|
||||
|
||||
|
||||
@ -240,10 +261,15 @@ class ChatClovaX(BaseChatModel):
|
||||
|
||||
@property
|
||||
def lc_secrets(self) -> Dict[str, str]:
|
||||
return {
|
||||
"ncp_clovastudio_api_key": "NCP_CLOVASTUDIO_API_KEY",
|
||||
"ncp_apigw_api_key": "NCP_APIGW_API_KEY",
|
||||
}
|
||||
if not self._is_new_api_key():
|
||||
return {
|
||||
"ncp_clovastudio_api_key": "NCP_CLOVASTUDIO_API_KEY",
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"ncp_clovastudio_api_key": "NCP_CLOVASTUDIO_API_KEY",
|
||||
"ncp_apigw_api_key": "NCP_APIGW_API_KEY",
|
||||
}
|
||||
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
@ -285,10 +311,8 @@ class ChatClovaX(BaseChatModel):
|
||||
get_from_env("ncp_clovastudio_api_key", "NCP_CLOVASTUDIO_API_KEY")
|
||||
)
|
||||
|
||||
if not self.ncp_apigw_api_key:
|
||||
self.ncp_apigw_api_key = convert_to_secret_str(
|
||||
get_from_env("ncp_apigw_api_key", "NCP_APIGW_API_KEY", "")
|
||||
)
|
||||
if not self._is_new_api_key():
|
||||
self._init_fields_on_old_api_key()
|
||||
|
||||
if not self.base_url:
|
||||
self.base_url = get_from_env(
|
||||
@ -311,6 +335,18 @@ class ChatClovaX(BaseChatModel):
|
||||
|
||||
return self
|
||||
|
||||
def _is_new_api_key(self) -> bool:
|
||||
if self.ncp_clovastudio_api_key:
|
||||
return self.ncp_clovastudio_api_key.get_secret_value().startswith("nv-")
|
||||
else:
|
||||
return False
|
||||
|
||||
def _init_fields_on_old_api_key(self) -> None:
|
||||
if not self.ncp_apigw_api_key:
|
||||
self.ncp_apigw_api_key = convert_to_secret_str(
|
||||
get_from_env("ncp_apigw_api_key", "NCP_APIGW_API_KEY", "")
|
||||
)
|
||||
|
||||
def default_headers(self) -> Dict[str, Any]:
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
@ -322,16 +358,22 @@ class ChatClovaX(BaseChatModel):
|
||||
if self.ncp_clovastudio_api_key
|
||||
else None
|
||||
)
|
||||
if clovastudio_api_key:
|
||||
headers["X-NCP-CLOVASTUDIO-API-KEY"] = clovastudio_api_key
|
||||
|
||||
apigw_api_key = (
|
||||
self.ncp_apigw_api_key.get_secret_value()
|
||||
if self.ncp_apigw_api_key
|
||||
else None
|
||||
)
|
||||
if apigw_api_key:
|
||||
headers["X-NCP-APIGW-API-KEY"] = apigw_api_key
|
||||
if self._is_new_api_key():
|
||||
### headers on new api key
|
||||
headers["Authorization"] = f"Bearer {clovastudio_api_key}"
|
||||
else:
|
||||
### headers on old api key
|
||||
if clovastudio_api_key:
|
||||
headers["X-NCP-CLOVASTUDIO-API-KEY"] = clovastudio_api_key
|
||||
|
||||
apigw_api_key = (
|
||||
self.ncp_apigw_api_key.get_secret_value()
|
||||
if self.ncp_apigw_api_key
|
||||
else None
|
||||
)
|
||||
if apigw_api_key:
|
||||
headers["X-NCP-APIGW-API-KEY"] = apigw_api_key
|
||||
|
||||
return headers
|
||||
|
||||
@ -348,7 +390,6 @@ class ChatClovaX(BaseChatModel):
|
||||
def _completion_with_retry(self, **kwargs: Any) -> Any:
|
||||
from httpx_sse import (
|
||||
ServerSentEvent,
|
||||
SSEError,
|
||||
connect_sse,
|
||||
)
|
||||
|
||||
|
||||
@ -15,6 +15,7 @@ from pydantic import (
|
||||
from typing_extensions import Self
|
||||
|
||||
_DEFAULT_BASE_URL = "https://clovastudio.apigw.ntruss.com"
|
||||
_DEFAULT_BASE_URL_ON_NEW_API_KEY = "https://clovastudio.stream.ntruss.com"
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -91,20 +92,28 @@ class ClovaXEmbeddings(BaseModel, Embeddings):
|
||||
|
||||
@property
|
||||
def lc_secrets(self) -> Dict[str, str]:
|
||||
return {
|
||||
"ncp_clovastudio_api_key": "NCP_CLOVASTUDIO_API_KEY",
|
||||
"ncp_apigw_api_key": "NCP_APIGW_API_KEY",
|
||||
}
|
||||
if not self._is_new_api_key():
|
||||
return {
|
||||
"ncp_clovastudio_api_key": "NCP_CLOVASTUDIO_API_KEY",
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"ncp_clovastudio_api_key": "NCP_CLOVASTUDIO_API_KEY",
|
||||
"ncp_apigw_api_key": "NCP_APIGW_API_KEY",
|
||||
}
|
||||
|
||||
@property
|
||||
def _api_url(self) -> str:
|
||||
"""GET embedding api url"""
|
||||
app_type = "serviceapp" if self.service_app else "testapp"
|
||||
model_name = self.model_name if self.model_name != "bge-m3" else "v2"
|
||||
return (
|
||||
f"{self.base_url}/{app_type}"
|
||||
f"/v1/api-tools/embedding/{model_name}/{self.app_id}"
|
||||
)
|
||||
if self._is_new_api_key():
|
||||
return f"{self.base_url}/{app_type}" f"/v1/api-tools/embedding/{model_name}"
|
||||
else:
|
||||
return (
|
||||
f"{self.base_url}/{app_type}"
|
||||
f"/v1/api-tools/embedding/{model_name}/{self.app_id}"
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_model_after(self) -> Self:
|
||||
@ -113,18 +122,13 @@ class ClovaXEmbeddings(BaseModel, Embeddings):
|
||||
get_from_env("ncp_clovastudio_api_key", "NCP_CLOVASTUDIO_API_KEY")
|
||||
)
|
||||
|
||||
if not self.ncp_apigw_api_key:
|
||||
self.ncp_apigw_api_key = convert_to_secret_str(
|
||||
get_from_env("ncp_apigw_api_key", "NCP_APIGW_API_KEY", "")
|
||||
)
|
||||
if self._is_new_api_key():
|
||||
self._init_fields_on_new_api_key()
|
||||
else:
|
||||
self._init_fields_on_old_api_key()
|
||||
|
||||
if not self.base_url:
|
||||
self.base_url = get_from_env(
|
||||
"base_url", "NCP_CLOVASTUDIO_API_BASE_URL", _DEFAULT_BASE_URL
|
||||
)
|
||||
|
||||
if not self.app_id:
|
||||
self.app_id = get_from_env("app_id", "NCP_CLOVASTUDIO_APP_ID")
|
||||
raise ValueError("base_url dose not exist.")
|
||||
|
||||
if not self.client:
|
||||
self.client = httpx.Client(
|
||||
@ -133,7 +137,7 @@ class ClovaXEmbeddings(BaseModel, Embeddings):
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
if not self.async_client:
|
||||
if not self.async_client and self.base_url:
|
||||
self.async_client = httpx.AsyncClient(
|
||||
base_url=self.base_url,
|
||||
headers=self.default_headers(),
|
||||
@ -142,6 +146,32 @@ class ClovaXEmbeddings(BaseModel, Embeddings):
|
||||
|
||||
return self
|
||||
|
||||
def _is_new_api_key(self) -> bool:
|
||||
if self.ncp_clovastudio_api_key:
|
||||
return self.ncp_clovastudio_api_key.get_secret_value().startswith("nv-")
|
||||
else:
|
||||
return False
|
||||
|
||||
def _init_fields_on_new_api_key(self) -> None:
|
||||
if not self.base_url:
|
||||
self.base_url = get_from_env(
|
||||
"base_url",
|
||||
"NCP_CLOVASTUDIO_API_BASE_URL",
|
||||
_DEFAULT_BASE_URL_ON_NEW_API_KEY,
|
||||
)
|
||||
|
||||
def _init_fields_on_old_api_key(self) -> None:
|
||||
if not self.ncp_apigw_api_key:
|
||||
self.ncp_apigw_api_key = convert_to_secret_str(
|
||||
get_from_env("ncp_apigw_api_key", "NCP_APIGW_API_KEY", "")
|
||||
)
|
||||
if not self.base_url:
|
||||
self.base_url = get_from_env(
|
||||
"base_url", "NCP_CLOVASTUDIO_API_BASE_URL", _DEFAULT_BASE_URL
|
||||
)
|
||||
if not self.app_id:
|
||||
self.app_id = get_from_env("app_id", "NCP_CLOVASTUDIO_APP_ID")
|
||||
|
||||
def default_headers(self) -> Dict[str, Any]:
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
@ -153,16 +183,22 @@ class ClovaXEmbeddings(BaseModel, Embeddings):
|
||||
if self.ncp_clovastudio_api_key
|
||||
else None
|
||||
)
|
||||
if clovastudio_api_key:
|
||||
headers["X-NCP-CLOVASTUDIO-API-KEY"] = clovastudio_api_key
|
||||
|
||||
apigw_api_key = (
|
||||
self.ncp_apigw_api_key.get_secret_value()
|
||||
if self.ncp_apigw_api_key
|
||||
else None
|
||||
)
|
||||
if apigw_api_key:
|
||||
headers["X-NCP-APIGW-API-KEY"] = apigw_api_key
|
||||
if self._is_new_api_key():
|
||||
### headers on new api key
|
||||
headers["Authorization"] = f"Bearer {clovastudio_api_key}"
|
||||
else:
|
||||
### headers on old api key
|
||||
if clovastudio_api_key:
|
||||
headers["X-NCP-CLOVASTUDIO-API-KEY"] = clovastudio_api_key
|
||||
|
||||
apigw_api_key = (
|
||||
self.ncp_apigw_api_key.get_secret_value()
|
||||
if self.ncp_apigw_api_key
|
||||
else None
|
||||
)
|
||||
if apigw_api_key:
|
||||
headers["X-NCP-APIGW-API-KEY"] = apigw_api_key
|
||||
|
||||
return headers
|
||||
|
||||
@ -175,7 +211,7 @@ class ClovaXEmbeddings(BaseModel, Embeddings):
|
||||
|
||||
async def _aembed_text(self, text: str) -> List[float]:
|
||||
payload = {"text": text}
|
||||
async_client = cast(httpx.AsyncClient, self.client)
|
||||
async_client = cast(httpx.AsyncClient, self.async_client)
|
||||
response = await async_client.post(url=self._api_url, json=payload)
|
||||
await _araise_on_error(response)
|
||||
return response.json()["result"]["embedding"]
|
||||
|
||||
@ -1,71 +1,131 @@
|
||||
"""Test ChatNaver chat model."""
|
||||
"""Test ChatClovaX chat model."""
|
||||
|
||||
from langchain_core.messages import AIMessage, AIMessageChunk
|
||||
import pytest
|
||||
from httpx_sse import SSEError
|
||||
from langchain_core.messages import (
|
||||
AIMessage,
|
||||
AIMessageChunk,
|
||||
)
|
||||
|
||||
from langchain_community.chat_models import ChatClovaX
|
||||
|
||||
|
||||
def test_stream() -> None:
|
||||
"""Test streaming tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
for token in llm.stream("I'm Clova"):
|
||||
assert isinstance(token, AIMessageChunk)
|
||||
assert isinstance(token.content, str)
|
||||
if token.response_metadata:
|
||||
assert "input_length" in token.response_metadata
|
||||
assert "output_length" in token.response_metadata
|
||||
assert "stop_reason" in token.response_metadata
|
||||
assert "ai_filter" in token.response_metadata
|
||||
|
||||
|
||||
async def test_astream() -> None:
|
||||
"""Test streaming tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
async for token in llm.astream("I'm Clova"):
|
||||
assert isinstance(token, AIMessageChunk)
|
||||
assert isinstance(token.content, str)
|
||||
if token.response_metadata:
|
||||
assert "input_length" in token.response_metadata
|
||||
assert "output_length" in token.response_metadata
|
||||
assert "stop_reason" in token.response_metadata
|
||||
assert "ai_filter" in token.response_metadata
|
||||
|
||||
|
||||
async def test_abatch() -> None:
|
||||
"""Test streaming tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
result = await llm.abatch(["I'm Clova", "I'm not Clova"])
|
||||
for token in result:
|
||||
assert isinstance(token, AIMessage)
|
||||
assert isinstance(token.content, str)
|
||||
if token.response_metadata:
|
||||
assert "input_length" in token.response_metadata
|
||||
assert "output_length" in token.response_metadata
|
||||
assert "stop_reason" in token.response_metadata
|
||||
assert "ai_filter" in token.response_metadata
|
||||
|
||||
|
||||
async def test_abatch_tags() -> None:
|
||||
"""Test batch tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
result = await llm.abatch(["I'm Clova", "I'm not Clova"], config={"tags": ["foo"]})
|
||||
for token in result:
|
||||
assert isinstance(token, AIMessage)
|
||||
assert isinstance(token.content, str)
|
||||
if token.response_metadata:
|
||||
assert "input_length" in token.response_metadata
|
||||
assert "output_length" in token.response_metadata
|
||||
assert "stop_reason" in token.response_metadata
|
||||
assert "ai_filter" in token.response_metadata
|
||||
|
||||
|
||||
def test_batch() -> None:
|
||||
"""Test batch tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
result = llm.batch(["I'm Clova", "I'm not Clova"])
|
||||
for token in result:
|
||||
assert isinstance(token, AIMessage)
|
||||
assert isinstance(token.content, str)
|
||||
if token.response_metadata:
|
||||
assert "input_length" in token.response_metadata
|
||||
assert "output_length" in token.response_metadata
|
||||
assert "stop_reason" in token.response_metadata
|
||||
assert "ai_filter" in token.response_metadata
|
||||
|
||||
|
||||
async def test_ainvoke() -> None:
|
||||
"""Test invoke tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
result = await llm.ainvoke("I'm Clova", config={"tags": ["foo"]})
|
||||
assert isinstance(result, AIMessage)
|
||||
assert isinstance(result.content, str)
|
||||
if result.response_metadata:
|
||||
assert "input_length" in result.response_metadata
|
||||
assert "output_length" in result.response_metadata
|
||||
assert "stop_reason" in result.response_metadata
|
||||
assert "ai_filter" in result.response_metadata
|
||||
|
||||
|
||||
def test_invoke() -> None:
|
||||
"""Test invoke tokens from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
llm = ChatClovaX(include_ai_filters=True)
|
||||
|
||||
result = llm.invoke("I'm Clova", config=dict(tags=["foo"]))
|
||||
assert isinstance(result, AIMessage)
|
||||
assert isinstance(result.content, str)
|
||||
if result.response_metadata:
|
||||
assert "input_length" in result.response_metadata
|
||||
assert "output_length" in result.response_metadata
|
||||
assert "stop_reason" in result.response_metadata
|
||||
assert "ai_filter" in result.response_metadata
|
||||
|
||||
|
||||
def test_stream_error_event() -> None:
|
||||
"""Test streaming error event from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
prompt = "What is the best way to reduce my carbon footprint?"
|
||||
|
||||
with pytest.raises(SSEError):
|
||||
for _ in llm.stream(prompt * 1000):
|
||||
pass
|
||||
|
||||
|
||||
async def test_astream_error_event() -> None:
|
||||
"""Test streaming error event from ChatClovaX."""
|
||||
llm = ChatClovaX()
|
||||
prompt = "What is the best way to reduce my carbon footprint?"
|
||||
|
||||
with pytest.raises(SSEError):
|
||||
async for _ in llm.astream(prompt * 1000):
|
||||
pass
|
||||
|
||||
@ -4,7 +4,7 @@ from langchain_community.embeddings import ClovaXEmbeddings
|
||||
|
||||
|
||||
def test_embedding_documents() -> None:
|
||||
"""Test cohere embeddings."""
|
||||
"""Test ClovaX embeddings."""
|
||||
documents = ["foo bar"]
|
||||
embedding = ClovaXEmbeddings()
|
||||
output = embedding.embed_documents(documents)
|
||||
@ -13,7 +13,7 @@ def test_embedding_documents() -> None:
|
||||
|
||||
|
||||
async def test_aembedding_documents() -> None:
|
||||
"""Test cohere embeddings."""
|
||||
"""Test ClovaX embeddings."""
|
||||
documents = ["foo bar"]
|
||||
embedding = ClovaXEmbeddings()
|
||||
output = await embedding.aembed_documents(documents)
|
||||
@ -22,7 +22,7 @@ async def test_aembedding_documents() -> None:
|
||||
|
||||
|
||||
def test_embedding_query() -> None:
|
||||
"""Test cohere embeddings."""
|
||||
"""Test ClovaX embeddings."""
|
||||
document = "foo bar"
|
||||
embedding = ClovaXEmbeddings()
|
||||
output = embedding.embed_query(document)
|
||||
@ -30,7 +30,7 @@ def test_embedding_query() -> None:
|
||||
|
||||
|
||||
async def test_aembedding_query() -> None:
|
||||
"""Test cohere embeddings."""
|
||||
"""Test ClovaX embeddings."""
|
||||
document = "foo bar"
|
||||
embedding = ClovaXEmbeddings()
|
||||
output = await embedding.aembed_query(document)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user