XFE Git
XFE Studio Git
Git 首页 全局搜索
XFE 主站 文档 NuGet
公开
关注 0 Fork 0 Star 0
UTF-8
from __future__ import annotations

import random

from ...typing import AsyncResult, Messages, MediaListType
from ...errors import ResponseError
from ..base_provider import AsyncGeneratorProvider, ProviderModelMixin

from .BAAI_Ling                       import BAAI_Ling
from .BlackForestLabs_Flux1Dev       import BlackForestLabs_Flux1Dev
from .BlackForestLabs_Flux1KontextDev import BlackForestLabs_Flux1KontextDev
from .CohereForAI_C4AI_Command       import CohereForAI_C4AI_Command
from .DeepseekAI_JanusPro7b          import DeepseekAI_JanusPro7b
from .Microsoft_Phi_4_Multimodal     import Microsoft_Phi_4_Multimodal
from .Qwen_Qwen_2_5                  import Qwen_Qwen_2_5
from .Qwen_Qwen_2_5M                 import Qwen_Qwen_2_5M
from .Qwen_Qwen_2_5_Max              import Qwen_Qwen_2_5_Max
from .Qwen_Qwen_2_72B                import Qwen_Qwen_2_72B
from .Qwen_Qwen_3                    import Qwen_Qwen_3
from .StabilityAI_SD35Large          import StabilityAI_SD35Large

class HuggingSpace(AsyncGeneratorProvider, ProviderModelMixin):
    url = "https://huggingface.co/spaces"
    working = True
    active_by_default = True

    default_model = Qwen_Qwen_2_72B.default_model
    default_image_model = BlackForestLabs_Flux1Dev.default_model
    default_vision_model = Microsoft_Phi_4_Multimodal.default_model
    providers = [
        BAAI_Ling,
        BlackForestLabs_Flux1Dev,
        BlackForestLabs_Flux1KontextDev,
        CohereForAI_C4AI_Command,
        DeepseekAI_JanusPro7b,
        Microsoft_Phi_4_Multimodal,
        Qwen_Qwen_2_5,
        Qwen_Qwen_2_5M,
        Qwen_Qwen_2_5_Max,
        Qwen_Qwen_2_72B,
        Qwen_Qwen_3,
        StabilityAI_SD35Large,
    ]

    @classmethod
    def get_parameters(cls, **kwargs) -> dict:
        parameters = {}
        for provider in cls.providers:
            parameters = {**parameters, **provider.get_parameters(**kwargs)}
        return parameters

    @classmethod
    def get_models(cls, **kwargs) -> list[str]:
        if not cls.models:
            models = []
            image_models = []
            vision_models = []
            cls.model_aliases = {}
            for provider in cls.providers:
                models.extend(provider.get_models(**kwargs))
                models.extend(provider.model_aliases.keys())
                image_models.extend(provider.image_models)
                vision_models.extend(provider.vision_models)
                cls.model_aliases.update(provider.model_aliases)
            models = list(set(models))
            models.sort()
            cls.models = models
            cls.image_models = list(set(image_models))
            cls.vision_models = list(set(vision_models))
        return cls.models

    @classmethod
    async def create_async_generator(
        cls, model: str, messages: Messages, media: MediaListType = None, **kwargs
    ) -> AsyncResult:
        if not model and media is not None:
            model = cls.default_vision_model
        is_started = False
        random.shuffle(cls.providers)
        for provider in cls.providers:
            if model in provider.model_aliases or model in provider.get_models():
                alias = provider.model_aliases[model] if model in provider.model_aliases else model
                async for chunk in provider.create_async_generator(alias, messages, media=media, **kwargs):
                    is_started = True
                    yield chunk
            if is_started:
                return

for provider in HuggingSpace.providers:
    provider.parent = HuggingSpace.__name__
    provider.hf_space = True