Generation Strategies

This document describes the Strategy Pattern implementation for generation processing in the Backend.

---

Overview

The Backend uses a Strategy Pattern to decouple generation types from their execution logic. Each strategy encapsulates:

1. How to create Generation and Prediction records
2. How to build and submit tasks to the Compute Server
3. Any additional record creation (e.g., Refine records)

flowchart TB
    I[IGenerationStrategy]

    I --> S1[Base]
    I --> S2[Editing]
    I --> S3[Refine]
    I --> S4[Video]
    I --> S5[BackgroundFix]
    I --> S6[ImageAdjust]
    I --> S7[FaceFix]

Strategy	Full Class Name
Base	BaseGenerationStrategy
Editing	EditingGenerationStrategy
Refine	RefineGenerationStrategy
Video	VideoGenerationStrategy
BackgroundFix	BackgroundFixingGenerationStrategy
ImageAdjust	ImageAdjustingGenerationStrategy
FaceFix	FaceFixingGenerationStrategy

---

Strategy Interface

All strategies implement the IGenerationStrategy protocol:

class IGenerationStrategy(Protocol):
    @property
    def strategy(self) -> GenerationStrategy:
        """Return which strategy this implements."""
        ...

    def create(
        self,
        session: Session,
        generation: Generation,
        owner_id: UUID,
    ) -> Generation:
        """Create Generation and Prediction records."""
        ...

    async def generate(
        self,
        session: Session,
        generation: Generation,
        prediction_ids: list[UUID],
        provider_config: dict | None,
    ) -> list[str]:
        """Submit tasks to Compute Server, return task IDs."""
        ...

---

Workflow Service

The GenerationWorkflowService orchestrates strategy selection and execution:

sequenceDiagram
    participant Route as API Route
    participant WFS as WorkflowService
    participant Workflow as Workflow
    participant Strategy as Strategy
    participant Compute as Compute Server

    Route->>WFS: create(generation)
    WFS->>WFS: Select workflow by generation_type
    WFS->>Workflow: create(generation)
    Workflow->>Strategy: create(generation)
    Strategy-->>Workflow: Generation with Predictions
    Workflow-->>WFS: Generation
    WFS-->>Route: Generation

    Note over Route,Compute: Background task

    Route->>WFS: start(generation)
    WFS->>Workflow: start(generation, prediction_ids)
    Workflow->>Strategy: generate(prediction_ids)
    Strategy->>Compute: Submit tasks
    Compute-->>Strategy: task_ids
    Strategy-->>Workflow: task_ids
    Workflow-->>WFS: task_ids

---

Workflow Types

Base Workflow

Handles GenerationType.BASE — direct task submission.

class BaseGenerationWorkflow:
    generation_type = GenerationType.BASE

    def create(self, session, generation, owner_id):
        # Delegate to strategy
        strategy = self._get_strategy(generation.generation_strategy)
        return strategy.create(session, generation, owner_id)

    async def start(self, generation, prediction_ids, session, provider_config):
        strategy = self._get_strategy(generation.generation_strategy)
        task_ids = await strategy.generate(
            session, generation, prediction_ids, provider_config
        )
        return task_ids

Agentic Workflow

Handles GenerationType.AGENTIC — adds orchestration layer.

class AgenticGenerationWorkflow(BaseGenerationWorkflow):
    generation_type = GenerationType.AGENTIC

    def create(self, session, generation, owner_id):
        # Create base generation
        generation = super().create(session, generation, owner_id)

        # Create OrchestratedShot for each prediction
        for prediction in generation.predictions:
            orchestrated_shot = OrchestratedShot(
                id=prediction.id,
                generation_id=generation.id,
                status=OrchestratedShotStatus.PENDING,
                orchestration_metadata={
                    "max_attempts": generation.max_attempts,
                    "min_confidence_score": generation.min_confidence_score,
                    "evaluation_provider": generation.evaluation_provider,
                }
            )
            session.add(orchestrated_shot)

        return generation

    async def start(self, generation, prediction_ids, session, provider_config):
        task_ids = []
        for prediction_id in prediction_ids:
            # Use orchestrator instead of direct submission
            task_id = await shot_orchestrator.start_shot_orchestration(
                prediction_id=prediction_id,
                generation=generation,
                provider_config=provider_config,
            )
            task_ids.append(task_id)
        return task_ids

---

Strategy Implementations

BaseGenerationStrategy

Creates standard image generations.

