tmp

.gitignore

@@ -1,3 +1,4 @@
 .*
 !.gitignore
+logs/*
 __pycache__

config/dashscope.yml

@@ -13,10 +13,13 @@ tongyiqianwen:
   class_name: ChatAiTongYiQianWen
   api_key: ${DASH_SCOPE_API_KEY}    # must be set when using this api
   model_name: qwen2-1.5b-instruct   # you can change it
-  is_stream: true
+  use_stream_api: true
+  return_as_stream: true
 
 sambert:
   path: dashscope/sambert
   class_name: TTSSambert
   api_key: ${DASH_SCOPE_API_KEY}    # must be set when using this api
   model_name: sambert-zhichu-v1     # you can change it
+  # is_stream: true
+  return_as_stream: true

config/sounds.yml

@@ -0,0 +1,16 @@
+# sounds.yml
+
+root_path: ../src/offical
+
+sounds_play:
+  path: sounds/sounds_play_engine
+  class_name: SoundsPlayEngine
+  module: pyaudio
+  sounds: wav
+  # is_stream: true
+
+sounds_record:
+  path: sounds/sounds_recode_engine
+  class_name: SoundsRecordEngine
+  module: pyaudio
+  sounds: wav

main.py

@@ -1,52 +1,28 @@
 from src import ExecutePipeline
 from src import Plugins
-from src import EchoEngine
+from src import EchoEngine, TeeEngine
-from pathlib import Path
 
 def main():
     from plugins.example import MyEngine
+    from pathlib import Path
 
     plug_conf = Plugins(Path(__file__).resolve().parent)
     pipe = ExecutePipeline(
-        plug_conf.load_engine("asr_engine"),
+        # plug_conf.load_engine("asr_engine"),
-        MyEngine(),
+        # MyEngine(),
         plug_conf.load_engine("chat_ai_engine"),
-        EchoEngine(),
+        TeeEngine(),
         plug_conf.load_engine("tts_engine"),
+        plug_conf.load_engine("sounds_play_engine"),
+        EchoEngine()
         )
-
+    # exe_input = './tests/offical/sounds/asr_example.wav'
-    res = pipe.execute_engines('./tests/offical/sounds/asr_example.wav')
+    # exe_input = input('input: ')
-    import wave
+    exe_input = '你好'
-    import io
+    res = pipe.execute_engines(exe_input)
-    import pyaudio
+    print(res)
-    # with open('output.wav', 'wb') as f:
-    #     f.write(res)
-
-    wf = wave.open(io.BytesIO(res), 'rb')
-    _audio  = pyaudio.PyAudio()
-    _stream = _audio.open(format=_audio.get_format_from_width(wf.getsampwidth()),
-                channels=wf.getnchannels(),
-                rate=wf.getframerate(),
-                output=True)
-    data = wf.readframes(1024)  # 假定块大小为1024
-    while data:
-        _stream.write(data)
-        data = wf.readframes(1024)
-
-    _stream.stop_stream()
-    _stream.close()
-
-    _audio.terminate()
-    wf.close()
 
 if __name__ == '__main__':
     from dotenv import load_dotenv
     load_dotenv()
-    # import importlib
-    # from pathlib import Path
-    # pug_path = Path(__file__).resolve().parent / "src/engine"
-    # import sys
-    # sys.path.append(str(pug_path))
-    # importlib.import_module("dashscope",
-    #                         str(pug_path))
     main()

plugins/example.py

@@ -1,9 +1,7 @@
-from typing import override
 from src import Engine
 
 class MyEngine(Engine):
-    @override
+    def __init__(self):
-    def __init__(is_stream: bool = False):
         super().__init__()
     
     def execute(self, data):

requirements.txt

@@ -5,5 +5,5 @@ python-dotenv
 # server optional
 # tornado
 
-# tqdm
+tqdm
-# rich # optional
+rich # optional

src/__init__.py

@@ -6,12 +6,27 @@ from .core.core import Engine as Engine
 from .core.core import ExecutePipeline as ExecutePipeline
 from .core.echo import EchoMiddleware as EchoMiddleware
 from .core.echo import EchoEngine as EchoEngine
+from .core.tee import TeeEngine as TeeEngine
 
+from .engine.stream_engine import StreamEngine as StreamEngine
 from .engine.asr_engine import ASREngine as ASREngine
 from .engine.tts_engine import TTSEngine as TTSEngine
 from .engine.nlu_engine import NLUEngine as NLUEngine
 from .engine.chat_ai_engine import ChatAIEngine as ChatAIEngine
 
+
 from .plugins.dynamic_package_import import dynamic_package_import as dynamic_package_import
 from .plugins.plugins_conf import PluginsConfig as PluginsConfig
 from .plugins.plugins import Plugins as Plugins
+
+# from .utils.logger import setup_logger as setup_logger
+import logging
+from sys import stdout
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+stream_handle = logging.StreamHandler(stdout)
+formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s',
+                              datefmt="%Y-%m-%d %H:%M:%S")
+stream_handle.setFormatter(formatter)
+logger.addHandler(stream_handle)
+logger.propagate = False

src/core/core.py

@@ -31,7 +31,6 @@ class Middleware(MiddlewareInterface):
                     self._data = middleware.process(self._data)
             return None if self._next_middleware is None else data
         except Exception as e:
-            logger.exception(f"Error occurred during processing: {e}")
             raise e
 
     def get_next(self) -> Optional[List[MiddlewareInterface]]:
@@ -72,8 +71,7 @@ class Engine(EngineInterface):
     ```
 
     """
-    def __init__(self, is_stream: bool = False) -> None:
+    def __init__(self) -> None:
-        self._is_stream = is_stream
         self._metadata = None
 
     def prepare_input(self, data: Any) -> Any:
@@ -158,14 +156,12 @@ class ExecutePipeline:
             return data
         except ValueError:
             return None
-        except Exception as e:
-            raise e
 
-    def execute_engines(self, data: Any) -> Any:
+    def execute_engines(self, data: Any, metadata = None) -> Any:
         """
         执行引擎
         """
-        res = self.execute({'__data__': data, '__metadata__': None})
+        res = self.execute({'__data__': data, '__metadata__': metadata})
         if not res:
             return None
         else:

src/core/echo.py

@@ -1,5 +1,5 @@
 import sys
-from typing import Iterator, List, Optional
+from typing import Generator, Iterator, List, Optional
 from .. import Middleware, MiddlewareInterface, Engine
 
 class EchoMiddleware(Middleware):
@@ -20,7 +20,7 @@ class EchoMiddleware(Middleware):
 
 class EchoEngine(Engine):
     def execute(self, data):
-        if isinstance(data, Iterator):
+        if isinstance(data, Generator):
             ret = ''
             for i in data:
                 ret += i

src/core/tee.py

@@ -0,0 +1,19 @@
+import sys
+from typing import Iterator, List, Optional
+from .. import Middleware, MiddlewareInterface, Engine
+
+class TeeEngine(Engine):
+    def execute(self, data):
+        if isinstance(data, Iterator):
+            def _():
+                for item in data:
+                    sys.stdout.write(str(item))
+                    sys.stdout.flush()
+                    yield item
+                sys.stdout.write('\n')
+                sys.stdout.flush()
+            ret = _()
+        else:
+            ret = data
+            print(data)
+        return ret

src/engine/asr_engine.py

@@ -1,6 +1,5 @@
-from ..core import Engine
+from .. import Engine
 
 class ASREngine(Engine):
-    def __init__(self, is_stream: bool = False) -> None:
+    def __init__(self) -> None:
-        super().__init__(is_stream)
+        super().__init__()
-    pass

src/engine/chat_ai_engine.py

@@ -1,10 +1,10 @@
-from typing import Dict
+from typing import Dict, Optional
 from logging import getLogger
 logger = getLogger(__name__)
-from ..core import Engine
+from .. import Engine
 
 class ChatAIEngine(Engine):
     """基础AI引擎类, 提供历史记录管理的基础框架。"""
-    def __init__(self, history:Dict = None, is_stream = False) -> None:
+    def __init__(self, history: Optional[Dict] = None) -> None:
-        super().__init__(is_stream)
         self._history = history
+        Engine.__init__(self)

src/engine/nlu_engine.py

@@ -1,4 +1,4 @@
-from ..core import Engine
+from .. import Engine
 
 class NLUEngine(Engine):
     def __init__(self, is_stream: bool = False) -> None:

src/engine/stream_engine.py

@@ -0,0 +1,56 @@
+from typing import Any, Generator, Iterable, Iterator
+from .. import Engine
+
+class StreamEngine(Engine):
+    def __init__(self, use_stream_api: bool = False, return_as_stream: bool = False):
+        self._use_stream_api = use_stream_api
+        self._return_as_stream = return_as_stream
+
+    def execute_stream(self, data) -> Generator | Iterator:
+        raise NotImplementedError
+
+    def execute_nonstream(self, data) -> Any:
+        raise NotImplementedError
+
+    def _execute_stream(self, data: Any, return_as_stream: bool) -> Any:
+        results = self.execute_stream(data)
+        return results if return_as_stream else list(results)
+
+    def _execute_nonstream(self, data: Any, return_as_stream: bool) -> Any:
+        result = self.execute_nonstream(data)
+        if return_as_stream:
+            def _():
+                yield from result
+            return _()
+        else:
+            return result
+
+    def execute(self, data: Any) -> Any:
+        if not isinstance(data, Generator) or isinstance(data, bytes):
+            if self._use_stream_api:
+                return self._execute_stream([data], self._return_as_stream)
+            else:
+                return self._execute_nonstream(data, self._return_as_stream)
+        else:
+            if self._use_stream_api:
+                if self._return_as_stream:
+                    def stream_results():
+                        for item in data:
+                            yield from self._execute_stream([item], True)
+                    return stream_results()
+                else:
+                    return [self._execute_stream([item], False) for item in data]
+            else:
+                if self._return_as_stream:
+                    def non_stream_results():
+                        for item in data:
+                            yield self._execute_nonstream(item, False)
+                    return non_stream_results()
+                else:
+                    res = self._execute_nonstream(next(data), False)
+                    for item in data:
+                        _ = self._execute_nonstream(item, False)
+                        if _ is None:
+                            continue
+                        res += _
+                    return res

src/engine/tts_engine.py

@@ -1,4 +1,4 @@
-from ..core import Engine
+from .. import Engine
 
 class TTSEngine(Engine):
     def __init__(self, is_stream: bool = False) -> None:

src/offical/dashscope/__init__.py

@@ -1,4 +1,5 @@
-print("dashscope plugins start")
+import logging
+logger = logging.getLogger(__name__)
 
 from src import dynamic_package_import
 dynamic_package_import([

src/offical/dashscope/paraformer.py

@@ -24,7 +24,7 @@ class ASRParaformer(ASREngine):
                  _model : str = "paraformer-realtime-v1",
                  _is_stream : bool = False,
                  *args, **kwargs) -> None:
-        super().__init__(kwargs.get("is_stream", _is_stream))
+        super().__init__()
         dashscope.api_key = api_key
         self.recognition = Recognition(kwargs.get("model", _model),
                           format='wav',

src/offical/dashscope/sambert.py

@@ -1,12 +1,16 @@
 # https://help.aliyun.com/zh/dashscope/developer-reference/quick-start-13?spm=a2c4g.11186623.0.0.26772e5cs8Vl59
 
-from typing import Any
+import sys
+from typing import Any, Generator, Iterable, Iterator
 from src import TTSEngine as TTSEngine
 
 import dashscope
-from dashscope.audio.tts import SpeechSynthesizer
+from dashscope.api_entities.dashscope_response import SpeechSynthesisResponse
+from dashscope.audio.tts import ResultCallback, SpeechSynthesizer, SpeechSynthesisResult
 
 from logging import getLogger
+
+from src import StreamEngine as StreamEngine
 logger = getLogger(__name__)
 
 # import requests
@@ -16,23 +20,64 @@ logger = getLogger(__name__)
 # with open('asr_example.wav', 'wb') as f:
 #     f.write(r.content)
 
-class TTSSambert(TTSEngine):
+class TTSSambert(TTSEngine, StreamEngine):
 
     def __init__(self, api_key : str,
                  _model : str = "sambert-zhichu-v1",
                  _is_stream : bool = False,
                  *args, **kwargs) -> None:
-        super().__init__(kwargs.get("is_stream", _is_stream))
+        TTSEngine.__init__(self)
+        StreamEngine.__init__(self,
+                              use_stream_api=kwargs.get('use_stream_api', False),
+                              return_as_stream=kwargs.get('return_as_stream', False))
         dashscope.api_key = api_key
         self.model = kwargs.get("model", _model)
-    def process_output(self, data):
-        return super().process_output(data)
 
-    def execute(self, data: Any) -> Any:
+    class Callback(ResultCallback):
+        def __init__(self, generator) -> None:
+            super().__init__()
+            self.generator = generator
+        def on_open(self):
+            logger.debug('Speech synthesizer is opened.')
+
+        def on_complete(self):
+            logger.debug('Speech synthesizer is completed.')
+
+        def on_error(self, response: SpeechSynthesisResponse):
+            logger.error('Speech synthesizer failed, response is %s' % (str(response)))
+        def on_close(self):
+            self.generator.send(None)
+            logger.debug('Speech synthesizer is closed.')
+
+        def on_event(self, result: SpeechSynthesisResult):
+            if result.get_audio_frame() is not None:
+                logger.debug('audio result length:', sys.getsizeof(result.get_audio_frame()))
+                res = result.get_audio_frame()
+                self.generator.send(res)
+
+            if result.get_timestamp() is not None:
+                logger.debug('timestamp result:', str(result.get_timestamp()))
+
+    def execute_nonstream(self, data) -> bytes:
         result = SpeechSynthesizer.call(model=self.model,
                                 text=data,
                                 sample_rate=48000)
-        if result.get_audio_data() is not None:
         return result.get_audio_data()
-        else:
+    
-            raise RuntimeError('Error: ' + result.message)
+    # def execute_stream(self, data) -> Generator[Any, None, None] | Iterator:
+    #     pass
+    #     if self._is_stream:
+    #         def audio_generator():
+    #             while True:
+    #                 data = yield
+    #                 if data is None:
+    #                     break
+    #         gen = audio_generator()
+    #         next(gen)
+    #         callback = self.Callback(gen)
+    #         SpeechSynthesizer.call(model=self.model,
+    #                             text=data,
+    #                             sample_rate=48000,
+    #                             callback=callback,
+    #                             word_timestamp_enabled=True,
+    #                             phoneme_timestamp_enabled=True)

src/offical/requests/__init__.py

@@ -1,4 +1,3 @@
-print("requests plugins start")
 
 from src import dynamic_package_import
 dynamic_package_import([

src/offical/requests/remote_engine.py

@@ -1,4 +1,5 @@
 from logging import getLogger
+from typing import Any, Dict
 logger = getLogger(__name__)
 import requests
 
@@ -13,7 +14,7 @@ class RemoteEngine():
             'Authorization': 'Bearer ' + self._api_key
         }
 
-    def post_prompt(self, messages : list[dict[str, str]], is_stream : bool = False) -> requests.Response:
+    def post_prompt(self, messages : Dict, is_stream : bool = False) -> requests.Response:
         try:
             response = requests.post(self.url, headers=self.header, json=messages, stream=is_stream)
             if (response.status_code != 200):

src/offical/requests/tongyiqianwen.py

@@ -1,20 +1,21 @@
 # https://dashscope.console.aliyun.com/model
 import json
-from typing import Any
+from typing import Any, Generator, Iterator
 from .remote_engine import RemoteEngine
 from src import ChatAIEngine as ChatAIEngine
+from src import StreamEngine as StreamEngine
 
 from logging import getLogger
 logger = getLogger(__name__)
 
-class ChatAiTongYiQianWen(ChatAIEngine, RemoteEngine):
+class ChatAiTongYiQianWen(ChatAIEngine, RemoteEngine, StreamEngine):
     API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
     def __init__(self, api_key : str,
                  _model : str = "qwen-1.8b-chat",
-                 _is_stream : bool = False,
                  *args, **kwargs) -> None:
         self._modual_name = kwargs.get("model", _model)
-        ChatAIEngine.__init__(self, is_stream=kwargs.get("is_stream", _is_stream))
+        StreamEngine.__init__(self, use_stream_api=kwargs.get("use_stream_api", False), return_as_stream=kwargs.get("return_as_stream", False))
+        ChatAIEngine.__init__(self)
         RemoteEngine.__init__(self, api_key=api_key,
                          base_url=self.API_URL)
     def _transform_raw(self, response):
@@ -26,13 +27,7 @@ class ChatAiTongYiQianWen(ChatAIEngine, RemoteEngine):
             json_string = input_string[input_string.find("data:") + 5:].strip()
             yield self._transform_raw(json.loads(json_string))
 
-    def process_output(self, data):
+    def execute_stream(self, data) -> Generator[Any, None, None] | Iterator:
-        res =  self._transform_iterator(data) if self._is_stream else\
-            self._transform_raw(data)
-        return super().process_output(res)
-
-    def _execute_raw(self, data: Any) -> Any:
-        if self._is_stream:
         self.header['X-DashScope-SSE'] = 'enable'
         response = self.post_prompt({
                 'model': self._modual_name,
@@ -41,15 +36,29 @@ class ChatAiTongYiQianWen(ChatAIEngine, RemoteEngine):
                 },
                 "parameters": {
                     "result_format": "message",
-                    "incremental_output": self._is_stream
+                    "incremental_output": True
                 }
             },
-            is_stream=self._is_stream)
+            is_stream=True)
-        if self._is_stream:
-            ret = response.iter_content(chunk_size=None)
-        else:
-            ret = response.json()
-        return ret
 
-    def execute(self, data: Any) -> Any:
+        ret = response.iter_content(chunk_size=None)
-        return self._execute_raw([{'role': 'user', 'content': f'{data}'}])
+        return self._transform_iterator(ret)
+    
+    def execute_nonstream(self, data) -> Any:
+        response = self.post_prompt({
+                'model': self._modual_name,
+                "input": {
+                    "messages": data
+                },
+                "parameters": {
+                    "result_format": "message",
+                    "incremental_output": False
+                }
+            },
+            is_stream=False)
+        ret = response.json()
+        return self._transform_raw(ret)
+
+    def prepare_input(self, data: Any) -> Any:
+        data['__data__'] = {'role': 'user', 'content': f'{data['__data__']}'}
+        return super().prepare_input(data)

src/offical/sounds/__init__.py

@@ -1,5 +1,3 @@
-print("sounds plugins start")
-
 from src import dynamic_package_import
 dynamic_package_import([
     ('pyaudio', None),

src/offical/sounds/sounds_play_engine.py

@@ -0,0 +1,35 @@
+from typing import Any, Generator
+from src import Engine
+import wave
+import io
+import pyaudio
+
+from src import StreamEngine
+
+class SoundsPlayEngine(StreamEngine):
+    def __init__(self, **kwargs) -> None:
+        # super().__init__(kwargs.get("is_stream", _is_stream))
+        StreamEngine.__init__(self,
+                              use_stream_api=kwargs.get('use_stream_api', False),
+                              return_as_stream=kwargs.get('return_as_stream', False))
+
+    def execute_nonstream(self, data) -> Any:
+        if data is None:
+            return None
+        wf = wave.open(io.BytesIO(data), 'rb')
+        _audio  = pyaudio.PyAudio()
+        _stream = _audio.open(
+            format=_audio.get_format_from_width(wf.getsampwidth()),
+            channels=wf.getnchannels(),
+            rate=wf.getframerate(),
+            output=True)
+        data = wf.readframes(1024)  # 假定块大小为1024
+        while data:
+            _stream.write(data)
+            data = wf.readframes(1024)
+
+        _stream.stop_stream()
+        _stream.close()
+
+        _audio.terminate()
+        wf.close()

src/offical/sounds/sounds_record_engine.py

@@ -0,0 +1,2 @@
+class SoundsRecordEngine:
+    pass

src/offical/sounds/sounds_wapper.py

@@ -1,4 +0,0 @@
-import pyaudio
-
-class SoundsWapper:
-    pass

src/offical/sounds/sounds_wrapper.py

@@ -0,0 +1,9 @@
+import pyaudio
+
+class SoundsWrapper:
+    def __init__(self, chunk=1024, format=pyaudio.paInt16, channels=1, rate=44100, input=True, output=True, input_device_index=None, output_device_index=None) -> None:
+        self.chunk = chunk
+        self.format = format
+        self.channels = channels
+        self.rate = rate
+        self.input =input

src/plugins/dynamic_package_import.py

@@ -13,9 +13,11 @@ dynamic_package_import(required_packages)
 import importlib.metadata
 import subprocess
 import sys
-from typing import List, Tuple
+from typing import List, Optional, Tuple
+import logging
+logger = logging.getLogger(__name__)
 
-def install_or_upgrade_package(package: str, version: str = None) -> None:
+def install_or_upgrade_package(package: str, version: Optional[str] = None) -> None:
     """
     Installs or upgrades a Python package using pip via Popen.
     
@@ -26,17 +28,15 @@ def install_or_upgrade_package(package: str, version: str = None) -> None:
     try:
         dist = importlib.metadata.distribution(package)
         current_version = dist.version
-        print(f"{package} is already installed at version {current_version}.")
         if version and current_version != version:
-            print(f"Upgrading {package} to version {version}.")
+            logger.info(f"Upgrading {package} to version {version}.")
             _execute_pip_command(f"install {package}=={version}")
         else:
-            print("No upgrade needed.")
+            logger.info(f"{package} is already installed at version {current_version}.")
     except importlib.metadata.PackageNotFoundError:
-        print(f"{package} is not installed. Installing...")
+        logger.info(f"{package} is not installed. Installing...")
         _execute_pip_command(f"install {package}{'==' + version if version else ''}")
 
-
 def _execute_pip_command(command: str) -> None:
     """
     Executes a pip command using subprocess.Popen.
@@ -44,16 +44,14 @@ def _execute_pip_command(command: str) -> None:
     :param command: The pip command to execute (e.g., 'install numpy').
     """
     pip_cmd = [sys.executable, "-m", "pip", *command.split()]
-    with subprocess.Popen(pip_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) as proc:
+    with subprocess.Popen(pip_cmd, stderr=subprocess.PIPE, universal_newlines=True) as proc:
-        stdout, stderr = proc.communicate()
+        _, stderr = proc.communicate()
         if proc.returncode != 0:
-            print(f"Error executing pip command: {command}")
+            logger.error(f"executing pip command: {command}\n\tdetails: {stderr}")
-            print(f"Error details: {stderr}")
         else:
-            print(f"Pip command '{command}' executed successfully.")
+            logger.info(f"Pip command '{command}' executed successfully.")
 
-
+def dynamic_package_import(required_packages: List[Tuple[str, Optional[str]]]) -> None:
-def dynamic_package_import(required_packages: List[Tuple[str, str]]) -> None:
     """
     Checks for the presence of required packages and installs/upgrades them if necessary.
     

src/plugins/plugins.py

@@ -8,7 +8,6 @@ from src import EngineInterface
 from .plugins_conf import PluginsConfig
 from .plugins_import import PluginsImport
 
-
 class Plugins:
     def __init__(self,
                  base_path : Path | str = CURRENT_DIR_PATH,
@@ -22,10 +21,18 @@ class Plugins:
         # self.core_path:Path = self.base_path / (global_conf.get('core_path', '') or '../')
         # self.plugin_path:Path = self.base_path / (global_conf.get('plugin_path', '') or './')
         # sys.path.append(str(self.base_path))
-    def load_engine(self, engine_name: str) -> EngineInterface:
+
+    def _get_engine_import(self, engine_name: str):
         engine_conf, plg_root_path, plg_path = PluginsConfig.get_engine_conf(engine_name, self.base_path, self.conf_path)
         engine_conf.get('plugin_path', '')
         plg_import = PluginsImport(plg_root_path)
         plg_import.get_module(engine_conf['class_name'], plg_path)
-        n = engine_conf['class_name']
+        return plg_import, engine_conf
-        return plg_import.get_instance(n, n, **engine_conf)
+
+    def load_engine_class(self, engine_name: str):
+        plg_import, engine_conf = self._get_engine_import(engine_name)
+        return plg_import.get_class(engine_conf['class_name'], engine_conf['class_name'])
+
+    def load_engine(self, engine_name: str) -> EngineInterface:
+        plg_import, engine_conf = self._get_engine_import(engine_name)
+        return plg_import.get_instance(engine_conf['class_name'], engine_conf['class_name'], **engine_conf)

src/plugins/plugins_conf.py

@@ -34,7 +34,7 @@ class PluginsConfig:
             return yaml.load(f, Loader=YamlEnvLoader)
 
     @staticmethod
-    def recursive_load(base_path: Path, config: Dict[str, Any]):
+    def recursive_load(base_path: Path, config: Dict[str, Any] | Any):
         """Recursively load configurations when 'plugin' and 'path' are present."""
         if isinstance(config, dict) and 'plugin' in config and 'path' in config:
             plugin_path:Path = base_path / config.get('path', '')