feat: add Qianfan model provider implementation with various model credentials and functionalities

apps/models_provider/impl/qianfan_model_provider/__init__.py

@@ -0,0 +1,8 @@
+# coding=utf-8
+"""
+@project: maxkb
+@Author：虎
+@file： __init__.py.py
+@date：2023/10/31 17:16
+@desc:
+"""

apps/models_provider/impl/qianfan_model_provider/credential/embedding.py

@@ -0,0 +1,58 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author：虎
+@file： embedding.py
+@date：2024/10/17 15:40
+@desc:
+"""
+
+from typing import Dict
+
+from django.utils.translation import gettext as _
+
+from common import forms
+from common.exception.app_exception import AppApiException
+from common.forms import BaseForm
+from common.utils.logger import maxkb_logger
+from models_provider.base_model_provider import BaseModelCredential, ValidCode
+
+
+class QianfanEmbeddingCredential(BaseForm, BaseModelCredential):
+    api_base = forms.TextInputField("API URL", required=True, default_value="https://qianfan.baidubce.com/v2")
+    api_key = forms.PasswordInputField("API Key", required=True)
+
+    def is_valid(
+        self,
+        model_type: str,
+        model_name,
+        model_credential: Dict[str, object],
+        model_params,
+        provider,
+        raise_exception=False,
+    ):
+        for key in ["api_base", "api_key"]:
+            if key not in model_credential:
+                if raise_exception:
+                    raise AppApiException(ValidCode.valid_error.value, _("{key}  is required").format(key=key))
+                return False
+
+        try:
+            model = provider.get_model(model_type, model_name, model_credential)
+            model.embed_query(_("Hello"))
+        except Exception as e:
+            maxkb_logger.error(f"Exception: {e}", exc_info=True)
+            if isinstance(e, AppApiException):
+                raise e
+            if raise_exception:
+                raise AppApiException(
+                    ValidCode.valid_error.value,
+                    _("Verification failed, please check whether the parameters are correct: {error}").format(
+                        error=str(e)
+                    ),
+                )
+            return False
+        return True
+
+    def encryption_dict(self, model: Dict[str, object]):
+        return {**model, "api_key": super().encryption(model.get("api_key", ""))}

apps/models_provider/impl/qianfan_model_provider/credential/image.py

@@ -0,0 +1,89 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@file： image.py
+@desc: 千帆视觉理解模型凭据
+"""
+
+from typing import Dict
+
+from django.utils.translation import gettext, gettext_lazy as _
+from langchain_core.messages import HumanMessage
+
+from common import forms
+from common.exception.app_exception import AppApiException
+from common.forms import BaseForm, TooltipLabel
+from common.utils.logger import maxkb_logger
+from models_provider.base_model_provider import BaseModelCredential, ValidCode
+
+
+class QianfanImageModelParams(BaseForm):
+    temperature = forms.SliderField(
+        TooltipLabel(
+            _("Temperature"),
+            _("Higher values make the output more random, while lower values make it more focused and deterministic"),
+        ),
+        required=True,
+        default_value=0.95,
+        _min=0.1,
+        _max=1.0,
+        _step=0.01,
+        precision=2,
+    )
+
+    max_tokens = forms.SliderField(
+        TooltipLabel(
+            _("Output the maximum Tokens"), _("Specify the maximum number of tokens that the model can generate")
+        ),
+        required=True,
+        default_value=1024,
+        _min=1,
+        _max=100000,
+        _step=1,
+        precision=0,
+    )
+
+
+class QianfanImageModelCredential(BaseForm, BaseModelCredential):
+    api_base = forms.TextInputField("API URL", required=True, default_value="https://qianfan.baidubce.com/v2")
+    api_key = forms.PasswordInputField("API Key", required=True)
+
+    def is_valid(
+        self,
+        model_type: str,
+        model_name,
+        model_credential: Dict[str, object],
+        model_params,
+        provider,
+        raise_exception=False,
+    ):
+        for key in ["api_base", "api_key"]:
+            if key not in model_credential:
+                if raise_exception:
+                    raise AppApiException(ValidCode.valid_error.value, gettext("{key}  is required").format(key=key))
+                return False
+
+        try:
+            model = provider.get_model(model_type, model_name, model_credential, **model_params)
+            response = model.stream([HumanMessage(content=[{"type": "text", "text": gettext("Hello")}])])
+            for _chunk in response:
+                break
+        except Exception as e:
+            maxkb_logger.error(f"Exception: {e}", exc_info=True)
+            if isinstance(e, AppApiException):
+                raise e
+            if raise_exception:
+                raise AppApiException(
+                    ValidCode.valid_error.value,
+                    gettext("Verification failed, please check whether the parameters are correct: {error}").format(
+                        error=str(e)
+                    ),
+                )
+            return False
+        return True
+
+    def encryption_dict(self, model: Dict[str, object]):
+        return {**model, "api_key": super().encryption(model.get("api_key", ""))}
+
+    def get_model_params_setting_form(self, model_name):
+        return QianfanImageModelParams()

apps/models_provider/impl/qianfan_model_provider/credential/llm.py

