from __future__ import annotations import json import os import time import random import string import asyncio import aiohttp import base64 import requests from pathlib import Path from datetime import datetime from typing import Union, AsyncIterator, Iterator, Awaitable, Optional, List, Dict, Any, Type from ..image.copy_images import copy_media, get_media_dir from ..typing import Messages, ImageType from ..providers.types import ProviderType, BaseProvider from ..providers.response import * from ..errors import NoMediaResponseError, ProviderNotFoundError from ..providers.retry_provider import IterListProvider from ..providers.asyncio import to_sync_generator from ..providers.any_provider import AnyProvider from ..Provider import OpenaiAccount, PollinationsImage, ProviderUtils from ..Provider.template import OpenaiTemplate from ..tools.run_tools import async_iter_run_tools, iter_run_tools from ..cookies import get_cookies_dir from .stubs import ChatCompletion, ChatCompletionChunk, Image, ImagesResponse, UsageModel, ToolCallModel, ClientResponse from .models import ClientModels from .types import IterResponse, Client as BaseClient from .service import convert_to_provider from .helper import find_stop, filter_json, filter_none, safe_aclose from .. import debug ChatCompletionResponseType = Iterator[Union[ChatCompletion, ChatCompletionChunk, BaseConversation]] AsyncChatCompletionResponseType = AsyncIterator[Union[ChatCompletion, ChatCompletionChunk, BaseConversation]] try: anext # Python 3.8+ except NameError: async def anext(aiter): try: return await aiter.__anext__() except StopAsyncIteration: raise StopIteration def add_chunk(content, chunk): if content == "" and isinstance(chunk, (MediaResponse, AudioResponse)): content = chunk elif not isinstance(chunk, (Reasoning, ToolCalls)): content = str(content) + str(chunk) return content def resolve_media(kwargs: dict, image = None, image_name: str = None) -> None: if image is not None: kwargs["media"] = [(image, getattr(image, "name", image_name))] elif "images" in kwargs: kwargs["media"] = kwargs.pop("images") if kwargs.get("media") is None: kwargs.pop("media", None) elif not isinstance(kwargs["media"], list): kwargs["media"] = [kwargs["media"]] for idx, media in enumerate(kwargs.get("media", [])): if not isinstance(media, (list, tuple)): kwargs["media"][idx] = (media, getattr(media, "name", None)) # Synchronous iter_response function def iter_response( response: Union[Iterator[Union[str, ResponseType]]], stream: bool, response_format: Optional[dict] = None, max_tokens: Optional[int] = None, stop: Optional[list[str]] = None, provider_info: Optional[ProviderInfo] = None ) -> ChatCompletionResponseType: content = "" reasoning = [] finish_reason = None tool_calls = None usage = None provider_info: ProviderInfo = None conversation: JsonConversation = None completion_id = ''.join(random.choices(string.ascii_letters + string.digits, k=28)) idx = 0 if hasattr(response, '__aiter__'): response = to_sync_generator(response) for chunk in response: if isinstance(chunk, FinishReason): finish_reason = chunk.reason continue elif isinstance(chunk, JsonConversation): conversation = chunk continue elif isinstance(chunk, ToolCalls): if not stream: tool_calls = chunk.get_list() continue elif isinstance(chunk, Usage): usage = chunk continue elif isinstance(chunk, ProviderInfo): provider_info = chunk continue elif isinstance(chunk, Reasoning): reasoning.append(chunk) elif isinstance(chunk, (HiddenResponse, Exception, JsonRequest, JsonResponse)): continue elif not chunk: continue content = add_chunk(content, chunk) idx += 1 if max_tokens is not None and idx >= max_tokens: finish_reason = "length" first, content, chunk = find_stop(stop, content, chunk if stream else None) if first != -1: finish_reason = "stop" if stream: chunk = ChatCompletionChunk.model_construct(chunk, None, completion_id, int(time.time())) if provider_info is not None: chunk.provider = provider_info.name chunk.model = provider_info.model yield chunk if finish_reason is not None: break if usage is None: usage = UsageModel.model_construct(completion_tokens=idx, total_tokens=idx) else: usage = UsageModel.model_construct(**usage.get_dict()) finish_reason = "tool_calls" if tool_calls else ("stop" if finish_reason is None else finish_reason) if stream: chat_completion = ChatCompletionChunk.model_construct( None, finish_reason, completion_id, int(time.time()), usage=usage ) else: if response_format is not None and "type" in response_format: if response_format["type"] == "json_object": content = filter_json(content) chat_completion = ChatCompletion.model_construct( content, finish_reason, completion_id, int(time.time()), usage=usage, **(filter_none( tool_calls=[ToolCallModel.model_construct(**tool_call) for tool_call in tool_calls] ) if tool_calls is not None else {}), conversation=None if conversation is None else conversation.get_dict(), reasoning=reasoning if reasoning else None ) if provider_info is not None: chat_completion.provider = provider_info.name chat_completion.model = provider_info.model yield chat_completion async def async_iter_response( response: AsyncIterator[Union[str, ResponseType]], stream: bool, response_format: Optional[dict] = None, max_tokens: Optional[int] = None, stop: Optional[list[str]] = None, provider_info: Optional[ProviderInfo] = None ) -> AsyncChatCompletionResponseType: content = "" reasoning = [] finish_reason = None completion_id = ''.join(random.choices(string.ascii_letters + string.digits, k=28)) idx = 0 tool_calls = None usage = None conversation: JsonConversation = None try: async for chunk in response: if isinstance(chunk, FinishReason): finish_reason = chunk.reason continue elif isinstance(chunk, JsonConversation): conversation = chunk continue elif isinstance(chunk, ToolCalls): if not stream: tool_calls = chunk.get_list() continue elif isinstance(chunk, Usage): usage = chunk continue elif isinstance(chunk, ProviderInfo): provider_info = chunk continue elif isinstance(chunk, Reasoning) and not stream: reasoning.append(chunk) elif isinstance(chunk, (HiddenResponse, Exception, JsonRequest, JsonResponse)): continue elif not chunk: continue content = add_chunk(content, chunk) idx += 1 if max_tokens is not None and idx >= max_tokens: finish_reason = "length" first, content, chunk = find_stop(stop, content, chunk if stream else None) if first != -1: finish_reason = "stop" if stream: chunk = ChatCompletionChunk.model_construct(chunk, None, completion_id, int(time.time())) if provider_info is not None: chunk.provider = provider_info.name chunk.model = provider_info.model yield chunk if finish_reason is not None: break finish_reason = "tool_calls" if tool_calls else ("stop" if finish_reason is None else finish_reason) if usage is None: usage = UsageModel.model_construct(completion_tokens=idx, total_tokens=idx) else: usage = UsageModel.model_construct(**usage.get_dict()) if stream: chat_completion = ChatCompletionChunk.model_construct( None, finish_reason, completion_id, int(time.time()), usage=usage, conversation=conversation ) else: if response_format is not None and "type" in response_format: if response_format["type"] == "json_object": content = filter_json(content) chat_completion = ChatCompletion.model_construct( content, finish_reason, completion_id, int(time.time()), usage=usage, **(filter_none( tool_calls=[ToolCallModel.model_construct(**tool_call) for tool_call in tool_calls] ) if tool_calls else {}), conversation=conversation, reasoning=reasoning if reasoning else None ) if provider_info is not None: chat_completion.provider = provider_info.name chat_completion.model = provider_info.model yield chat_completion finally: await safe_aclose(response) class Client(BaseClient): def __init__( self, provider: Optional[ProviderType] = None, media_provider: Optional[ProviderType] = None, **kwargs ) -> None: super().__init__(**kwargs) if self.base_url and provider is None: provider = create_custom_provider(base_url=self.base_url, api_key=self.api_key) self.chat: Chat = Chat(self, provider) if media_provider is None: media_provider = kwargs.get("image_provider", provider) self.models: ClientModels = ClientModels(self, provider, media_provider) self.images: Images = Images(self, media_provider) self.media: Images = self.images class Completions: def __init__(self, client: Client, provider: Optional[ProviderType] = None): self.client: Client = client self.provider: ProviderType = provider def create( self, messages: Messages, model: str = "", provider: Optional[ProviderType] = None, stream: Optional[bool] = False, proxy: Optional[str] = None, image: Optional[ImageType] = None, image_name: Optional[str] = None, response_format: Optional[dict] = None, max_tokens: Optional[int] = None, stop: Optional[Union[list[str], str]] = None, ignore_stream: Optional[bool] = False, raw: Optional[bool] = False, **kwargs ) -> ChatCompletion: if isinstance(messages, str): messages = [{"role": "user", "content": messages}] resolve_media(kwargs, image, image_name) if hasattr(model, "name"): model = model.get_long_name() if provider is None: provider = self.provider if provider is None: provider = AnyProvider if isinstance(provider, str): provider = convert_to_provider(provider) stop = [stop] if isinstance(stop, str) else stop if ignore_stream: kwargs["ignore_stream"] = True response = iter_run_tools( provider, model=model, messages=messages, stream=stream, **filter_none( proxy=self.client.proxy if proxy is None else proxy, max_tokens=max_tokens, stop=stop, api_key=self.client.api_key, base_url=self.client.base_url ), **kwargs ) provider_info = ProviderInfo(**provider.get_dict(), model=model) def fallback(response): return iter_response(response, stream, response_format, max_tokens, stop, provider_info) if raw: def raw_response(response): chunks = [] started = False for chunk in response: if isinstance(chunk, JsonResponse): yield chunk started = True else: chunks.append(chunk) if not started: for chunk in fallback(chunks): yield chunk if stream: return raw_response(response) return next(raw_response(response)) if stream: return fallback(response) return next(fallback(response)) def stream( self, messages: Messages, model: str = "", **kwargs ) -> IterResponse: return self.create(messages, model, stream=True, **kwargs) class Chat: completions: Completions def __init__(self, client: Client, provider: Optional[ProviderType] = None): self.completions = Completions(client, provider) class Images: def __init__(self, client: Client, provider: Optional[ProviderType] = None): self.client: Client = client self.provider: Optional[ProviderType] = provider def generate( self, prompt: str, model: str = None, provider: Optional[ProviderType] = None, response_format: Optional[str] = None, proxy: Optional[str] = None, **kwargs ) -> ImagesResponse: """ Synchronous generate method that runs the async_generate method in an event loop. """ return asyncio.run(self.async_generate(prompt, model, provider, response_format, proxy, **kwargs)) async def get_provider_handler(self, model: Optional[str], provider: Optional[ProviderType], default: ProviderType) -> ProviderType: if provider is None: provider_handler = self.provider if provider_handler is None: provider_handler = self.client.models.get(model, default) else: provider_handler = provider if isinstance(provider_handler, str): provider_handler = convert_to_provider(provider_handler) return provider_handler async def async_generate( self, prompt: str, model: Optional[str] = None, provider: Optional[ProviderType] = None, response_format: Optional[str] = None, proxy: Optional[str] = None, api_key: Optional[str] = None, **kwargs ) -> ImagesResponse: provider_handler = await self.get_provider_handler(model, provider, PollinationsImage) provider_name = provider_handler.__name__ if hasattr(provider_handler, "__name__") else type(provider_handler).__name__ if proxy is None: proxy = self.client.proxy if api_key is None: api_key = self.client.api_key error = None response = None if isinstance(provider_handler, IterListProvider): for provider in provider_handler.providers: try: response = await self._generate_image_response(provider, provider.__name__, model, prompt, proxy=proxy, api_key=api_key, **kwargs) if response is not None: provider_name = provider.__name__ break except Exception as e: error = e debug.error(f"{provider.__name__}:", e) else: response = await self._generate_image_response(provider_handler, provider_name, model, prompt, proxy=proxy, api_key=api_key, **kwargs) if response is None: if error is not None: raise error raise NoMediaResponseError(f"No media response from {provider_name}") return await self._process_image_response( response, model, provider_name, kwargs.get("download_media", True), response_format, proxy ) async def _generate_image_response( self, provider_handler: ProviderType, provider_name: str, model: str, prompt: str, prompt_prefix: str = "Generate a image: ", api_key: str = None, **kwargs ) -> MediaResponse: messages = [{"role": "user", "content": f"{prompt_prefix}{prompt}"}] items: list[MediaResponse] = [] if isinstance(api_key, dict): api_key = api_key.get(provider_handler.get_parent()) if hasattr(provider_handler, "create_async_generator"): async for item in provider_handler.create_async_generator( model, messages, stream=True, prompt=prompt, api_key=api_key, **kwargs ): if isinstance(item, (MediaResponse, AudioResponse)) and not isinstance(item, HiddenResponse): items.append(item) elif hasattr(provider_handler, "create_completion"): for item in provider_handler.create_completion( model, messages, True, prompt=prompt, api_key=api_key, **kwargs ): if isinstance(item, (MediaResponse, AudioResponse)) and not isinstance(item, HiddenResponse): items.append(item) else: raise ValueError(f"Provider {provider_name} does not support image generation") urls = [] for item in items: if isinstance(item, AudioResponse): urls.append(item.to_uri()) elif isinstance(item.urls, str): urls.append(item.urls) elif isinstance(item.urls, list): urls.extend(item.urls) if not urls: return None alt = getattr(items[0], "alt", "") return MediaResponse(urls, alt, items[0].options) def create_variation( self, image: ImageType, image_name: str = None, prompt: str = "Create a variation of this image", model: str = None, provider: Optional[ProviderType] = None, response_format: Optional[str] = None, **kwargs ) -> ImagesResponse: return asyncio.run(self.async_create_variation( image=image, image_name=image_name, prompt=prompt, model=model, provider=provider, response_format=response_format, **kwargs )) async def async_create_variation( self, *, image: ImageType, image_name: str = None, prompt: str = "Create a variation of this image", model: Optional[str] = None, provider: Optional[ProviderType] = None, response_format: Optional[str] = None, proxy: Optional[str] = None, **kwargs ) -> ImagesResponse: provider_handler = await self.get_provider_handler(model, provider, OpenaiAccount) provider_name = provider_handler.__name__ if hasattr(provider_handler, "__name__") else type(provider_handler).__name__ if proxy is None: proxy = self.client.proxy resolve_media(kwargs, image, image_name) error = None response = None if isinstance(provider_handler, IterListProvider): for provider in provider_handler.providers: try: response = await self._generate_image_response(provider, provider.__name__, model, prompt, **kwargs) if response is not None: provider_name = provider.__name__ break except Exception as e: error = e debug.error(f"{provider.__name__}:", e) else: response = await self._generate_image_response(provider_handler, provider_name, model, prompt, **kwargs) if response is None: if error is not None: raise error raise NoMediaResponseError(f"No media response from {provider_name}") return await self._process_image_response( response, model, provider_name, kwargs.get("download_media", True), response_format, proxy ) async def _process_image_response( self, response: MediaResponse, model: str, provider: str, download_media: bool, response_format: Optional[str] = None, proxy: str = None ) -> ImagesResponse: if response_format == "url": # Return original URLs without saving locally images = [Image.model_construct(url=image, revised_prompt=response.alt) for image in response.get_list()] elif response_format == "b64_json": # Convert URLs directly to base64 without saving async def get_b64_from_url(url: str) -> Image: if url.startswith("/media/"): with open(os.path.join(get_media_dir(), os.path.basename(url)), "rb") as f: b64_data = base64.b64encode(f.read()).decode() return Image.model_construct(b64_json=b64_data, revised_prompt=response.alt) async with aiohttp.ClientSession(cookies=response.get("cookies")) as session: async with session.get(url, proxy=proxy) as resp: if resp.status == 200: b64_data = base64.b64encode(await resp.read()).decode() return Image.model_construct(b64_json=b64_data, revised_prompt=response.alt) return Image.model_construct(url=url, revised_prompt=response.alt) images = await asyncio.gather(*[get_b64_from_url(image) for image in response.get_list()]) else: # Save locally for None (default) case images = response.get_list() if download_media or response.get("cookies") or response.get("headers"): images = await copy_media(response.get_list(), response.get("cookies"), response.get("headers"), proxy, response.alt) images = [Image.model_construct(url=image, revised_prompt=response.alt) for image in images] return ImagesResponse.model_construct( created=int(time.time()), data=images, model=model, provider=provider ) class AsyncClient(BaseClient): def __init__( self, provider: Optional[ProviderType] = None, media_provider: Optional[ProviderType] = None, **kwargs ) -> None: super().__init__(**kwargs) self.chat: AsyncChat = AsyncChat(self, provider) if media_provider is None: media_provider = kwargs.get("image_provider", provider) self.models: ClientModels = ClientModels(self, provider, media_provider) self.images: AsyncImages = AsyncImages(self, media_provider) self.media: AsyncImages = self.images class AsyncChat: completions: AsyncCompletions def __init__(self, client: AsyncClient, provider: Optional[ProviderType] = None): self.completions = AsyncCompletions(client, provider) class AsyncCompletions: def __init__(self, client: AsyncClient, provider: Optional[ProviderType] = None): self.client: AsyncClient = client self.provider: ProviderType = provider def create( self, messages: Messages, model: str = "", provider: Optional[ProviderType] = None, stream: Optional[bool] = False, proxy: Optional[str] = None, image: Optional[ImageType] = None, image_name: Optional[str] = None, response_format: Optional[dict] = None, max_tokens: Optional[int] = None, stop: Optional[Union[list[str], str]] = None, ignore_stream: Optional[bool] = False, raw: Optional[bool] = False, **kwargs ) -> Awaitable[ChatCompletion]: if isinstance(messages, str): messages = [{"role": "user", "content": messages}] resolve_media(kwargs, image, image_name) if hasattr(model, "name"): model = model.get_long_name() if provider is None: provider = self.provider if provider is None: provider = AnyProvider if isinstance(provider, str): provider = convert_to_provider(provider) stop = [stop] if isinstance(stop, str) else stop if ignore_stream: kwargs["ignore_stream"] = True response = async_iter_run_tools( provider, model=model, messages=messages, stream=stream, **filter_none( proxy=self.client.proxy if proxy is None else proxy, max_tokens=max_tokens, stop=stop, api_key=self.client.api_key, base_url=self.client.base_url ), **kwargs ) def fallback(response): provider_info = ProviderInfo(**provider.get_dict(), model=model) return async_iter_response(response, stream, response_format, max_tokens, stop, provider_info) if raw: async def raw_response(response): chunks = [] started = False async for chunk in response: if isinstance(chunk, JsonResponse): yield chunk started = True else: chunks.append(chunk) if not started: for chunk in fallback(chunks): yield chunk if stream: return raw_response(response) return anext(raw_response(response)) if stream: return fallback(response) return anext(fallback(response)) def stream( self, messages: Messages, model: str = "", **kwargs ) -> AsyncIterator[ChatCompletionChunk]: return self.create(messages, model, stream=True, **kwargs) class AsyncImages(Images): def __init__(self, client: AsyncClient, provider: Optional[ProviderType] = None): self.client: AsyncClient = client self.provider: Optional[ProviderType] = provider async def generate( self, prompt: str, model: Optional[str] = None, provider: Optional[ProviderType] = None, response_format: Optional[str] = None, **kwargs ) -> ImagesResponse: return await self.async_generate(prompt, model, provider, response_format, **kwargs) async def create_variation( self, image: ImageType, model: str = None, provider: ProviderType = None, response_format: Optional[str] = None, **kwargs ) -> ImagesResponse: return await self.async_create_variation( image=image, model=model, provider=provider, response_format=response_format, **kwargs ) def create_custom_provider( base_url: str, api_key: str = None, name: str = None, working: bool = True, default_model: str = "", models: List[str] = None, **kwargs ) -> Type[OpenaiTemplate]: """ Create a custom provider class based on OpenaiTemplate. Args: base_url: The base URL for the API (e.g., "https://api.example.com/v1") api_key: Optional API key for authentication name: Optional name for the provider (defaults to derived from base_url) working: Whether the provider is working (default: True) default_model: Default model to use models: List of available models **kwargs: Additional attributes to set on the provider class Returns: A custom provider class that extends OpenaiTemplate """ if name is None: # Derive name from base_url from urllib.parse import urlparse parsed = urlparse(base_url) name = parsed.netloc.replace(".", "_").replace("-", "_").title().replace("_", "") if not name: name = "CustomProvider" # Create a new class that extends OpenaiTemplate class_attrs = { "url": base_url, "base_url": base_url.rstrip("/"), "api_key": api_key, "working": working, "default_model": default_model, "models": models or [], **kwargs } CustomProvider = type(name, (OpenaiTemplate,), class_attrs) print(f"Created custom provider class '{name}' with base URL '{base_url}'") return CustomProvider class ClientFactory: """ Factory class for creating Client and AsyncClient instances with various provider configurations. Supports: - Named providers (e.g., "PollinationsAI", "DeepInfra") - Custom providers with custom API base URLs - Live providers (dynamically loaded providers) Example usage: # Create client with a named provider client = ClientFactory.create_client("PollinationsAI") # Create client with custom provider client = ClientFactory.create_client( base_url="https://api.example.com/v1", api_key="your-api-key" ) # Create async client async_client = ClientFactory.create_async_client("PollinationsAI") """ # Registry of live/custom providers _live_providers_url = "https://g4f.dev/dist/js/providers.json" _live_providers: Dict[str, Dict] = {} @classmethod def create_provider( cls, name: str, provider: Union[Type[BaseProvider], str], base_url: str = None, api_key: str = None, **kwargs ) -> Type[BaseProvider]: """ Register a live/custom provider that can be used by name. Args: name: Name to register the provider under provider: Either a provider class or "custom" to create a custom provider base_url: Base URL for custom providers api_key: API key for custom providers **kwargs: Additional arguments for custom provider creation Returns: The registered provider class """ if not isinstance(provider, str): return provider elif provider.startswith("custom:"): if provider.startswith("custom:"): serverId = provider[7:] base_url = f"https://g4f.space/custom/{serverId}" if not base_url: raise ValueError("base_url is required for custom providers") provider = create_custom_provider(base_url, api_key, name=name, **kwargs) elif provider in ProviderUtils.convert: provider = ProviderUtils.convert[provider] else: if not cls._live_providers: path = Path(get_cookies_dir()) / "models" / datetime.today().strftime('%Y-%m-%d') / f"providers.json" path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): with open(path, "r", encoding="utf-8") as f: cls._live_providers = json.load(f) cls._live_providers = requests.get(cls._live_providers_url).json() with open(path, "w", encoding="utf-8") as f: json.dump(cls._live_providers, f, indent=4) if provider in cls._live_providers.get("providers", {}): config = cls._live_providers["providers"][provider] if "provider" in config and config.get("provider") in ProviderUtils.convert: return ProviderUtils.convert[config.get("provider")] return create_custom_provider( base_url=config.get("baseUrl") if api_key else config.get("backupUrl", config.get("baseUrl")), api_key=api_key, name=provider, default_model=cls._live_providers["defaultModels"].get(provider, ""), ) else: raise ProviderNotFoundError(f"Provider '{name}' not found") return provider @classmethod def create_client( cls, provider: Union[str, Type[BaseProvider], None] = None, media_provider: Union[str, Type[BaseProvider], None] = None, base_url: str = None, api_key: str = None, proxies: Union[dict, str] = None, **kwargs ) -> "Client": """ Create a synchronous Client instance. Args: provider: Provider name(s), class(es), or None for default media_provider: Provider for media/image generation base_url: API base URL for custom provider api_key: API key for authentication proxies: Proxy configuration **kwargs: Additional arguments passed to Client Returns: Configured Client instance Example: # Named provider client = ClientFactory.create_client("PollinationsAI") # Custom provider client = ClientFactory.create_client( base_url="https://api.openai.com/v1", api_key="sk-..." ) """ return Client( provider=cls.create_provider(None, provider, base_url, api_key, **kwargs), media_provider=media_provider, api_key=api_key, base_url=base_url, proxies=proxies, **kwargs ) @classmethod def create_async_client( cls, provider: Union[str, Type[BaseProvider], None] = None, media_provider: Union[str, Type[BaseProvider], None] = None, base_url: str = None, api_key: str = None, proxies: Union[dict, str] = None, **kwargs ) -> "AsyncClient": """ Create an asynchronous AsyncClient instance. Args: provider: Provider name(s), class(es), or None for default media_provider: Provider for media/image generation base_url: API base URL for custom provider api_key: API key for authentication proxies: Proxy configuration **kwargs: Additional arguments passed to AsyncClient Returns: Configured AsyncClient instance Example: # Named provider client = ClientFactory.create_async_client("PollinationsAI") # Custom provider client = ClientFactory.create_async_client( base_url="https://api.openai.com/v1", api_key="sk-..." ) """ return AsyncClient( provider=cls.create_provider(None, provider, base_url, api_key, **kwargs), media_provider=media_provider, api_key=api_key, base_url=base_url, proxies=proxies, **kwargs )