Pipeline Interface

def pipeline(
    task: str = None,
    model: Union[str, List[str], Model, List[Model]] = None,
    preprocessor = None,
    config_file: str = None,
    pipeline_name: str = None,
    framework: str = None,
    device: str = None,
    model_revision: Optional[str] = DEFAULT_MODEL_REVISION,
    ignore_file_pattern: List[str] = None,
    **kwargs
) -> Pipeline:
    """
    Create a pipeline for a specific AI task.
    
    Parameters:
    - task: Task identifier (e.g., 'text-classification', 'image-classification')
    - model: Model identifier(s), local path(s), or Model instance(s) - supports single values or lists
    - preprocessor: Custom preprocessor instance
    - config_file: Path to configuration file
    - pipeline_name: Name for the pipeline
    - framework: ML framework to use ('pytorch', 'tensorflow', etc.)
    - device: Device to run inference on ('cpu', 'cuda', 'gpu')
    - model_revision: Specific model version/revision (default: DEFAULT_MODEL_REVISION)
    - ignore_file_pattern: List of file patterns to ignore during model loading
    - **kwargs: Additional task-specific parameters
    
    Returns:
    Task-specific pipeline instance
    """

class Pipeline:
    """
    Base pipeline class providing common interface for all AI tasks.
    """
    
    def __init__(
        self,
        model,
        preprocessor = None,
        config_file: str = None,
        device: str = None,
        auto_collate: bool = True,
        **kwargs
    ):
        """
        Initialize pipeline with model and preprocessing components.
        
        Parameters:
        - model: Model instance or identifier
        - preprocessor: Preprocessor instance
        - config_file: Path to configuration file
        - device: Target device for inference
        - auto_collate: Whether to automatically collate batch inputs
        """
    
    def __call__(self, inputs, **kwargs):
        """
        Process inputs through the complete pipeline.
        
        Parameters:
        - inputs: Input data (format varies by task)
        - **kwargs: Task-specific parameters
        
        Returns:
        Processed outputs in task-specific format
        """
    
    def preprocess(self, inputs):
        """
        Preprocess raw inputs for model consumption.
        
        Parameters:
        - inputs: Raw input data
        
        Returns:
        Preprocessed data ready for model inference
        """
    
    def forward(self, inputs):
        """
        Run model inference on preprocessed inputs.
        
        Parameters:
        - inputs: Preprocessed input data
        
        Returns:
        Raw model outputs
        """
    
    def postprocess(self, inputs):
        """
        Process model outputs into final user-friendly format.
        
        Parameters:
        - inputs: Raw model outputs
        
        Returns:
        Final processed outputs
        """

# Text Classification
from modelscope import pipeline

classifier = pipeline('text-classification', 
                     model='damo/nlp_structbert_sentence-similarity_chinese')
result = classifier('这是一个积极的评论')

# Text Generation  
generator = pipeline('text-generation',
                    model='damo/nlp_gpt3_text-generation_chinese')
result = generator('人工智能的未来发展', max_length=100)

# Named Entity Recognition
ner = pipeline('named-entity-recognition',
              model='damo/nlp_raner_named-entity-recognition_chinese')
result = ner('张三在北京工作')

# Text Summarization
summarizer = pipeline('summarization',
                     model='damo/nlp_pegasus_summarization_chinese')
result = summarizer('长篇文本内容...')

# Image Classification
classifier = pipeline('image-classification',
                     model='damo/cv_resnet50_image-classification_imagenet')
result = classifier('path/to/image.jpg')

# Object Detection
detector = pipeline('object-detection',
                   model='damo/cv_yolox-s_detection_coco')
result = detector('path/to/image.jpg')

# Image Segmentation
segmenter = pipeline('image-segmentation',
                    model='damo/cv_resnet101_image-multiple-human-parsing')
result = segmenter('path/to/image.jpg')

# Face Recognition
face_recognition = pipeline('face-recognition',
                           model='damo/cv_ir50_face-recognition_faceid')
result = face_recognition('path/to/face.jpg')

# Speech Recognition
asr = pipeline('automatic-speech-recognition',
              model='damo/speech_paraformer-large_asr_nat-zh-cn')
