15308555518/ MaxKB
1
0
Fork 0
mirror of https://github.com/1Panel-dev/MaxKB.git synced 2025-12-25 17:22:55 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

feat: Support Azure image tts stt model

This commit is contained in:
CaptainB 2024-12-19 12:02:24 +08:00
parent 82f263bc23
commit 4a8caab6f1
9 changed files with 520 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
apps/setting/models_provider/impl/azure_model_provider/azure_model_provider.py
View File

@ -12,13 +12,25 @@ from common.util.file_util import get_file_content
from setting.models_provider.base_model_provider import IModelProvider, ModelProvideInfo, ModelInfo, \
ModelTypeConst, ModelInfoManage
from setting.models_provider.impl.azure_model_provider.credential.embedding import AzureOpenAIEmbeddingCredential
from setting.models_provider.impl.azure_model_provider.credential.image import AzureOpenAIImageModelCredential
from setting.models_provider.impl.azure_model_provider.credential.llm import AzureLLMModelCredential
from setting.models_provider.impl.azure_model_provider.credential.stt import AzureOpenAISTTModelCredential
from setting.models_provider.impl.azure_model_provider.credential.tti import AzureOpenAITextToImageModelCredential
from setting.models_provider.impl.azure_model_provider.credential.tts import AzureOpenAITTSModelCredential
from setting.models_provider.impl.azure_model_provider.model.azure_chat_model import AzureChatModel
from setting.models_provider.impl.azure_model_provider.model.embedding import AzureOpenAIEmbeddingModel
from setting.models_provider.impl.azure_model_provider.model.image import AzureOpenAIImage
from setting.models_provider.impl.azure_model_provider.model.stt import AzureOpenAISpeechToText
from setting.models_provider.impl.azure_model_provider.model.tti import AzureOpenAITextToImage
from setting.models_provider.impl.azure_model_provider.model.tts import AzureOpenAITextToSpeech
from smartdoc.conf import PROJECT_DIR
base_azure_llm_model_credential = AzureLLMModelCredential()
base_azure_embedding_model_credential = AzureOpenAIEmbeddingCredential()
base_azure_image_model_credential = AzureOpenAIImageModelCredential()
base_azure_tti_model_credential = AzureOpenAITextToImageModelCredential()
base_azure_tts_model_credential = AzureOpenAITTSModelCredential()
base_azure_stt_model_credential = AzureOpenAISTTModelCredential()
default_model_info = ModelInfo('Azure OpenAI', '具体的基础模型由部署名决定', ModelTypeConst.LLM,
base_azure_llm_model_credential, AzureChatModel, api_version='2024-02-15-preview'
@ -36,12 +48,47 @@ embedding_model_info = [
),
]
image_model_info = [
ModelInfo('gpt-4o', '', ModelTypeConst.IMAGE,
base_azure_image_model_credential, AzureOpenAIImage, api_version='2023-05-15'
),
ModelInfo('gpt-4o-mini', '', ModelTypeConst.IMAGE,
base_azure_image_model_credential, AzureOpenAIImage, api_version='2023-05-15'
),
]
tti_model_info = [
ModelInfo('dall-e-3', '', ModelTypeConst.TTI,
base_azure_tti_model_credential, AzureOpenAITextToImage, api_version='2023-05-15'
),
]
tts_model_info = [
ModelInfo('tts', '', ModelTypeConst.TTS,
base_azure_tts_model_credential, AzureOpenAITextToSpeech, api_version='2023-05-15'
),
]
stt_model_info = [
ModelInfo('whisper', '', ModelTypeConst.STT,
base_azure_stt_model_credential, AzureOpenAISpeechToText, api_version='2023-05-15'
),
]
model_info_manage = (
ModelInfoManage.builder()
.append_default_model_info(default_model_info)
.append_model_info(default_model_info)
.append_model_info_list(embedding_model_info)
.append_default_model_info(embedding_model_info[0])
.append_model_info_list(image_model_info)
.append_default_model_info(image_model_info[0])
.append_model_info_list(stt_model_info)
.append_default_model_info(stt_model_info[0])
.append_model_info_list(tts_model_info)
.append_default_model_info(tts_model_info[0])
.append_model_info_list(tti_model_info)
.append_default_model_info(tti_model_info[0])
.build()
)

66
apps/setting/models_provider/impl/azure_model_provider/credential/image.py Normal file
View File

