test_providers.py 1.1 KB

1234567891011121314151617181920212223242526272829303132
  1. from g4f.Provider import __all__, ProviderUtils
  2. from g4f import ChatCompletion
  3. import concurrent.futures
  4. _ = [
  5. 'BaseProvider',
  6. 'AsyncProvider',
  7. 'AsyncGeneratorProvider',
  8. 'RetryProvider'
  9. ]
  10. def test_provider(provider):
  11. try:
  12. provider = (ProviderUtils.convert[provider])
  13. if provider.working and not provider.needs_auth:
  14. print('testing', provider.__name__)
  15. completion = ChatCompletion.create(model='gpt-3.5-turbo',
  16. messages=[{"role": "user", "content": "hello"}], provider=provider)
  17. return completion, provider.__name__
  18. except Exception as e:
  19. #print(f'Failed to test provider: {provider} | {e}')
  20. return None
  21. with concurrent.futures.ThreadPoolExecutor() as executor:
  22. futures = [
  23. executor.submit(test_provider, provider)
  24. for provider in __all__
  25. if provider not in _
  26. ]
  27. for future in concurrent.futures.as_completed(futures):
  28. if result := future.result():
  29. print(f'{result[1]} | {result[0]}')