result = asr('path/to/audio.wav')

# Text-to-Speech
tts = pipeline('text-to-speech',
              model='damo/speech_sambert-hifigan_tts_zh-cn_16k')
result = tts('你好，欢迎使用ModelScope')

# Audio Classification
audio_classifier = pipeline('audio-classification',
                           model='damo/speech_xvector_sv-zh-cn-cnceleb-16k')
result = audio_classifier('path/to/audio.wav')

# Image Captioning
captioner = pipeline('image-captioning',
                    model='damo/multi-modal_clip-vit-base-patch16_zh')
result = captioner('path/to/image.jpg')

# Visual Question Answering
vqa = pipeline('visual-question-answering',
              model='damo/multi-modal_clip-vit-base-patch16_zh')
result = vqa(image='path/to/image.jpg', question='图片中有什么？')

# Text-to-Image Generation
text2img = pipeline('text-to-image-synthesis',
                   model='damo/multi-modal_latent-diffusion_text-to-image-synthesis')
result = text2img('一只可爱的小猫')

from modelscope import pipeline

# CPU inference
pipe_cpu = pipeline('text-classification', 
                   model='model_name',
                   device='cpu')

# GPU inference
pipe_gpu = pipeline('text-classification',
                   model='model_name', 
                   device='cuda')

# Automatic device selection
pipe_auto = pipeline('text-classification',
                    model='model_name')  # Automatically selects best available device

from modelscope import pipeline, Preprocessor

# Custom preprocessor
class MyPreprocessor(Preprocessor):
    def __call__(self, data):
        # Custom preprocessing logic
        return processed_data

# Use custom preprocessor
pipe = pipeline('text-classification',
               model='model_name',
               preprocessor=MyPreprocessor())

# Single input
result = pipe('single input text')

# Batch inputs
results = pipe(['input 1', 'input 2', 'input 3'])

# Large batch processing with automatic collation
pipe = pipeline('text-classification', 
               model='model_name',
               auto_collate=True)
results = pipe(large_input_list)

generator = pipeline('text-generation', model='model_name')

result = generator(
    'Input text',
    max_length=100,          # Maximum output length
    min_length=10,           # Minimum output length
    temperature=0.7,         # Sampling temperature
    top_k=50,               # Top-k sampling
    top_p=0.9,              # Top-p (nucleus) sampling
    repetition_penalty=1.2,  # Repetition penalty
    do_sample=True          # Enable sampling
)

detector = pipeline('object-detection', model='model_name')

result = detector(
    'path/to/image.jpg',
    score_thresh=0.5,        # Confidence threshold
    nms_thresh=0.3,          # Non-maximum suppression threshold
    max_det=100             # Maximum detections
)

asr = pipeline('automatic-speech-recognition', model='model_name')

result = asr(
    'path/to/audio.wav',
    sample_rate=16000,       # Audio sample rate
    chunk_size=6000,         # Chunk size for long audio
    merge_chunks=True        # Merge chunk results
)

from modelscope import pipeline

try:
    # Create pipeline
    pipe = pipeline('text-classification', model='invalid_model')
except Exception as e:
    print(f"Pipeline creation failed: {e}")

try:
    # Process inputs
    result = pipe('input text')
except Exception as e:
    print(f"Processing failed: {e}")

# Create multiple pipelines for complex workflows
preprocessor_pipe = pipeline('text-preprocessing', model='model1')
classifier_pipe = pipeline('text-classification', model='model2')

# Chain pipelines
preprocessed = preprocessor_pipe(raw_text)
final_result = classifier_pipe(preprocessed)

from modelscope import Pipeline, Model, Preprocessor

class CustomTextPipeline(Pipeline):
    def __init__(self, model, preprocessor=None, **kwargs):
        super().__init__(model, preprocessor, **kwargs)
    
    def preprocess(self, inputs):
        # Custom preprocessing logic
        return processed_inputs
    
    def postprocess(self, inputs):
        # Custom postprocessing logic
        return final_outputs

# Register and use custom pipeline
pipeline = CustomTextPipeline(model=my_model, preprocessor=my_preprocessor)
result = pipeline('input text')