2 files changed, 69 insertions, 34 deletions

diff --git a/g4f/Provider/base_provider.py b/g4f/Provider/base_provider.py
index 79f8f617..9d45aa44 100644
--- a/g4f/Provider/base_provider.py
+++ b/g4f/Provider/base_provider.py
@@ -25,6 +25,7 @@ class BaseProvider(ABC):
         
         raise NotImplementedError()
 
+
     @classmethod
     @property
     def params(cls):
@@ -46,6 +47,8 @@ class AsyncProvider(BaseProvider):
         stream: bool = False,
         **kwargs
     ) -> CreateResult:
+        check_running_loop()
+
         yield asyncio.run(cls.create_async(model, messages, **kwargs))
 
     @staticmethod
@@ -67,10 +70,17 @@ class AsyncGeneratorProvider(AsyncProvider):
         stream: bool = True,
         **kwargs
     ) -> CreateResult:
-        loop = asyncio.new_event_loop()
+        check_running_loop()
+
+        # Force use selector event loop on windows
+        loop = asyncio.SelectorEventLoop()
         try:
-            asyncio.set_event_loop(loop)
-            generator = cls.create_async_generator(model, messages, stream=stream, **kwargs)
+            generator = cls.create_async_generator(
+                model,
+                messages,
+                stream=stream,
+                **kwargs
+            )
             gen  = generator.__aiter__()
             while True:
                 try:
@@ -78,10 +88,8 @@ class AsyncGeneratorProvider(AsyncProvider):
                 except StopAsyncIteration:
                     break
         finally:
-            asyncio.set_event_loop(None)
             loop.close()
 
-
     @classmethod
     async def create_async(
         cls,
@@ -100,6 +108,11 @@ class AsyncGeneratorProvider(AsyncProvider):
     ) -> AsyncGenerator:
         raise NotImplementedError()
 
+# Don't create a new loop in a running loop
+def check_running_loop():
+    if asyncio.events._get_running_loop() is not None:
+        raise RuntimeError(
+            'Use "create_async" instead of "create" function in a async loop.')
 
 _cookies = {}
 
diff --git a/g4f/__init__.py b/g4f/__init__.py
index a49e60e9..e42be8a8 100644
--- a/g4f/__init__.py
+++ b/g4f/__init__.py
@@ -1,11 +1,42 @@
 from __future__ import annotations
 from g4f        import models
-from .Provider  import BaseProvider
+from .Provider  import BaseProvider, AsyncProvider
 from .typing    import Any, CreateResult, Union
 import random
 
 logging = False
 
+def get_model_and_provider(model: Union[models.Model, str], provider: type[BaseProvider], stream: bool):
+    if isinstance(model, str):
+        if model in models.ModelUtils.convert:
+            model = models.ModelUtils.convert[model]
+        else:
+            raise Exception(f'The model: {model} does not exist')
+
+    if not provider:
+        if isinstance(model.best_provider, list):
+            if stream:
+                provider = random.choice([p for p in model.best_provider if p.supports_stream])
+            else:
+                provider = random.choice(model.best_provider)
+        else:
+            provider = model.best_provider
+
+    if not provider:
+        raise Exception(f'No provider found for model: {model}')
+    
+    if not provider.working:
+        raise Exception(f'{provider.__name__} is not working')
+    
+    if not provider.supports_stream and stream:
+        raise Exception(
+                f'ValueError: {provider.__name__} does not support "stream" argument')
+    
+    if logging:
+        print(f'Using {provider.__name__} provider')
+
+    return model, provider
+
 class ChatCompletion:
     @staticmethod
     def create(
@@ -13,28 +44,11 @@ class ChatCompletion:
         messages : list[dict[str, str]],
         provider : Union[type[BaseProvider], None] = None,
         stream   : bool                            = False,
-        auth     : Union[str, None]                = None, **kwargs: Any) -> Union[CreateResult, str]:
+        auth     : Union[str, None]                = None,
+        **kwargs
+    ) -> Union[CreateResult, str]:
 
-        if isinstance(model, str):
-            if model in models.ModelUtils.convert:
-                model = models.ModelUtils.convert[model]
-            else:
-                raise Exception(f'The model: {model} does not exist')
-
-        if not provider:
-            if isinstance(model.best_provider, list):
-                if stream:
-                    provider = random.choice([p for p in model.best_provider if p.supports_stream])
-                else:
-                    provider = random.choice(model.best_provider)
-            else:
-                provider = model.best_provider
-
-        if not provider:
-            raise Exception(f'No provider found')
-
-        if not provider.working:
-            raise Exception(f'{provider.__name__} is not working')
+        model, provider = get_model_and_provider(model, provider, stream)
 
         if provider.needs_auth and not auth:
             raise Exception(
@@ -43,12 +57,20 @@ class ChatCompletion:
         if provider.needs_auth:
             kwargs['auth'] = auth
 
-        if not provider.supports_stream and stream:
-            raise Exception(
-                f'ValueError: {provider.__name__} does not support "stream" argument')
-
-        if logging:
-            print(f'Using {provider.__name__} provider')
-
         result = provider.create_completion(model.name, messages, stream, **kwargs)
         return result if stream else ''.join(result)
+
+    @staticmethod
+    async def create_async(
+        model    : Union[models.Model, str],
+        messages : list[dict[str, str]],
+        provider : Union[type[BaseProvider], None] = None,
+        **kwargs
+    ) -> str:
+        
+        model, provider = get_model_and_provider(model, provider, False)
+
+        if not issubclass(provider, AsyncProvider):
+            raise Exception(f"Provider: {provider.__name__} doesn't support create_async")
+
+        return await provider.create_async(model.name, messages, **kwargs)