flowchart TB
    A[Receive Generation Request] --> B[Load Subject/Product/Style]
    B --> C[Generate or Use Prompt]
    C --> D[Create Generation Record]
    D --> E[Create N Predictions]
    E --> F[For Each Prediction]
    F --> G[Build CreateGenerationTask]
    G --> H[Add LoRA Weights]
    H --> I[Submit to Compute]
    I --> J[Return Task IDs]

Task building:

def _build_task(self, generation, prediction) -> CreateGenerationTask:
    return CreateGenerationTask(
        prompt=generation.generated_prompt,
        width=generation.width,
        height=generation.height,
        seed=prediction.seed,
        subject_lora_weight=generation.subject.lora_weight,
        product_lora_weight=generation.product.lora_weight,
        style_lora_weight=generation.style.lora_weight if generation.style else None,
        reference_image_url=generation.reference_image_url,
        reference_image_strength=generation.reference_strength,
        provider_type=provider_config.get("type"),
        provider_config=provider_config,
    )

EditingGenerationStrategy

Handles garment placement and compositing.

flowchart TB
    A[Receive Edit Request] --> B[Load Base Image]
    B --> C[Extract Dimensions]
    C --> D[Process Reference Images]
    D --> E[Create Generation Record]
    E --> F[Create Predictions]
    F --> G[Build CreateEditingTask]
    G --> H{Has Reference Images?}
    H -->|Yes| I[Add Detail Enhancer Step]
    H -->|No| J[Submit to Compute]
    I --> J

Key differences: - Uses base_image_url instead of generating from scratch - Supports reference images for garment placement - May chain with DETAIL_ENHANCER task

RefineGenerationStrategy

Handles upscaling and enhancement.

flowchart TB
    A[Receive Refine Request] --> B[Load Base Image]
    B --> C[Create Generation Record]
    C --> D[Create Predictions]
    D --> E[Create Refine Records]
    E --> F[Build CreateRefineTask]
    F --> G[Configure Upscale Settings]
    G --> H[Configure Face Detail Settings]
    H --> I[Submit to Compute]

Additional records: - Creates Refine record per prediction - Stores all refinement parameters for reference

VideoGenerationStrategy

Handles video generation from images.

flowchart TB
    A[Receive Video Request] --> B[Load Source Image]
    B --> C[Create Generation Record]
    C --> D[Create Predictions]
    D --> E[Build CreateVideoTask]
    E --> F[Configure Motion Parameters]
    F --> G[Submit to Compute]

---

Strategy Selection

Strategy selection happens based on generation_strategy field:

STRATEGY_MAP = {
    GenerationStrategy.GENERATE: BaseGenerationStrategy,
    GenerationStrategy.SUBJECT_GENERATE: SubjectBaseGenerationStrategy,
    GenerationStrategy.EDITING: EditingGenerationStrategy,
    GenerationStrategy.REFINE: RefineGenerationStrategy,
    GenerationStrategy.VIDEO: VideoGenerationStrategy,
    GenerationStrategy.BACKGROUND_FIXING: BackgroundFixingGenerationStrategy,
    GenerationStrategy.IMAGE_ADJUSTING: ImageAdjustingGenerationStrategy,
    GenerationStrategy.FACE_FIXING: FaceFixingGenerationStrategy,
}

def _get_strategy(self, strategy: GenerationStrategy) -> IGenerationStrategy:
    return STRATEGY_MAP[strategy]()

---

Adding New Strategies

To add a new generation type:

1. Define strategy enum value:

   class GenerationStrategy(str, Enum):
       MY_NEW_STRATEGY = "MY_NEW_STRATEGY"
   

2. Create task type (if needed):

   class TaskType(str, Enum):
       MY_NEW_TASK = "MY_NEW_TASK"
   

3. Implement strategy class:

   class MyNewStrategy(IGenerationStrategy):
       @property
       def strategy(self) -> GenerationStrategy:
           return GenerationStrategy.MY_NEW_STRATEGY
   
       def create(self, session, generation, owner_id) -> Generation:
           # Create records
           ...
   
       async def generate(self, session, generation, prediction_ids, config) -> list[str]:
           # Build and submit tasks
           ...
   

4. Register in strategy map:

   STRATEGY_MAP[GenerationStrategy.MY_NEW_STRATEGY] = MyNewStrategy
   

5. Add API route (if needed):

   @router.post("/my-new-type")
   async def create_my_new_generation(...):
       ...
   

---

Related Documentation

• Generation Pipeline — Overview
• Generation Types — Type details
• Compute Server Tasks — Task parameters