| ) -> Generator[Union[str, AgentScratchpadUnit.Action], None, None]: | ) -> Generator[Union[str, AgentScratchpadUnit.Action], None, None]: | ||||
| def parse_action(json_str): | def parse_action(json_str): | ||||
| try: | try: | ||||
| action = json.loads(json_str) | |||||
| action = json.loads(json_str, strict=False) | |||||
| action_name = None | action_name = None | ||||
| action_input = None | action_input = None | ||||
| from openai import OpenAI | from openai import OpenAI | ||||
| from core.model_runtime.entities.common_entities import I18nObject | |||||
| from core.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType | |||||
| from core.model_runtime.errors.validate import CredentialsValidateFailedError | from core.model_runtime.errors.validate import CredentialsValidateFailedError | ||||
| from core.model_runtime.model_providers.__base.speech2text_model import Speech2TextModel | from core.model_runtime.model_providers.__base.speech2text_model import Speech2TextModel | ||||
| from core.model_runtime.model_providers.openai._common import _CommonOpenAI | from core.model_runtime.model_providers.openai._common import _CommonOpenAI | ||||
| response = client.audio.transcriptions.create(model=model, file=file) | response = client.audio.transcriptions.create(model=model, file=file) | ||||
| return response.text | return response.text | ||||
| def get_customizable_model_schema(self, model: str, credentials: dict) -> AIModelEntity | None: | |||||
| """ | |||||
| used to define customizable model schema | |||||
| """ | |||||
| entity = AIModelEntity( | |||||
| model=model, | |||||
| label=I18nObject(en_US=model), | |||||
| fetch_from=FetchFrom.CUSTOMIZABLE_MODEL, | |||||
| model_type=ModelType.SPEECH2TEXT, | |||||
| model_properties={}, | |||||
| parameter_rules=[], | |||||
| ) | |||||
| return entity |
| import requests | import requests | ||||
| from core.model_runtime.entities.common_entities import I18nObject | |||||
| from core.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType | |||||
| from core.model_runtime.errors.invoke import InvokeBadRequestError | from core.model_runtime.errors.invoke import InvokeBadRequestError | ||||
| from core.model_runtime.errors.validate import CredentialsValidateFailedError | from core.model_runtime.errors.validate import CredentialsValidateFailedError | ||||
| from core.model_runtime.model_providers.__base.speech2text_model import Speech2TextModel | from core.model_runtime.model_providers.__base.speech2text_model import Speech2TextModel | ||||
| self._invoke(model, credentials, audio_file) | self._invoke(model, credentials, audio_file) | ||||
| except Exception as ex: | except Exception as ex: | ||||
| raise CredentialsValidateFailedError(str(ex)) | raise CredentialsValidateFailedError(str(ex)) | ||||
| def get_customizable_model_schema(self, model: str, credentials: dict) -> AIModelEntity | None: | |||||
| """ | |||||
| used to define customizable model schema | |||||
| """ | |||||
| entity = AIModelEntity( | |||||
| model=model, | |||||
| label=I18nObject(en_US=model), | |||||
| fetch_from=FetchFrom.CUSTOMIZABLE_MODEL, | |||||
| model_type=ModelType.SPEECH2TEXT, | |||||
| model_properties={}, | |||||
| parameter_rules=[], | |||||
| ) | |||||
| return entity |
| result = [] | result = [] | ||||
| outputs = data.get("outputs", {}) | |||||
| outputs, files = self._extract_files(outputs) | |||||
| for file in files: | |||||
| result.append(self.create_file_var_message(file)) | |||||
| outputs = data.get("outputs") | |||||
| if outputs == None: | |||||
| outputs = {} | |||||
| else: | |||||
| outputs, files = self._extract_files(outputs) | |||||
| for file in files: | |||||
| result.append(self.create_file_var_message(file)) | |||||
| result.append(self.create_text_message(json.dumps(outputs, ensure_ascii=False))) | result.append(self.create_text_message(json.dumps(outputs, ensure_ascii=False))) | ||||
| result.append(self.create_json_message(outputs)) | result.append(self.create_json_message(outputs)) |