@ -0,0 +1,66 @@
# coding=utf-8
import base64
import os
from typing import Dict
from langchain_core.messages import HumanMessage
from common import forms
from common.exception.app_exception import AppApiException
from common.forms import BaseForm, TooltipLabel
from setting.models_provider.base_model_provider import BaseModelCredential, ValidCode
class AzureOpenAIImageModelParams(BaseForm):
temperature = forms.SliderField(TooltipLabel('温度', '较高的数值会使输出更加随机，而较低的数值会使其更加集中和确定'),
required=True, default_value=0.7,
_min=0.1,
_max=1.0,
_step=0.01,
precision=2)
max_tokens = forms.SliderField(
TooltipLabel('输出最大Tokens', '指定模型可生成的最大token个数'),
required=True, default_value=800,
_min=1,
_max=100000,
_step=1,
precision=0)
class AzureOpenAIImageModelCredential(BaseForm, BaseModelCredential):
api_version = forms.TextInputField("API 版本 (api_version)", required=True)
api_base = forms.TextInputField('API 域名 (azure_endpoint)', required=True)
api_key = forms.PasswordInputField("API Key (api_key)", required=True)
def is_valid(self, model_type: str, model_name, model_credential: Dict[str, object], provider,
raise_exception=False):
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, f'{model_type} 模型类型不支持')
for key in ['api_base', 'api_key', 'api_version']:
if key not in model_credential:
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'{key} 字段为必填字段')
else:
return False
try:
model = provider.get_model(model_type, model_name, model_credential)
res = model.stream([HumanMessage(content=[{"type": "text", "text": "你好"}])])
for chunk in res:
print(chunk)
except Exception as e:
if isinstance(e, AppApiException):
raise e
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'校验失败,请检查参数是否正确: {str(e)}')
else:
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 AzureOpenAIImageModelParams()

43
apps/setting/models_provider/impl/azure_model_provider/credential/stt.py Normal file
View File

@ -0,0 +1,43 @@
# coding=utf-8
from typing import Dict
from common import forms
from common.exception.app_exception import AppApiException
from common.forms import BaseForm
from setting.models_provider.base_model_provider import BaseModelCredential, ValidCode
class AzureOpenAISTTModelCredential(BaseForm, BaseModelCredential):
api_version = forms.TextInputField("API 版本 (api_version)", required=True)
api_base = forms.TextInputField('API 域名 (azure_endpoint)', required=True)
api_key = forms.PasswordInputField("API Key (api_key)", required=True)
def is_valid(self, model_type: str, model_name, model_credential: Dict[str, object], provider,
raise_exception=False):
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, f'{model_type} 模型类型不支持')
for key in ['api_base', 'api_key', 'api_version']:
if key not in model_credential:
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'{key} 字段为必填字段')
else:
return False
try:
model = provider.get_model(model_type, model_name, model_credential)
model.check_auth()
except Exception as e:
if isinstance(e, AppApiException):
raise e
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'校验失败,请检查参数是否正确: {str(e)}')
else:
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):
pass

83
apps/setting/models_provider/impl/azure_model_provider/credential/tti.py Normal file
View File

@ -0,0 +1,83 @@
# coding=utf-8
import base64
import os
from typing import Dict
from langchain_core.messages import HumanMessage
from common import forms
from common.exception.app_exception import AppApiException
from common.forms import BaseForm, TooltipLabel
from setting.models_provider.base_model_provider import BaseModelCredential, ValidCode
class AzureOpenAITTIModelParams(BaseForm):
size = forms.SingleSelect(
TooltipLabel('图片尺寸', '指定生成图片的尺寸, 如: 1024x1024'),
required=True,
default_value='1024x1024',
option_list=[
{'value': '1024x1024', 'label': '1024x1024'},
{'value': '1024x1792', 'label': '1024x1792'},
{'value': '1792x1024', 'label': '1792x1024'},
],
text_field='label',
value_field='value'
)
quality = forms.SingleSelect(
TooltipLabel('图片质量', ''),
required=True,
default_value='standard',
option_list=[
{'value': 'standard', 'label': 'standard'},
{'value': 'hd', 'label': 'hd'},
],
text_field='label',
value_field='value'
)
n = forms.SliderField(
TooltipLabel('图片数量', '指定生成图片的数量'),
required=True, default_value=1,
_min=1,
_max=10,
_step=1,
precision=0)
class AzureOpenAITextToImageModelCredential(BaseForm, BaseModelCredential):
api_version = forms.TextInputField("API 版本 (api_version)", required=True)
api_base = forms.TextInputField('API 域名 (azure_endpoint)', required=True)
api_key = forms.PasswordInputField("API Key (api_key)", required=True)
def is_valid(self, model_type: str, model_name, model_credential: Dict[str, object], provider,
raise_exception=False):
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, f'{model_type} 模型类型不支持')
for key in ['api_base', 'api_key', 'api_version']:
if key not in model_credential:
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'{key} 字段为必填字段')
else:
return False
try:
model = provider.get_model(model_type, model_name, model_credential)
res = model.check_auth()
print(res)
except Exception as e:
if isinstance(e, AppApiException):
raise e
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'校验失败,请检查参数是否正确: {str(e)}')
else:
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 AzureOpenAITTIModelParams()

