Auto-generate meaningful workflow titles via debounced LLM (#SKY-7287) (#4652)

skyvern/forge/prompts/skyvern/generate-workflow-title.j2

@@ -0,0 +1,15 @@
+Generate a brief, descriptive title for a browser automation workflow.
+
+Rules:
+- Maximum 5 words
+- Start with an action verb when possible
+- Be specific about what the workflow does
+- Examples: "Scrape LinkedIn job listings", "Extract Amazon product prices", "Submit insurance form"
+
+Workflow blocks ({{ blocks|length }} total):
+{% for block in blocks %}
+- {{ block.block_type }}{% if block.url %} on {{ block.url }}{% endif %}{% if block.goal %}: {{ block.goal }}{% endif %}
+{% endfor %}
+
+Respond with JSON only:
+{"title": "generated title here"}

skyvern/forge/sdk/routes/prompts.py

@@ -10,8 +10,14 @@ from skyvern.forge.prompts import prompt_engine
 from skyvern.forge.sdk.api.llm.exceptions import LLMProviderError
 from skyvern.forge.sdk.routes.routers import base_router
 from skyvern.forge.sdk.schemas.organizations import Organization
-from skyvern.forge.sdk.schemas.prompts import ImprovePromptRequest, ImprovePromptResponse
+from skyvern.forge.sdk.schemas.prompts import (
+    GenerateWorkflowTitleRequest,
+    GenerateWorkflowTitleResponse,
+    ImprovePromptRequest,
+    ImprovePromptResponse,
+)
 from skyvern.forge.sdk.services import org_auth_service
+from skyvern.forge.sdk.workflow.service import generate_title_from_blocks_info
 
 LOG = structlog.get_logger()
 
@@ -124,3 +130,42 @@ async def improve_prompt(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail=f"Failed to improve prompt: {str(e)}",
         )
+
+
+@base_router.post(
+    "/prompts/generate-workflow-title",
+    tags=["Prompts"],
+    description="Generate a meaningful workflow title from block content",
+    summary="Generate workflow title",
+    include_in_schema=False,
+)
+async def generate_workflow_title(
+    request: GenerateWorkflowTitleRequest,
+    current_org: Organization = Depends(org_auth_service.get_current_org),
+) -> GenerateWorkflowTitleResponse:
+    """Generate a meaningful workflow title based on block content using LLM."""
+    LOG.info(
+        "Generating workflow title",
+        organization_id=current_org.organization_id,
+        num_blocks=len(request.blocks),
+    )
+
+    try:
+        blocks_info = [block.model_dump(exclude_none=True) for block in request.blocks]
+        title = await generate_title_from_blocks_info(
+            organization_id=current_org.organization_id,
+            blocks_info=blocks_info,
+        )
+        return GenerateWorkflowTitleResponse(title=title)
+    except LLMProviderError:
+        LOG.error("Failed to generate workflow title", exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Failed to generate title. Please try again later.",
+        )
+    except Exception as e:
+        LOG.error("Unexpected error generating workflow title", error=str(e), exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Failed to generate title: {str(e)}",
+        )

skyvern/forge/sdk/schemas/prompts.py

@@ -30,3 +30,20 @@ class ImprovePromptResponse(BaseModel):
     error: str | None = Field(None, description="Error message if prompt improvement failed")
     improved: str = Field(..., description="The improved version of the prompt")
     original: str = Field(..., description="The original prompt provided for improvement")
+
+
+class BlockInfoForTitle(BaseModel):
+    block_type: str = Field(..., description="The type of the workflow block")
+    url: str | None = Field(None, description="URL associated with the block")
+    goal: str | None = Field(None, description="Goal or prompt text for the block")
+
+
+class GenerateWorkflowTitleRequest(BaseModel):
+    blocks: list[BlockInfoForTitle] = Field(
+        ...,
+        description="List of block info objects for title generation",
+    )
+
+
+class GenerateWorkflowTitleResponse(BaseModel):
+    title: str | None = Field(None, description="The generated workflow title")

skyvern/forge/sdk/workflow/service.py

@@ -134,6 +134,76 @@ BLOCK_TYPES_THAT_SHOULD_BE_CACHED = {
 }
 
 
+def _extract_blocks_info(blocks: list[BLOCK_YAML_TYPES]) -> list[dict[str, str]]:
+    """Extract lightweight info from blocks for title generation (limit to first 5)."""
+    blocks_info: list[dict[str, str]] = []
+    for block in blocks[:5]:
+        info: dict[str, str] = {"block_type": block.block_type.value}
+
+        # Extract URL if present
+        if hasattr(block, "url") and block.url:
+            info["url"] = block.url
+
+        # Extract goal/prompt
+        goal = None
+        if hasattr(block, "navigation_goal") and block.navigation_goal:
+            goal = block.navigation_goal
+        elif hasattr(block, "data_extraction_goal") and block.data_extraction_goal:
+            goal = block.data_extraction_goal
+        elif hasattr(block, "prompt") and block.prompt:
+            goal = block.prompt
+
+        if goal:
+            # Truncate long goals
+            info["goal"] = goal[:150] if len(goal) > 150 else goal
+
+        blocks_info.append(info)
+    return blocks_info
+
+
+async def generate_title_from_blocks_info(
+    organization_id: str,
+    blocks_info: list[dict[str, Any]],
+) -> str | None:
+    """Call LLM to generate a workflow title from pre-extracted block info."""
+    if not blocks_info:
+        return None
+
+    try:
+        llm_prompt = prompt_engine.load_prompt(
+            "generate-workflow-title",
+            blocks=blocks_info,
+        )
+
+        response = await app.SECONDARY_LLM_API_HANDLER(
+            prompt=llm_prompt,
+            prompt_name="generate-workflow-title",
+            organization_id=organization_id,
+        )
+
+        if isinstance(response, dict) and "title" in response:
+            title = str(response["title"]).strip()
+            if title and len(title) <= 100:  # Sanity check on length
+                return title
+
+        return None
+    except Exception:
+        LOG.exception("Failed to generate workflow title")
+        return None
+
+
+async def generate_workflow_title(
+    organization_id: str,
+    blocks: list[BLOCK_YAML_TYPES],
+) -> str | None:
+    """Generate a meaningful workflow title based on block content using LLM."""
+    if not blocks:
+        return None
+
+    blocks_info = _extract_blocks_info(blocks)
+    return await generate_title_from_blocks_info(organization_id, blocks_info)
+
+
 @dataclass
 class CacheInvalidationPlan:
     reason: CacheInvalidationReason | None = None
@@ -3210,10 +3280,26 @@ class WorkflowService:
         delete_code_cache_is_ok: bool = True,
     ) -> Workflow:
         organization_id = organization.organization_id
+
+        # Generate meaningful title if using default and has blocks
+        title = request.title
+        if title == DEFAULT_WORKFLOW_TITLE and request.workflow_definition.blocks:
+            generated_title = await generate_workflow_title(
+                organization_id=organization_id,
+                blocks=request.workflow_definition.blocks,
+            )
+            if generated_title:
+                title = generated_title
+                LOG.info(
+                    "Generated workflow title",
+                    organization_id=organization_id,
+                    generated_title=title,
+                )
+
         LOG.info(
             "Creating workflow from request",
             organization_id=organization_id,
-            title=request.title,
+            title=title,
         )
         new_workflow_id: str | None = None
 
@@ -3233,7 +3319,7 @@ class WorkflowService:
 
                 # NOTE: it's only potential, as it may be immediately deleted!
                 potential_workflow = await self.create_workflow(
-                    title=request.title,
+                    title=title,
                     workflow_definition=WorkflowDefinition(parameters=[], blocks=[]),
                     description=request.description,
                     organization_id=organization_id,
@@ -3259,7 +3345,7 @@ class WorkflowService:
             else:
                 # NOTE: it's only potential, as it may be immediately deleted!
                 potential_workflow = await self.create_workflow(
-                    title=request.title,
+                    title=title,
                     workflow_definition=WorkflowDefinition(parameters=[], blocks=[]),
                     description=request.description,
                     organization_id=organization_id,
@@ -3291,7 +3377,7 @@ class WorkflowService:
             updated_workflow = await self.update_workflow_definition(
                 workflow_id=potential_workflow.workflow_id,
                 organization_id=organization_id,
-                title=request.title,
+                title=title,
                 description=request.description,
                 workflow_definition=workflow_definition,
             )
@@ -3313,7 +3399,7 @@ class WorkflowService:
                 )
                 await self.delete_workflow_by_id(workflow_id=new_workflow_id, organization_id=organization_id)
             else:
-                LOG.exception(f"Failed to create workflow from request, title: {request.title}")
+                LOG.exception(f"Failed to create workflow from request, title: {title}")
             raise e
 
     @staticmethod