@@ -0,0 +1,88 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author：虎
+@file： llm.py
+@date：2024/7/12 10:19
+@desc:
+"""
+
+from typing import Dict
+
+from django.utils.translation import gettext, gettext_lazy as _
+from langchain_core.messages import HumanMessage
+
+from common import forms
+from common.exception.app_exception import AppApiException
+from common.forms import BaseForm, TooltipLabel
+from common.utils.logger import maxkb_logger
+from models_provider.base_model_provider import BaseModelCredential, ValidCode
+
+
+class QianfanLLMModelParams(BaseForm):
+    temperature = forms.SliderField(
+        TooltipLabel(
+            _("Temperature"),
+            _("Higher values make the output more random, while lower values make it more focused and deterministic"),
+        ),
+        required=True,
+        default_value=0.95,
+        _min=0.1,
+        _max=1.0,
+        _step=0.01,
+        precision=2,
+    )
+
+    max_tokens = forms.SliderField(
+        TooltipLabel(
+            _("Output the maximum Tokens"), _("Specify the maximum number of tokens that the model can generate")
+        ),
+        required=True,
+        default_value=1024,
+        _min=2,
+        _max=100000,
+        _step=1,
+        precision=0,
+    )
+
+
+class QianfanLLMModelCredential(BaseForm, BaseModelCredential):
+    api_base = forms.TextInputField("API URL", required=True, default_value="https://qianfan.baidubce.com/v2")
+    api_key = forms.PasswordInputField("API Key", required=True)
+
+    def is_valid(
+        self,
+        model_type: str,
+        model_name,
+        model_credential: Dict[str, object],
+        model_params,
+        provider,
+        raise_exception=False,
+    ):
+        for key in ["api_base", "api_key"]:
+            if key not in model_credential:
+                if raise_exception:
+                    raise AppApiException(ValidCode.valid_error.value, gettext("{key}  is required").format(key=key))
+                return False
+
+        try:
+            model = provider.get_model(model_type, model_name, model_credential, **{**model_params, "max_tokens": 1})
+            model.invoke([HumanMessage(content="1")])
+        except Exception as e:
+            maxkb_logger.error(f"Exception: {e}", exc_info=True)
+            raise e
+        return True
+
+    def encryption_dict(self, model_info: Dict[str, object]):
+        return {**model_info, "api_key": super().encryption(model_info.get("api_key", ""))}
+
+    def build_model(self, model_info: Dict[str, object]):
+        for key in ["api_base", "api_key", "model"]:
+            if key not in model_info:
+                raise AppApiException(500, gettext("{key}  is required").format(key=key))
+        self.api_base = model_info.get("api_base")
+        self.api_key = model_info.get("api_key")
+        return self
+
+    def get_model_params_setting_form(self, model_name):
+        return QianfanLLMModelParams()

apps/models_provider/impl/qianfan_model_provider/credential/reranker.py

@@ -0,0 +1,74 @@
+from typing import Dict
+
+from langchain_core.documents import Document
+
+from common import forms
+from common.exception.app_exception import AppApiException
+from common.forms import BaseForm, TooltipLabel
+from models_provider.base_model_provider import BaseModelCredential, ValidCode
+from django.utils.translation import gettext_lazy as _
+from common.utils.logger import maxkb_logger
+from models_provider.impl.qianfan_model_provider.model.reranker import QfBgeReranker
+
+
+class QfRerankerModelParams(BaseForm):
+    top_n = forms.SliderField(
+        TooltipLabel(_("Top N"), _("Number of top documents to return after reranking")),
+        required=True,
+        default_value=3,
+        _min=1,
+        _max=100,
+        _step=1,
+        precision=0,
+    )
+
+
+class QfRerankerCredential(BaseForm, BaseModelCredential):
+    api_url = forms.TextInputField("API URL", required=True)
+    api_key = forms.PasswordInputField("API Key", required=True)
+
+    def is_valid(
+        self,
+        model_type: str,
+        model_name,
+        model_credential: Dict[str, object],
+        model_params,
+        provider,
+        raise_exception=True,
+    ):
+        model_type_list = provider.get_model_type_list()
+        if not any(list(filter(lambda mt: mt.get("value") == model_type, model_type_list))):
+            raise AppApiException(
+                ValidCode.valid_error.value, _("{model_type} Model type is not supported").format(model_type=model_type)
+            )
+
+        for key in ["api_url", "api_key"]:
+            if key not in model_credential:
+                if raise_exception:
+                    raise AppApiException(ValidCode.valid_error.value, _("{key}  is required").format(key=key))
+                else:
+                    return False
+        try:
+            model: QfBgeReranker = provider.get_model(model_type, model_name, model_credential)
+            test_text = str(_("Hello"))
+            model.compress_documents([Document(page_content=test_text)], test_text)
+        except Exception as e:
+            maxkb_logger.error(f"Exception: {e}", exc_info=True)
+            if isinstance(e, AppApiException):
+                raise e
+            if raise_exception:
+                raise AppApiException(
+                    ValidCode.valid_error.value,
+                    _("Verification failed, please check whether the parameters are correct: {error}").format(
+                        error=str(e)
+                    ),
+                )
+            return False
+
+        return True
+
+    def encryption_dict(self, model_info: Dict[str, object]):
+        return {**model_info, "api_key": super().encryption(model_info.get("api_key", ""))}
+
+    def get_model_params_setting_form(self, model_name: str) -> QfRerankerModelParams:
+        return QfRerankerModelParams()