59
apps/setting/models_provider/impl/azure_model_provider/credential/tts.py Normal file
View File

@ -0,0 +1,59 @@
# coding=utf-8
from typing import Dict
from common import forms
from common.exception.app_exception import AppApiException
from common.forms import BaseForm, TooltipLabel
from setting.models_provider.base_model_provider import BaseModelCredential, ValidCode
class AzureOpenAITTSModelGeneralParams(BaseForm):
# alloy, echo, fable, onyx, nova, shimmer
voice = forms.SingleSelect(
TooltipLabel('Voice', '尝试不同的声音（合金、回声、寓言、缟玛瑙、新星和闪光），找到一种适合您所需的音调和听众的声音。当前的语音针对英语进行了优化。'),
required=True, default_value='alloy',
text_field='value',
value_field='value',
option_list=[
{'text': 'alloy', 'value': 'alloy'},
{'text': 'echo', 'value': 'echo'},
{'text': 'fable', 'value': 'fable'},
{'text': 'onyx', 'value': 'onyx'},
{'text': 'nova', 'value': 'nova'},
{'text': 'shimmer', 'value': 'shimmer'},
])
class AzureOpenAITTSModelCredential(BaseForm, BaseModelCredential):
api_version = forms.TextInputField("API 版本 (api_version)", required=True)
api_base = forms.TextInputField('API 域名 (azure_endpoint)', required=True)
api_key = forms.PasswordInputField("API Key (api_key)", required=True)
def is_valid(self, model_type: str, model_name, model_credential: Dict[str, object], provider,
raise_exception=False):
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, f'{model_type} 模型类型不支持')
for key in ['api_base', 'api_key', 'api_version']:
if key not in model_credential:
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'{key} 字段为必填字段')
else:
return False
try:
model = provider.get_model(model_type, model_name, model_credential)
model.check_auth()
except Exception as e:
if isinstance(e, AppApiException):
raise e
if raise_exception:
raise AppApiException(ValidCode.valid_error.value, f'校验失败,请检查参数是否正确: {str(e)}')
else:
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 AzureOpenAITTSModelGeneralParams()

28
apps/setting/models_provider/impl/azure_model_provider/model/image.py Normal file
View File

@ -0,0 +1,28 @@
from typing import Dict
from langchain_openai import AzureChatOpenAI
from langchain_openai.chat_models import ChatOpenAI
from common.config.tokenizer_manage_config import TokenizerManage
from setting.models_provider.base_model_provider import MaxKBBaseModel
def custom_get_token_ids(text: str):
tokenizer = TokenizerManage.get_tokenizer()
return tokenizer.encode(text)
class AzureOpenAIImage(MaxKBBaseModel, AzureChatOpenAI):
@staticmethod
def new_instance(model_type, model_name, model_credential: Dict[str, object], **model_kwargs):
optional_params = MaxKBBaseModel.filter_optional_params(model_kwargs)
return AzureOpenAIImage(
model_name=model_name,
openai_api_key=model_credential.get('api_key'),
azure_endpoint=model_credential.get('api_base'),
openai_api_version=model_credential.get('api_version'),
openai_api_type="azure",
streaming=True,
**optional_params,
)

64
apps/setting/models_provider/impl/azure_model_provider/model/stt.py Normal file
View File

@ -0,0 +1,64 @@
import asyncio
import io
from typing import Dict
from openai import OpenAI, AzureOpenAI
from common.config.tokenizer_manage_config import TokenizerManage
from setting.models_provider.base_model_provider import MaxKBBaseModel
from setting.models_provider.impl.base_stt import BaseSpeechToText
def custom_get_token_ids(text: str):
tokenizer = TokenizerManage.get_tokenizer()
return tokenizer.encode(text)
class AzureOpenAISpeechToText(MaxKBBaseModel, BaseSpeechToText):
api_base: str
api_key: str
api_version: str
model: str
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.api_key = kwargs.get('api_key')
self.api_base = kwargs.get('api_base')
self.api_version = kwargs.get('api_version')
@staticmethod
def new_instance(model_type, model_name, model_credential: Dict[str, object], **model_kwargs):
optional_params = {}
if 'max_tokens' in model_kwargs and model_kwargs['max_tokens'] is not None:
optional_params['max_tokens'] = model_kwargs['max_tokens']
if 'temperature' in model_kwargs and model_kwargs['temperature'] is not None:
optional_params['temperature'] = model_kwargs['temperature']
return AzureOpenAISpeechToText(
model=model_name,
api_base=model_credential.get('api_base'),
api_key=model_credential.get('api_key'),
api_version=model_credential.get('api_version'),
**optional_params,
)
def check_auth(self):
client = AzureOpenAI(
azure_endpoint=self.api_base,
api_key=self.api_key,
api_version=self.api_version
)
response_list = client.models.with_raw_response.list()
# print(response_list)
def speech_to_text(self, audio_file):
client = AzureOpenAI(
azure_endpoint=self.api_base,
api_key=self.api_key,
api_version=self.api_version
)
audio_data = audio_file.read()
buffer = io.BytesIO(audio_data)
buffer.name = "file.mp3" # this is the important line
res = client.audio.transcriptions.create(model=self.model, language="zh", file=buffer)
return res.text

