Function _discover_pipelines

backend/python/diffusers/diffusers_dynamic_loader.py:217–264

Discover all subclasses of DiffusionPipeline from diffusers. This function uses the generic discover_diffusers_classes() internally and adds pipeline-specific task alias generation. It also includes AutoPipeline classes which are special utility classes for automatic pipeline s

()

def _discover_pipelines() -> Tuple[Dict[str, Type], Dict[str, List[str]]]:
    """
    Discover all subclasses of DiffusionPipeline from diffusers.

    This function uses the generic discover_diffusers_classes() internally
    and adds pipeline-specific task alias generation. It also includes
    AutoPipeline classes which are special utility classes for automatic
    pipeline selection.

    Returns:
        A tuple of (pipeline_registry, task_aliases) where:
        - pipeline_registry: Dict mapping class names to class objects
        - task_aliases: Dict mapping task aliases to lists of class names
    """
    # Use the generic discovery function
    pipeline_registry = discover_diffusers_classes("DiffusionPipeline", include_base=True)

    # Also add AutoPipeline classes - these are special utility classes that are
    # NOT subclasses of DiffusionPipeline but are commonly used
    import diffusers
    auto_pipeline_classes = [
        "AutoPipelineForText2Image",
        "AutoPipelineForImage2Image",
        "AutoPipelineForInpainting",
    ]
    for cls_name in auto_pipeline_classes:
        try:
            cls = getattr(diffusers, cls_name)
            if cls is not None:
                pipeline_registry[cls_name] = cls
        except AttributeError:
            # Class not available in this version of diffusers
            pass

    # Generate task aliases for pipelines
    task_aliases: Dict[str, List[str]] = {}
    for attr_name in pipeline_registry:
        if attr_name == "DiffusionPipeline":
            continue  # Skip base class for alias generation

        aliases = _extract_task_keywords(attr_name)
        for alias in aliases:
            if alias not in task_aliases:
                task_aliases[alias] = []
            if attr_name not in task_aliases[alias]:
                task_aliases[alias].append(attr_name)

    return pipeline_registry, task_aliases


def get_pipeline_registry() -> Dict[str, Type]: