refactor(api): move Workflow generation and collaboration readers

api/controllers/console/app/generator.py

@@ -31,6 +31,7 @@
 from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
 from core.workflow.generator.types import WorkflowGenerateErrorCode
+from extensions.ext_application_services import application_services
 from fields.base import ResponseModel
 from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.errors.invoke import InvokeError
@@ -533,7 +534,7 @@ class WorkflowGenerateApi(Resource):
     """Generate a Workflow / Chatflow draft graph from a natural-language description.
 
     Triggered by the cmd+k `/create` slash command. Returns a graph payload
-    shaped exactly like ``WorkflowService.sync_draft_workflow``'s input, so the
+    shaped exactly like ``WorkflowDraftService.sync``'s input, so the
     frontend can hand it straight to ``/apps/{id}/workflows/draft``.
     """
 
@@ -614,7 +615,7 @@ class WorkflowInstructionSuggestionsApi(Resource):
     @with_current_tenant_id
     @model_validate(WorkflowInstructionSuggestionsPayload)
     def post(self, req_data: WorkflowInstructionSuggestionsPayload, current_tenant_id: str):
-        suggestions = LLMGenerator.generate_workflow_instruction_suggestions(
+        suggestions = application_services().workflow_suggestions.generate_workflow_instruction_suggestions(
             tenant_id=current_tenant_id,
             mode=req_data.mode,
             language=req_data.language,

api/controllers/console/socketio/workflow.py

@@ -9,7 +9,7 @@
 from extensions.ext_socketio import sio
 from libs.passport import PassportService
 from libs.token import extract_access_token
-from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
+from repositories.workflow.collaboration_repository import WorkflowCollaborationRepository
 from services.workflow_collaboration_service import WorkflowCollaborationService
 
 repository = WorkflowCollaborationRepository()

api/core/llm_generator/llm_generator.py

@@ -2,7 +2,7 @@
 import logging
 import re
 from collections.abc import Sequence
-from typing import Any, Literal, Protocol, TypedDict, cast
+from typing import Any, Protocol, TypedDict, cast
 
 import json_repair
 from sqlalchemy import select
@@ -30,7 +30,6 @@
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.telemetry import PromptGenerationEvent, TelemetryContext
 from core.telemetry import emit as telemetry_emit
-from extensions.ext_database import db
 from extensions.ext_storage import storage
 from graphon.enums import WorkflowNodeExecutionMetadataKey
 from graphon.model_runtime.entities.llm_entities import LLMResult
@@ -64,53 +63,6 @@ def _normalize_completion_params(completion_params: dict[str, object]) -> tuple[
     return normalized_parameters, stop
 
 
-# ── Workflow instruction-suggestion tuning ────────────────────────────────
-# Suggestions are a soft, pre-model-pick enhancement: short, buildable example
-# instructions proposed from the tenant's DEFAULT model. Every failure path
-# degrades to an empty list, never an error.
-_SUGGESTION_MIN_COUNT = 1
-_SUGGESTION_MAX_COUNT = 6
-_SUGGESTION_MAX_TOKENS = 512
-_SUGGESTION_TEMPERATURE = 0.8
-# Bound the grounding context so the prompt stays small regardless of how many
-# knowledge bases / tools the tenant has installed.
-_SUGGESTION_KB_LIMIT = 10
-_SUGGESTION_TOOL_SAMPLE_LINES = 20
-
-_SUGGESTION_SYSTEM_PROMPT = (
-    "You help a user start building a Dify app by proposing example build instructions. "
-    "Each suggestion must be a SHORT (at most 8 words), concrete, and BUILDABLE instruction "
-    "describing an app to generate for the given app type. Make the suggestions diverse — cover "
-    "different use cases. When the listed knowledge bases or installed tools fit a suggestion, "
-    "prefer them, but NEVER invent tools or knowledge bases that are not listed. "
-    "Reply with ONLY a JSON array of strings and nothing else."
-)
-
-
-def _parse_string_list(text: str) -> list[str]:
-    """Extract a JSON array of strings from a (possibly noisy) LLM response.
-
-    Slices the first ``[...]`` span so surrounding prose / markdown fences are
-    tolerated, parses it with ``json`` and falls back to ``json_repair``, then
-    keeps only ``str`` items. Returns ``[]`` on any failure so callers can
-    treat parsing as best-effort.
-    """
-    match = re.search(r"\[.*\]", text.strip(), re.DOTALL)
-    if not match:
-        return []
-    raw = match.group(0)
-    try:
-        parsed = json.loads(raw)
-    except Exception:
-        try:
-            parsed = json_repair.loads(raw)
-        except Exception:
-            return []
-    if not isinstance(parsed, list):
-        return []
-    return [item for item in parsed if isinstance(item, str)]
-
-
 class WorkflowServiceInterface(Protocol):
     def get_draft_workflow(self, app_model: App, *, session: Session) -> Workflow | None:
         pass
@@ -250,122 +202,6 @@ def generate_conversation_name(
 
         return name
 
-    @classmethod
-    @with_credit_usage_created_by(CreditUsageCreatedBy.WORKFLOW_INSTRUCTION_SUGGESTIONS)
-    def generate_workflow_instruction_suggestions(
-        cls,
-        tenant_id: str,
-        *,
-        mode: Literal["workflow", "advanced-chat"],
-        language: str | None = None,
-        count: int = 4,
-    ) -> list[str]:
-        """Propose short, buildable example instructions for the workflow generator.
-
-        Runs BEFORE the user picks a model, so it uses the tenant's DEFAULT LLM
-        only. Suggestions are a soft enhancement, never a blocker: every failure
-        path (no default model, KB / tool lookup error, LLM error, unparseable
-        output) is swallowed and surfaced as an empty list — a valid result the
-        caller renders as "no suggestions". This method NEVER raises.
-        """
-        count = max(_SUGGESTION_MIN_COUNT, min(count, _SUGGESTION_MAX_COUNT))
-
-        try:
-            model_instance = ModelManager.for_tenant(tenant_id=tenant_id).get_default_model_instance(
-                tenant_id=tenant_id,
-                model_type=ModelType.LLM,
-            )
-        except Exception:
-            logger.info("Workflow instruction suggestions: no default model for tenant %s", tenant_id)
-            return []
-
-        context_block = cls._build_suggestion_context(tenant_id)
-        app_type_label = (
-            "Workflow — single-shot automation" if mode == "workflow" else "Chatflow — conversational multi-turn"
-        )
-
-        user_lines = [
-            f"App type: {app_type_label}",
-            context_block,
-            f"Return exactly {count} distinct ideas as a JSON array of strings.",
-        ]
-        if language:
-            user_lines.append(f"Write every idea in this language: {language}.")
-        user_prompt = "\n".join(line for line in user_lines if line)
-
-        prompt_messages: list[PromptMessage] = [
-            SystemPromptMessage(content=_SUGGESTION_SYSTEM_PROMPT),
-            UserPromptMessage(content=user_prompt),
-        ]
-
-        try:
-            response: LLMResult = model_instance.invoke_llm(
-                prompt_messages=prompt_messages,
-                model_parameters={"max_tokens": _SUGGESTION_MAX_TOKENS, "temperature": _SUGGESTION_TEMPERATURE},
-                stream=False,
-            )
-        except Exception:
-            logger.exception("Workflow instruction suggestions: LLM invocation failed")
-            return []
-
-        raw_suggestions = _parse_string_list(response.message.get_text_content() or "")
-
-        # Strip whitespace + surrounding quotes, drop empties, dedupe
-        # case-insensitively (preserving first-seen casing), cap to ``count``.
-        cleaned: list[str] = []
-        seen: set[str] = set()
-        for item in raw_suggestions:
-            idea = item.strip().strip("\"'").strip()
-            if not idea:
-                continue
-            key = idea.casefold()
-            if key in seen:
-                continue
-            seen.add(key)
-            cleaned.append(idea)
-            if len(cleaned) >= count:
-                break
-        return cleaned
-
-    @staticmethod
-    def _build_suggestion_context(tenant_id: str) -> str:
-        """Assemble an optional grounding block naming the tenant's KBs and tools.
-
-        Best-effort: each section is isolated in its own try/except so a failure
-        enumerating one (DB hiccup, plugin daemon down) never blocks the other
-        or the suggestion call itself. Returns "" when nothing is available.
-        """
-        sections: list[str] = []
-
-        try:
-            from models.dataset import Dataset
-
-            names = db.session.scalars(
-                select(Dataset.name)
-                .where(Dataset.tenant_id == tenant_id)
-                .order_by(Dataset.created_at.desc())
-                .limit(_SUGGESTION_KB_LIMIT)
-            ).all()
-            kb_names = [name for name in names if name]
-            if kb_names:
-                sections.append("Knowledge bases:\n" + "\n".join(f"- {name}" for name in kb_names))
-        except Exception:
-            logger.info("Workflow instruction suggestions: failed to load knowledge bases", exc_info=True)
-
-        try:
-            from core.workflow.generator.tool_catalogue import build_tool_catalogue, format_tool_catalogue
-
-            tool_text = format_tool_catalogue(build_tool_catalogue(tenant_id))
-            if tool_text:
-                sample = "\n".join(tool_text.splitlines()[:_SUGGESTION_TOOL_SAMPLE_LINES])
-                sections.append("Installed tools:\n" + sample)
-        except Exception:
-            logger.info("Workflow instruction suggestions: failed to load tool catalogue", exc_info=True)
-
-        if not sections:
-            return ""
-        return "\n\n".join(sections) + "\n\n"
-
     @classmethod
     @with_credit_usage_created_by(CreditUsageCreatedBy.RULE_CONFIG)
     def generate_rule_config(cls, tenant_id: str, args: RuleGeneratePayload, *, app_id: str | None = None):

api/core/workflow/generator/__init__.py

@@ -1,20 +1 @@
-"""
-Workflow generator package.
-
-Generates a Dify workflow graph (nodes, edges, viewport) from a natural-language
-instruction. Intended for the cmd+k `/create` slash command's preview/apply flow.
-
-Pipeline:
-
-    runner.WorkflowGenerator.generate_workflow_graph(...)
-        ├── planner_prompts: short LLM call → node and edge plan
-        ├── node_builder_prompts: bounded parallel calls → semantic node configs
-        └── postprocess: assemble wrappers, auto-layout, validate graph
-
-The runner is pure domain logic; ``WorkflowGeneratorService`` (in ``services/``)
-owns the model-manager dependency and is what controllers call.
-"""
-
-from .runner import WorkflowGenerator
-
-__all__ = ["WorkflowGenerator"]
+"""Graph generation values, prompts and pure graph transformations."""

api/core/workflow/generator/prompts/builder_prompts.py

@@ -5,7 +5,7 @@
 # Each entry mirrors the production ``defaultValue`` from
 # ``web/app/components/workflow/nodes/<type>/default.ts`` so the generated
 # graph loads in Studio identically to a manually-created node and survives
-# both ``WorkflowService.sync_draft_workflow``'s structural checks and the
+# both ``WorkflowDraftService.sync``'s structural checks and the
 # runtime entity validation each node performs when the workflow runs.
 #
 # Each snippet mirrors the production node default closely enough for one

api/core/workflow/generator/types.py

@@ -3,7 +3,7 @@
 
 These TypedDicts describe the planner payload and the runtime graph assembled
 from builder LLM responses after ``json_repair`` parsing. The graph types mirror
-the shape consumed by ``WorkflowService.sync_draft_workflow`` so the output can
+the shape consumed by ``WorkflowDraftService.sync`` so the output can
 be written straight into a draft workflow.
 """
 
@@ -145,7 +145,7 @@ class GraphViewportDict(TypedDict):
 
 
 class GraphDict(TypedDict):
-    """Full graph payload — matches ``WorkflowService.sync_draft_workflow``."""
+    """Full graph payload — matches ``WorkflowDraftService.sync``."""
 
     nodes: list[GraphNodeDict]
     edges: list[GraphEdgeDict]

api/extensions/application_services/workflow.py

@@ -159,3 +159,14 @@ def build_workflow_execution_dependencies(database_client: sessionmaker[Session]
             DifyCoreRepositoryFactory.create_workflow_node_execution_repository, session_factory=database_client
         ),
     )
+
+
+def build_workflow_suggestions(database_client: sessionmaker[Session]):
+    from repositories.knowledge.dataset_repository import SQLAlchemyDatasetRepository
+    from services.workflow.generation.suggestions import WorkflowInstructionSuggestions
+    from services.workflow.generation.tool_catalogue import build_tool_catalogue, format_tool_catalogue
+
+    return WorkflowInstructionSuggestions(
+        datasets=SQLAlchemyDatasetRepository(session_factory=database_client),
+        tools=lambda tenant_id: format_tool_catalogue(build_tool_catalogue(tenant_id)),
+    )

api/extensions/ext_application_services.py

@@ -40,6 +40,7 @@
 from extensions.application_services.rbac import RBACServices, build_rbac_services
 from extensions.application_services.resource_access_token import build_resource_access_token_service
 from extensions.application_services.trial_app import TrialAppServices, build_trial_app_services
+from extensions.application_services.workflow import build_workflow_suggestions
 from extensions.application_services.workspace import (
     WorkspaceServices,
     build_workspace_membership_services,
@@ -235,6 +236,7 @@
 from services.web_passport_service import WebPassportService
 from services.webapp_access_adapters import EnterpriseWebAppAccessPolicyGateway
 from services.webapp_access_query_service import WebAppAccessQueryService, WebAppAccessUnavailableError
+from services.workflow.generation.suggestions import WorkflowInstructionSuggestions
 from services.workflow_app_log_query_service import WorkflowAppLogQueryService
 from services.workflow_run_service import WorkflowRunService
 from services.workflow_statistic_query_service import WorkflowStatisticQueryService
@@ -294,6 +296,7 @@ class AppScopedEndUserServices:
 
 @dataclass(frozen=True, slots=True)
 class ApplicationServices:
+    workflow_suggestions: WorkflowInstructionSuggestions
     annotation_commands: AnnotationCommandService
     annotation_imports: AnnotationImportService
     annotation_queries: AnnotationQuery
@@ -602,6 +605,7 @@ def build_application_services(
     )
 
     return ApplicationServices(
+        workflow_suggestions=build_workflow_suggestions(database_client),
         annotation_commands=AnnotationCommandService(
             annotations=annotations,
             add_index=add_annotation_to_index_task.delay,

api/repositories/knowledge/dataset_repository.py

@@ -47,6 +47,17 @@ class SQLAlchemyDatasetRepository:
     def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
         self._session_factory = session_factory
 
+    def names(self, *, tenant_id: str, limit: int) -> list[str]:
+        with self._session_factory() as session:
+            return list(
+                session.scalars(
+                    select(Dataset.name)
+                    .where(Dataset.tenant_id == tenant_id)
+                    .order_by(Dataset.created_at.desc())
+                    .limit(limit)
+                )
+            )
+
     def get_maintainer_id(self, workspace_id: str, resource_id: str) -> str | None:
         with self._session_factory() as session:
             return session.scalar(