61
apps/setting/models_provider/impl/azure_model_provider/model/tti.py Normal file
View File

@ -0,0 +1,61 @@
from typing import Dict
from openai import OpenAI, AzureOpenAI
from common.config.tokenizer_manage_config import TokenizerManage
from setting.models_provider.base_model_provider import MaxKBBaseModel
from setting.models_provider.impl.base_tti import BaseTextToImage
def custom_get_token_ids(text: str):
tokenizer = TokenizerManage.get_tokenizer()
return tokenizer.encode(text)
class AzureOpenAITextToImage(MaxKBBaseModel, BaseTextToImage):
api_base: str
api_key: str
api_version: str
model: str
params: dict
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.api_key = kwargs.get('api_key')
self.api_base = kwargs.get('api_base')
self.api_version = kwargs.get('api_version')
self.model = kwargs.get('model')
self.params = kwargs.get('params')
@staticmethod
def new_instance(model_type, model_name, model_credential: Dict[str, object], **model_kwargs):
optional_params = {'params': {'size': '1024x1024', 'quality': 'standard', 'n': 1}}
for key, value in model_kwargs.items():
if key not in ['model_id', 'use_local', 'streaming']:
optional_params['params'][key] = value
return AzureOpenAITextToImage(
model=model_name,
api_base=model_credential.get('api_base'),
api_key=model_credential.get('api_key'),
api_version=model_credential.get('api_version'),
**optional_params,
)
def is_cache_model(self):
return False
def check_auth(self):
chat = AzureOpenAI(api_key=self.api_key, azure_endpoint=self.api_base, api_version=self.api_version)
response_list = chat.models.with_raw_response.list()
# self.generate_image('生成一个小猫图片')
def generate_image(self, prompt: str, negative_prompt: str = None):
chat = AzureOpenAI(api_key=self.api_key, azure_endpoint=self.api_base, api_version=self.api_version)
res = chat.images.generate(model=self.model, prompt=prompt, **self.params)
file_urls = []
for content in res.data:
url = content.url
file_urls.append(url)
return file_urls

69
apps/setting/models_provider/impl/azure_model_provider/model/tts.py Normal file
View File

@ -0,0 +1,69 @@
from typing import Dict
from dashscope import api_key
from dashscope.common.env import api_version
from openai import OpenAI, AzureOpenAI
from common.config.tokenizer_manage_config import TokenizerManage
from setting.models_provider.base_model_provider import MaxKBBaseModel
from setting.models_provider.impl.base_tts import BaseTextToSpeech
def custom_get_token_ids(text: str):
tokenizer = TokenizerManage.get_tokenizer()
return tokenizer.encode(text)
class AzureOpenAITextToSpeech(MaxKBBaseModel, BaseTextToSpeech):
api_base: str
api_key: str
api_version: str
model: str
params: dict
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.api_key = kwargs.get('api_key')
self.api_base = kwargs.get('api_base')
self.api_version = kwargs.get('api_version')
self.model = kwargs.get('model')
self.params = kwargs.get('params')
@staticmethod
def new_instance(model_type, model_name, model_credential: Dict[str, object], **model_kwargs):
optional_params = {'params': {'voice': 'alloy'}}
for key, value in model_kwargs.items():
if key not in ['model_id', 'use_local', 'streaming']:
optional_params['params'][key] = value
return AzureOpenAITextToSpeech(
model=model_name,
api_base=model_credential.get('api_base'),
api_key=model_credential.get('api_key'),
api_version=model_credential.get('api_version'),
**optional_params,
)
def check_auth(self):
client = AzureOpenAI(
azure_endpoint=self.api_base,
api_key=self.api_key,
api_version=self.api_version
)
response_list = client.models.with_raw_response.list()
# print(response_list)
def text_to_speech(self, text):
client = AzureOpenAI(
azure_endpoint=self.api_base,
api_key=self.api_key,
api_version=self.api_version
)
with client.audio.speech.with_streaming_response.create(
model=self.model,
input=text,
**self.params
) as response:
return response.read()
def is_cache_model(self):
return False