mirror of
https://github.com/langgenius/dify.git
synced 2026-07-19 16:38:31 +08:00
perf(workflow-generator): parallelize node config generation (#38975)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Crazywoola
GitHub
parent
50341357b3
commit
9de7e0fe44
22
.vscode/launch.json.template
vendored
22
.vscode/launch.json.template
vendored
@ -5,7 +5,9 @@
|
||||
"name": "Python: API (gevent)",
|
||||
"type": "debugpy",
|
||||
"request": "launch",
|
||||
"program": "${workspaceFolder}/api/app.py",
|
||||
"module": "gevent.monkey",
|
||||
"args": ["--module", "app"],
|
||||
"gevent": true,
|
||||
"jinja": true,
|
||||
"justMyCode": true,
|
||||
"cwd": "${workspaceFolder}/api",
|
||||
@ -33,22 +35,6 @@
|
||||
"justMyCode": false,
|
||||
"cwd": "${workspaceFolder}/api",
|
||||
"python": "${workspaceFolder}/api/.venv/bin/python"
|
||||
},
|
||||
{
|
||||
"name": "Next.js: debug full stack",
|
||||
"type": "node",
|
||||
"request": "launch",
|
||||
"program": "${workspaceFolder}/web/node_modules/next/dist/bin/next",
|
||||
"runtimeArgs": ["--inspect"],
|
||||
"skipFiles": ["<node_internals>/**"],
|
||||
"serverReadyAction": {
|
||||
"action": "debugWithChrome",
|
||||
"killOnServerStop": true,
|
||||
"pattern": "- Local:.+(https?://.+)",
|
||||
"uriFormat": "%s",
|
||||
"webRoot": "${workspaceFolder}/web"
|
||||
},
|
||||
"cwd": "${workspaceFolder}/web"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@ -561,6 +561,8 @@ WORKFLOW_MAX_EXECUTION_STEPS=500
|
||||
WORKFLOW_MAX_EXECUTION_TIME=1200
|
||||
WORKFLOW_CALL_MAX_DEPTH=5
|
||||
MAX_VARIABLE_SIZE=204800
|
||||
# Maximum concurrent node-builder LLM calls per workflow generation request
|
||||
WORKFLOW_GENERATOR_NODE_BUILDER_MAX_WORKERS=6
|
||||
|
||||
# GraphEngine Worker Pool Configuration
|
||||
# Minimum number of workers per GraphEngine instance (default: 1)
|
||||
|
||||
19
api/app.py
19
api/app.py
@ -1,5 +1,24 @@
|
||||
from __future__ import annotations
|
||||
|
||||
# ``python -m app`` (docker DEBUG=true, or IDE debugging) serves through the
|
||||
# gevent pywsgi server at the bottom of this file, so the stdlib must be
|
||||
# monkey-patched BEFORE any other import pulls in sockets or locks. Without
|
||||
# this, every request runs as a greenlet on one OS thread while blocking
|
||||
# calls (LLM invokes, ``Future.result`` waits, DB I/O) pin that thread — the
|
||||
# whole process freezes until the call returns. Gunicorn and Celery apply
|
||||
# their own patching (see gunicorn.conf.py / celery_entrypoint.py), and
|
||||
# ``flask run`` uses real Werkzeug threads, so both skip this branch.
|
||||
if __name__ == "__main__":
|
||||
from gevent import monkey
|
||||
|
||||
monkey.patch_all()
|
||||
|
||||
import psycogreen.gevent as psycogreen_gevent
|
||||
from grpc.experimental import gevent as grpc_gevent
|
||||
|
||||
grpc_gevent.init_gevent()
|
||||
psycogreen_gevent.patch_psycopg()
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
@ -784,6 +784,11 @@ class WorkflowConfig(BaseSettings):
|
||||
default=500,
|
||||
)
|
||||
|
||||
WORKFLOW_GENERATOR_NODE_BUILDER_MAX_WORKERS: PositiveInt = Field(
|
||||
description="Maximum concurrent node-builder LLM calls per workflow generation request",
|
||||
default=6,
|
||||
)
|
||||
|
||||
WORKFLOW_MAX_EXECUTION_TIME: PositiveInt = Field(
|
||||
description="Maximum execution time in seconds for a single workflow",
|
||||
default=1200,
|
||||
|
||||
@ -403,55 +403,6 @@ class LLMGenerator:
|
||||
return ""
|
||||
return "\n\n".join(sections) + "\n\n"
|
||||
|
||||
@classmethod
|
||||
def classify_workflow_mode(
|
||||
cls,
|
||||
tenant_id: str,
|
||||
instruction: str,
|
||||
model_config: ModelConfig,
|
||||
) -> Literal["workflow", "advanced-chat"]:
|
||||
"""Classify a free-text instruction into a concrete app mode.
|
||||
|
||||
One tiny LLM call using the model the user already picked (so no extra
|
||||
provider setup is needed). Parsed leniently; defaults to
|
||||
``advanced-chat`` on anything unexpected or any error, so a
|
||||
``mode="auto"`` request never blocks generation. NEVER raises.
|
||||
"""
|
||||
default_mode: Literal["workflow", "advanced-chat"] = "advanced-chat"
|
||||
try:
|
||||
model_instance = ModelManager.for_tenant(tenant_id=tenant_id).get_model_instance(
|
||||
tenant_id=tenant_id,
|
||||
model_type=ModelType.LLM,
|
||||
provider=model_config.provider,
|
||||
model=model_config.name,
|
||||
)
|
||||
prompt_messages: list[PromptMessage] = [
|
||||
UserPromptMessage(
|
||||
content=(
|
||||
"Reply with exactly one word: 'workflow' (one-shot automation, no chat) "
|
||||
"or 'advanced-chat' (conversational multi-turn). "
|
||||
f"Instruction: {instruction.strip()}"
|
||||
)
|
||||
),
|
||||
]
|
||||
response: LLMResult = model_instance.invoke_llm(
|
||||
prompt_messages=prompt_messages,
|
||||
model_parameters={"max_tokens": 4, "temperature": 0},
|
||||
stream=False,
|
||||
)
|
||||
text = (response.message.get_text_content() or "").strip().lower()
|
||||
except Exception:
|
||||
logger.info("Workflow mode classification failed; defaulting to %s", default_mode, exc_info=True)
|
||||
return default_mode
|
||||
|
||||
# Lenient parse: an affirmative "workflow" wins; everything else
|
||||
# (including a truncated / empty / garbled reply) falls back to the
|
||||
# conversational default. "advanced-chat" needs no positive match
|
||||
# because it IS the default.
|
||||
if "workflow" in text:
|
||||
return "workflow"
|
||||
return default_mode
|
||||
|
||||
@classmethod
|
||||
def generate_rule_config(cls, tenant_id: str, args: RuleGeneratePayload):
|
||||
output_parser = RuleConfigGeneratorOutputParser()
|
||||
|
||||
@ -4,12 +4,12 @@ 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 (slim, single-shot variant):
|
||||
Pipeline:
|
||||
|
||||
runner.WorkflowGenerator.generate_workflow_graph(...)
|
||||
├── planner_prompts: short LLM call → high-level node plan
|
||||
└── builder_prompts: structured-output LLM call → full graph JSON
|
||||
└── postprocess: fill defaults, auto-layout viewport, sanity-check edges
|
||||
├── 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.
|
||||
|
||||
@ -1,19 +1,4 @@
|
||||
"""
|
||||
Builder prompts.
|
||||
|
||||
The builder is the second step of the slim planner→builder pipeline. It takes
|
||||
the planner's high-level node list and emits the *full* graph JSON consumed by
|
||||
``WorkflowService.sync_draft_workflow``.
|
||||
|
||||
The builder owns: node configuration (prompts, code, headers, etc.), edge wiring,
|
||||
handle ids ("source"/"target"), positions, and the viewport. It is the only
|
||||
prompt that needs to know the concrete shape of each node type — keep its
|
||||
examples accurate or the LLM will invent fields.
|
||||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Iterable
|
||||
from typing import Any
|
||||
"""Compact semantic configuration references for workflow node builders."""
|
||||
|
||||
# Per-node-type configuration cheatsheet.
|
||||
#
|
||||
@ -23,50 +8,9 @@ from typing import Any
|
||||
# both ``WorkflowService.sync_draft_workflow``'s structural checks and the
|
||||
# runtime entity validation each node performs when the workflow runs.
|
||||
#
|
||||
# The cheatsheet is assembled DYNAMICALLY per request: the planner decides
|
||||
# which node types the workflow needs, and ``build_node_config_cheatsheet``
|
||||
# stitches together only the snippets for those types (plus the always-needed
|
||||
# wrapper / shared-field / edge-handle preamble, and the containers section
|
||||
# when an iteration / loop is planned). This keeps the builder prompt tight —
|
||||
# a 3-node summariser no longer carries the schema for 12 unrelated node
|
||||
# types — and lets each snippet document its FULL schema (e.g. a "file" start
|
||||
# variable's required ``allowed_file_types``) without bloating every prompt.
|
||||
#
|
||||
# The postprocessor in ``runner.py`` fills missing wrapper fields (``type``,
|
||||
# ``positionAbsolute``, ``width``, ``height``, ``sourcePosition`` /
|
||||
# ``targetPosition``, edge ``data.sourceType`` / ``data.targetType``), so the
|
||||
# LLM only needs to emit semantically meaningful fields.
|
||||
|
||||
# Always-included preamble: the node/edge wrapper shape and the shared
|
||||
# ``data`` fields that apply to every node type, plus the "## Per type" header
|
||||
# the per-type snippets slot under.
|
||||
_CHEATSHEET_PREAMBLE = """\
|
||||
## Node wrapper (every node, top-level)
|
||||
|
||||
{"id": "node1" (digits + letters only — see "Node IDs" below),
|
||||
"type": "custom", # ReactFlow renderer key. Iteration/loop
|
||||
# *start* children use special types
|
||||
# (see Containers below).
|
||||
"position": {"x": <number>, "y": <number>},
|
||||
"data": { ... per-type fields ... }}
|
||||
|
||||
Children of iteration / loop containers additionally need
|
||||
``parentId``, ``zIndex: 1002`` and ``extent: "parent"`` — see Containers.
|
||||
|
||||
## Shared "data" fields (every node)
|
||||
|
||||
{"type": "<node-type>", # e.g. "llm", "start", "if-else"
|
||||
"title": "<short label>",
|
||||
"desc": "<one-liner>",
|
||||
"selected": false}
|
||||
|
||||
## Per type — additional "data" fields (only the node types in your plan are shown)"""
|
||||
|
||||
|
||||
# node_type → its per-type schema snippet. Keyed by the exact ``node_type``
|
||||
# string the planner emits so ``build_node_config_cheatsheet`` can look each
|
||||
# one up directly. Iteration / loop are documented in the Containers section
|
||||
# (they are subgraphs, not leaf nodes) rather than here.
|
||||
# Each snippet mirrors the production node default closely enough for one
|
||||
# model call to emit only meaningful ``data`` fields. The runner owns wrappers,
|
||||
# topology, container metadata, layout, and validation.
|
||||
_NODE_SNIPPETS: dict[str, str] = {
|
||||
"start": """\
|
||||
- start:
|
||||
@ -248,484 +192,30 @@ _NODE_SNIPPETS: dict[str, str] = {
|
||||
Enable only the sub-features you need; ``conditions`` reuse the if-else
|
||||
condition shape (key / comparison_operator / value). Outputs: ``result``
|
||||
(the processed array), ``first_record``, ``last_record``.""",
|
||||
"assigner": """\
|
||||
- assigner (write to an existing conversation / loop variable):
|
||||
{"version": "2",
|
||||
"items": [{"variable_selector": ["<target-node>", "<target-var>"],
|
||||
"input_type": "variable",
|
||||
"operation": "over-write",
|
||||
"value": ["<source-node>", "<source-var>"]}]}
|
||||
``input_type`` is "variable" (value is a selector) or "constant".
|
||||
Operations: over-write | clear | append | extend | set | += | -= | *= |
|
||||
/= | remove-first | remove-last.""",
|
||||
"human-input": """\
|
||||
- human-input (pause for a person; use webapp delivery by default):
|
||||
{"delivery_methods": [{"id": "webapp", "type": "webapp", "enabled": true}],
|
||||
"form_content": "<short review / approval instructions>",
|
||||
"inputs": [{"type": "paragraph", "output_variable_name": "comment",
|
||||
"default": {"type": "constant", "selector": [], "value": ""}}],
|
||||
"user_actions": [{"id": "approve", "title": "Approve",
|
||||
"button_style": "primary"}],
|
||||
"timeout": 3, "timeout_unit": "day"}
|
||||
Each ``inputs[].output_variable_name`` is an output variable. Outgoing
|
||||
edges use the matching user-action id as ``sourceHandle``.""",
|
||||
}
|
||||
|
||||
|
||||
# Pulled into the cheatsheet only when an iteration / loop appears in the plan.
|
||||
_CONTAINERS_SECTION = """\
|
||||
## Containers — iteration / loop
|
||||
|
||||
These are SUBGRAPH nodes. To use one you MUST emit, in order:
|
||||
|
||||
1. The container node itself, e.g. for iteration:
|
||||
id: "nodeK"
|
||||
type: "custom"
|
||||
data: {"type": "iteration",
|
||||
"title": "<label>",
|
||||
"desc": "",
|
||||
"selected": false,
|
||||
"start_node_id": "nodeKstart",
|
||||
"iterator_selector": ["<src>", "<list-var>"],
|
||||
"output_selector": ["<inner-last-node>", "<out-var>"],
|
||||
"is_parallel": false,
|
||||
"parallel_nums": 10,
|
||||
"error_handle_mode": "terminated",
|
||||
"flatten_output": true}
|
||||
width: 808
|
||||
height: 204
|
||||
zIndex: 1
|
||||
|
||||
For loop, swap "iteration" → "loop" and use:
|
||||
data: {"type": "loop", "title": "...", "desc": "",
|
||||
"selected": false, "start_node_id": "nodeKstart",
|
||||
"break_conditions": [], "loop_count": 10,
|
||||
"logical_operator": "and"}
|
||||
|
||||
2. The auto-start child (one per container):
|
||||
id: "nodeKstart"
|
||||
type: "custom-iteration-start" # loop → "custom-loop-start"
|
||||
parentId: "nodeK"
|
||||
extent: "parent"
|
||||
draggable: false
|
||||
selectable: false
|
||||
zIndex: 1002
|
||||
position: {"x": 60, "y": 78} # relative to parent
|
||||
data: {"type": "iteration-start", # loop → "loop-start"
|
||||
"title": "", "desc": "",
|
||||
"isInIteration": true, # loop → "isInLoop": true
|
||||
"selected": false}
|
||||
|
||||
3. Each inner-pipeline node (any node type, follows normal data rules) MUST add:
|
||||
parentId: "nodeK"
|
||||
extent: "parent"
|
||||
zIndex: 1002
|
||||
position: {x, y} # relative to parent
|
||||
data: {..., "isInIteration": true, # loop → "isInLoop": true
|
||||
"iteration_id": "nodeK"} # loop → "loop_id"
|
||||
|
||||
4. Edges INSIDE a container must add to ``data``:
|
||||
"isInIteration": true # loop → "isInLoop": true
|
||||
"iteration_id": "nodeK" # loop → "loop_id"
|
||||
and use ``zIndex: 1002``. Edges OUTSIDE containers use the default
|
||||
``isInIteration: false`` / ``isInLoop: false``.
|
||||
|
||||
5. The container's incoming/outgoing edges connect to the container's id
|
||||
(``nodeK``), NOT to inner nodes. The first inner edge connects from
|
||||
``nodeKstart``."""
|
||||
|
||||
|
||||
# Always-included trailer: edge handle conventions for every graph.
|
||||
_EDGE_HANDLES_SECTION = """\
|
||||
## Edge handles
|
||||
|
||||
- Most nodes: sourceHandle "source", targetHandle "target".
|
||||
- if-else cases: sourceHandle is the case_id ("true" / "false" / ...).
|
||||
- question-classifier: sourceHandle is the class_id ("1" / "2" / ...).
|
||||
- iteration-start / sourceHandle "source"; the edge from the *start node
|
||||
loop-start: is what kicks off the first inner step."""
|
||||
|
||||
|
||||
# Container node types are described in ``_CONTAINERS_SECTION`` rather than as
|
||||
# leaf snippets; their presence in a plan pulls that section in.
|
||||
_CONTAINER_NODE_TYPES = frozenset({"iteration", "loop"})
|
||||
|
||||
|
||||
def build_node_config_cheatsheet(node_types: Iterable[str] | None = None) -> str:
|
||||
"""
|
||||
Assemble the builder cheatsheet for exactly the node types in the plan.
|
||||
|
||||
``node_types`` is the set of ``node_type`` strings the planner chose. We
|
||||
emit the always-on preamble (wrapper / shared fields), then only the
|
||||
per-type snippets for the requested types (``start`` is always included —
|
||||
every graph has one), the Containers section when an iteration / loop is
|
||||
planned, and the edge-handles trailer. Unknown / unrecognised type strings
|
||||
are ignored (the runtime / structural validator catches genuinely bogus
|
||||
types).
|
||||
|
||||
``None`` returns the FULL cheatsheet (every snippet + containers) — used to
|
||||
build the static back-compat constants below and as a safe fallback.
|
||||
"""
|
||||
if node_types is None:
|
||||
requested: set[str] = set(_NODE_SNIPPETS) | set(_CONTAINER_NODE_TYPES)
|
||||
else:
|
||||
requested = {str(t).strip() for t in node_types if str(t).strip()}
|
||||
requested.add("start") # every workflow has exactly one start node
|
||||
|
||||
parts: list[str] = [_CHEATSHEET_PREAMBLE]
|
||||
# Iterate _NODE_SNIPPETS (not ``requested``) to keep a stable, readable order.
|
||||
parts.extend(snippet for node_type, snippet in _NODE_SNIPPETS.items() if node_type in requested)
|
||||
if requested & _CONTAINER_NODE_TYPES:
|
||||
parts.append(_CONTAINERS_SECTION)
|
||||
parts.append(_EDGE_HANDLES_SECTION)
|
||||
return "\n\n".join(parts) + "\n"
|
||||
|
||||
|
||||
# Full cheatsheet (all node types) — retained as a module constant so callers
|
||||
# and tests that want the complete reference can import it directly. The
|
||||
# dynamic per-request prompt is built by ``get_builder_system_prompt``.
|
||||
NODE_CONFIG_CHEATSHEET = build_node_config_cheatsheet()
|
||||
|
||||
|
||||
_BASE_SYSTEM_PROMPT_HEAD = """You are a Dify workflow builder.
|
||||
|
||||
You are given:
|
||||
1. A user instruction (what the workflow should do).
|
||||
2. A node plan from the planner (which nodes to use, in execution order).
|
||||
|
||||
Your job: emit a complete Dify workflow graph as JSON. The graph will be written
|
||||
directly into a Studio draft, so it must be syntactically valid and structurally
|
||||
correct.
|
||||
|
||||
# Hard rules
|
||||
|
||||
1. The output is a single JSON object — no prose, no Markdown, no code fences.
|
||||
2. NODE IDs MUST USE ONLY ALPHANUMERICS + UNDERSCORES — never hyphens.
|
||||
Dify's run-time placeholder regex (see ``variable_pool.VARIABLE_PATTERN``)
|
||||
is ``\\{\\{#([a-zA-Z0-9_]{1,50}(?:\\.[a-zA-Z_][a-zA-Z0-9_]{0,29}){1,10})#\\}\\}``,
|
||||
so any placeholder pointing at a hyphenated id (e.g. ``{{#node-1.text#}}``)
|
||||
silently fails to match at run time and the literal string survives into
|
||||
the prompt — the user then sees ``{{#node-1.text#}}`` in their output.
|
||||
Use the EXACT ids from the plan, formatted as ``node1``, ``node2``, ... in
|
||||
plan order. Edge ``source`` / ``target`` must reference these ids.
|
||||
3. Every node has top-level fields: id, type, position, data.
|
||||
- "type" is always "custom" (ReactFlow node renderer).
|
||||
- "data.type" is the actual node type ("llm", "start", etc.).
|
||||
4. Every edge has top-level fields: id, source, target, type, sourceHandle, targetHandle.
|
||||
- "type" is always "custom".
|
||||
- "sourceHandle"/"targetHandle" follow the cheatsheet (default: "source"/"target").
|
||||
- Edge id format: "<source>-<sourceHandle>-<target>-<targetHandle>".
|
||||
5. Use the model from the planner context for ALL "llm" / "question-classifier" /
|
||||
"parameter-extractor" nodes (provider, name, mode, completion_params).
|
||||
6. Reference upstream outputs with the literal placeholder syntax
|
||||
``{{#<node-id>.<output-var>#}}`` — that's DOUBLE curly braces with ``#``
|
||||
markers inside (matching Dify's runtime placeholder regex
|
||||
``\\{\\{#[^#]+#\\}\\}``). NEVER emit single-brace ``{#…#}`` — Dify will
|
||||
not interpolate it, so the LLM at run time would see the literal
|
||||
placeholder string in its prompt and echo it back as output. Use
|
||||
``["<node-id>", "<output-var>"]`` for ``value_selector`` /
|
||||
``query_variable_selector`` / etc.
|
||||
7. The "start" node owns input variables; downstream nodes reference them as
|
||||
``["<start-node-id>", "<var-name>"]`` for selectors or
|
||||
``{{#<start-node-id>.<var-name>#}}`` inside prompt strings.
|
||||
8. NEVER emit "code" or "http-request" nodes if a tool from the "Available tools"
|
||||
section below covers the same task — replace them with a "tool" node referencing
|
||||
the exact provider/tool identifier from the catalogue. "code" / "http-request"
|
||||
are last-resort escape hatches for arbitrary transformations and APIs that no
|
||||
installed tool can express.
|
||||
9. EVERY variable reference MUST resolve to a real, declared variable on the
|
||||
source node — never invent a variable name. Specifically:
|
||||
- ``{{#<node-id>.<var>#}}`` inside a prompt / ``answer`` / ``template-transform``
|
||||
template (DOUBLE braces — single ``{#…#}`` is NOT a Dify placeholder
|
||||
and will NOT be substituted), AND ``["<node-id>", "<var>"]`` inside a
|
||||
``value_selector`` /
|
||||
``query_variable_selector`` / ``iterator_selector`` / ``output_selector`` /
|
||||
``tool_parameters[*].value`` (when ``type: "variable"``), MUST point at a
|
||||
value that the source node actually exposes:
|
||||
* ``start`` → one of the ``data.variables[*].variable`` entries you
|
||||
declared on the start node. Add an entry if you need a new input.
|
||||
* ``llm`` → ``text`` (the default LLM output) or, when structured
|
||||
output is enabled, a key from its schema.
|
||||
* ``code`` → a key in ``data.outputs``.
|
||||
* ``knowledge-retrieval`` → ``result`` (the standard array output).
|
||||
* ``parameter-extractor`` → one of the ``data.parameters[*].name``.
|
||||
* ``document-extractor`` → ``text`` (extracted file text; an array of
|
||||
strings when ``is_array_file`` is true).
|
||||
* ``variable-aggregator`` → ``output``.
|
||||
* ``list-operator`` → ``result`` (array), ``first_record``,
|
||||
``last_record``.
|
||||
* ``tool`` → any parameter declared by the tool — the run time
|
||||
validates these, so you can name them freely, but pick from the
|
||||
documented provider/tool.
|
||||
If the planner's "Start inputs" list (see user prompt) is non-empty,
|
||||
copy each entry verbatim into ``start.data.variables`` so the
|
||||
downstream references resolve.
|
||||
- In Advanced-Chat mode you may also reference ``sys.query`` and
|
||||
``sys.files`` without declaring them. In selector fields, spell these as
|
||||
exactly ``["sys", "query"]`` and ``["sys", "files"]`` — never as a
|
||||
one-item array such as ``["sys,query"]`` or ``["sys.query"]``.
|
||||
10. MULTIPLE KNOWLEDGE-RETRIEVAL INPUTS TO ONE LLM require a template fan-in.
|
||||
``context.variable_selector accepts only one selector`` and therefore
|
||||
cannot carry two retrieval outputs. When an LLM must synthesize two or
|
||||
more retrieval results:
|
||||
- Run the retrieval nodes as parallel siblings from the same query input.
|
||||
- Add one ``template-transform`` node after them. Give it one variable per
|
||||
retrieval, such as ``value_selector: ["node2", "result"]`` and
|
||||
``value_selector: ["node3", "result"]``, and render every source's
|
||||
content into one labelled text output.
|
||||
- Add an edge from EACH retrieval node to the template, then one edge from
|
||||
the template to the LLM. The LLM must NOT receive direct retrieval edges.
|
||||
- Enable the LLM's context, set ``context.variable_selector`` to the
|
||||
template's ``["<template-node-id>", "output"]``, and put
|
||||
``{{#context#}}`` in its prompt.
|
||||
- Do not use ``variable-aggregator`` for this: it selects the first value
|
||||
produced by mutually exclusive branches; it does not concatenate two
|
||||
retrieval results that both ran.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
_BASE_SYSTEM_PROMPT_TAIL = """\
|
||||
|
||||
# Layout
|
||||
|
||||
- Place nodes left-to-right with x=80 + 320 * index, y=280.
|
||||
- Viewport: {"x": 0, "y": 0, "zoom": 0.7}.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
_BASE_SYSTEM_PROMPT_FOOTER = """
|
||||
|
||||
# Output schema
|
||||
|
||||
{
|
||||
"nodes": [...],
|
||||
"edges": [...],
|
||||
"viewport": {"x": 0, "y": 0, "zoom": 0.7}
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
_WORKFLOW_MODE_RULES = """# Mode-specific rules — Workflow
|
||||
|
||||
- The graph MUST start with exactly one "start" node and end with exactly one "end" node.
|
||||
- Do NOT use "answer" nodes (those are for Advanced Chat only).
|
||||
- The "end" node's outputs[].value_selector must point at a real upstream output.
|
||||
"""
|
||||
|
||||
|
||||
_ADVANCED_CHAT_MODE_RULES = """# Mode-specific rules — Advanced Chat (Chatflow)
|
||||
|
||||
- The graph MUST start with exactly one "start" node and end with exactly one "answer" node.
|
||||
- Do NOT use "end" nodes (those are for plain Workflow apps).
|
||||
- The "start" node should expose "sys.query" / "sys.files" automatically; user-defined
|
||||
variables go in start.data.variables.
|
||||
- The "answer" node's "answer" field references upstream outputs as
|
||||
{{#<node-id>.<var>#}} and is what the user sees in chat.
|
||||
"""
|
||||
|
||||
|
||||
def _assemble_builder_system_prompt(mode: str, node_types: Iterable[str] | None) -> str:
|
||||
"""Stitch the builder system prompt for ``mode`` around a cheatsheet built
|
||||
for ``node_types`` (``None`` → full cheatsheet)."""
|
||||
mode_rules = _ADVANCED_CHAT_MODE_RULES if mode == "advanced-chat" else _WORKFLOW_MODE_RULES
|
||||
return (
|
||||
_BASE_SYSTEM_PROMPT_HEAD
|
||||
+ mode_rules
|
||||
+ _BASE_SYSTEM_PROMPT_TAIL
|
||||
+ build_node_config_cheatsheet(node_types)
|
||||
+ _BASE_SYSTEM_PROMPT_FOOTER
|
||||
)
|
||||
|
||||
|
||||
# Static full-cheatsheet prompts — the back-compat default returned by
|
||||
# ``get_builder_system_prompt`` when the caller doesn't pin a node-type set.
|
||||
BUILDER_SYSTEM_PROMPT_WORKFLOW = _assemble_builder_system_prompt("workflow", None)
|
||||
|
||||
BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT = _assemble_builder_system_prompt("advanced-chat", None)
|
||||
|
||||
|
||||
BUILDER_USER_PROMPT = """# User instruction
|
||||
|
||||
{instruction}
|
||||
|
||||
{ideal_output_section}\
|
||||
{existing_graph_section}\
|
||||
# Selected model (use for all LLM-based nodes)
|
||||
|
||||
provider={provider}, name={name}, mode={mode_label}
|
||||
|
||||
{tool_catalogue_section}\
|
||||
{start_inputs_section}\
|
||||
# Node plan (from planner — use these labels and node_types in this order)
|
||||
|
||||
{plan_block}
|
||||
|
||||
Now emit the complete workflow graph JSON.
|
||||
"""
|
||||
|
||||
|
||||
# Node wrapper fields that carry no meaning the builder needs: pure canvas /
|
||||
# selection state, plus geometry the runner's postprocess recomputes anyway.
|
||||
# Stripping them out of the refine prompt cuts its size roughly in half on
|
||||
# hand-edited graphs — fewer tokens in, and (because the builder echoes
|
||||
# untouched nodes verbatim) far fewer tokens out, which is where the latency
|
||||
# lives.
|
||||
_PRUNED_NODE_KEYS = frozenset(
|
||||
{
|
||||
"positionAbsolute",
|
||||
"sourcePosition",
|
||||
"targetPosition",
|
||||
"selected",
|
||||
"dragging",
|
||||
"measured",
|
||||
}
|
||||
)
|
||||
|
||||
# Additionally pruned from TOP-LEVEL nodes only: the layered auto-layout
|
||||
# recomputes their position and size defaults, so the builder never needs to
|
||||
# reproduce them. Container children keep ``position`` (relative to the
|
||||
# parent, which we cannot recompute) and containers keep ``width`` /
|
||||
# ``height`` (their canvas size is real config, not a default).
|
||||
_PRUNED_TOP_LEVEL_NODE_KEYS = _PRUNED_NODE_KEYS | {"position", "width", "height"}
|
||||
|
||||
_CONTAINER_DATA_TYPES = frozenset({"iteration", "loop"})
|
||||
|
||||
# Edge fields the builder must echo; everything else (ids, zIndex,
|
||||
# sourceType / targetType, isInIteration / isInLoop markers) is recomputed
|
||||
# by the runner's postprocess from the node topology.
|
||||
_KEPT_EDGE_KEYS = ("source", "target", "sourceHandle", "targetHandle")
|
||||
|
||||
|
||||
def compact_graph_for_builder(current_graph: dict) -> dict:
|
||||
"""
|
||||
Strip canvas noise out of a draft graph before prompt injection.
|
||||
|
||||
Keeps everything semantically meaningful — ids, wrapper ``type``,
|
||||
``parentId``, the full ``data`` config, child positions, container
|
||||
sizes — and drops geometry / selection state the postprocess pass
|
||||
recomputes. The builder echoes untouched nodes verbatim, so every byte
|
||||
removed here is removed twice (prompt AND completion).
|
||||
"""
|
||||
nodes_out: list[dict] = []
|
||||
for node in current_graph.get("nodes") or []:
|
||||
if not isinstance(node, dict):
|
||||
continue
|
||||
is_child = bool(node.get("parentId"))
|
||||
is_container = isinstance(node.get("data"), dict) and node["data"].get("type") in _CONTAINER_DATA_TYPES
|
||||
pruned = _PRUNED_NODE_KEYS if (is_child or is_container) else _PRUNED_TOP_LEVEL_NODE_KEYS
|
||||
compact = {k: v for k, v in node.items() if k not in pruned}
|
||||
if is_container:
|
||||
# Container position is still recomputed by the layout pass.
|
||||
compact.pop("position", None)
|
||||
nodes_out.append(compact)
|
||||
edges_out = [
|
||||
{k: edge[k] for k in _KEPT_EDGE_KEYS if k in edge}
|
||||
for edge in (current_graph.get("edges") or [])
|
||||
if isinstance(edge, dict)
|
||||
]
|
||||
return {"nodes": nodes_out, "edges": edges_out}
|
||||
|
||||
|
||||
def format_builder_existing_graph_section(current_graph: dict | None) -> str:
|
||||
"""
|
||||
Refine mode: give the builder the existing graph JSON so it can keep
|
||||
every node and edge the user's change does not touch byte-for-byte — same
|
||||
ids, same config, same prompt templates. Without the full config the
|
||||
builder would regenerate untouched nodes from scratch and silently drop
|
||||
the user's hand-tuned settings. Canvas-only fields are stripped first
|
||||
(see ``compact_graph_for_builder``) — they're recomputed in postprocess,
|
||||
so carrying them only slows the call down.
|
||||
|
||||
Returns an empty string in create mode (no ``current_graph``); the builder
|
||||
then behaves exactly as before, constructing the graph purely from the
|
||||
planner's node plan.
|
||||
"""
|
||||
if not current_graph:
|
||||
return ""
|
||||
graph_json = json.dumps(compact_graph_for_builder(current_graph), ensure_ascii=False, separators=(",", ":"))
|
||||
return (
|
||||
"# Existing graph to refine (JSON)\n\n"
|
||||
"You are REFINING this existing graph, NOT building from scratch. Apply "
|
||||
"ONLY the change the user instruction describes. Every node and edge the "
|
||||
"change does not affect MUST be preserved verbatim — keep the same node "
|
||||
"ids, the same `data` config, and the same prompt templates. The node "
|
||||
"plan below is the target node set after your change; use the existing "
|
||||
"graph as the source of truth for the config of nodes that carry over.\n\n"
|
||||
f"```json\n{graph_json}\n```\n\n"
|
||||
)
|
||||
|
||||
|
||||
def format_start_inputs_section(start_inputs: list[dict[str, Any]]) -> str:
|
||||
"""
|
||||
Surface the planner's ``start_inputs`` list to the builder so it can
|
||||
populate ``start.data.variables`` with the exact set of inputs every
|
||||
downstream variable reference will need. Empty list → empty section,
|
||||
because the builder may then declare no input variables (e.g. an
|
||||
Advanced-Chat workflow that only consumes ``sys.query``).
|
||||
"""
|
||||
if not start_inputs:
|
||||
return ""
|
||||
lines = ["# Start inputs (copy each entry verbatim into start.data.variables)"]
|
||||
lines.append("")
|
||||
for inp in start_inputs:
|
||||
variable = str(inp.get("variable") or "").strip()
|
||||
label = str(inp.get("label") or "").strip()
|
||||
type_ = str(inp.get("type") or "paragraph").strip()
|
||||
if not variable:
|
||||
continue
|
||||
lines.append(f"- variable={variable!r} label={label!r} type={type_!r}")
|
||||
lines.append("")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def format_builder_tool_catalogue_section(catalogue_text: str) -> str:
|
||||
"""
|
||||
Builder-facing catalogue block. The builder needs the same identifiers
|
||||
the planner saw, plus a stern reminder that ``tool`` nodes MUST set
|
||||
``provider_id`` / ``provider_name`` / ``tool_name`` to entries that
|
||||
actually exist in this list — hallucinated tools fail at draft sync.
|
||||
"""
|
||||
if not catalogue_text.strip():
|
||||
return ""
|
||||
return (
|
||||
"# Available tools (use these exact provider/tool identifiers — "
|
||||
"for each 'tool' node, set provider_id and provider_name to the "
|
||||
"provider portion and tool_name to the tool portion)\n\n"
|
||||
f"{catalogue_text}\n\n"
|
||||
)
|
||||
|
||||
|
||||
def format_plan_block(plan_nodes: list[dict[str, Any]]) -> str:
|
||||
"""
|
||||
Render the planner output as a numbered list the builder can quote.
|
||||
|
||||
Node IDs use no separator (``node1``, ``node2``, ...) because Dify's
|
||||
run-time placeholder regex requires ``[a-zA-Z0-9_]`` in the node-id
|
||||
slot — a hyphenated id like ``node-1`` would silently fail to match
|
||||
at run time and the literal ``{{#node-1.var#}}`` survives into the
|
||||
LLM prompt.
|
||||
|
||||
For container children (planner emitted a ``"parent": "<label>"`` key),
|
||||
we resolve the parent label to its ``nodeN`` id and surface it on the
|
||||
same line so the builder knows to set ``parentId`` and the
|
||||
``isInIteration`` / ``isInLoop`` markers on inner nodes.
|
||||
"""
|
||||
# First pass: label → node-id so we can resolve "parent" hints.
|
||||
label_to_id: dict[str, str] = {}
|
||||
for idx, node in enumerate(plan_nodes, start=1):
|
||||
label = str(node.get("label") or "")
|
||||
if label and label not in label_to_id:
|
||||
label_to_id[label] = f"node{idx}"
|
||||
|
||||
lines = []
|
||||
for idx, node in enumerate(plan_nodes, start=1):
|
||||
node_id = f"node{idx}"
|
||||
label = node.get("label", "")
|
||||
node_type = node.get("node_type", "")
|
||||
purpose = node.get("purpose", "")
|
||||
parent_label = str(node.get("parent") or "")
|
||||
parent_clause = ""
|
||||
if parent_label:
|
||||
parent_id = label_to_id.get(parent_label, "")
|
||||
if parent_id:
|
||||
parent_clause = f" parent={parent_id}"
|
||||
else:
|
||||
parent_clause = f" parent={parent_label!r}"
|
||||
lines.append(f"{idx}. id={node_id} type={node_type} label={label!r}{parent_clause}\n purpose: {purpose}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def get_builder_system_prompt(mode: str, node_types: Iterable[str] | None = None) -> str:
|
||||
"""
|
||||
Build the builder system prompt for ``mode``, with a cheatsheet scoped to
|
||||
``node_types`` (the planner's chosen node types).
|
||||
|
||||
When ``node_types`` is ``None`` we return the cached full-cheatsheet
|
||||
constant (back-compat default). When the runner passes the plan's node-type
|
||||
set we assemble a fresh prompt carrying only the relevant per-type schemas,
|
||||
so the builder isn't handed config for node types the workflow never uses.
|
||||
"""
|
||||
if node_types is None:
|
||||
return BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT if mode == "advanced-chat" else BUILDER_SYSTEM_PROMPT_WORKFLOW
|
||||
return _assemble_builder_system_prompt(mode, node_types)
|
||||
def get_node_config_snippet(node_type: str) -> str:
|
||||
"""Return the semantic config reference for one leaf node type."""
|
||||
return _NODE_SNIPPETS.get(node_type, "")
|
||||
|
||||
138
api/core/workflow/generator/prompts/node_builder_prompts.py
Normal file
138
api/core/workflow/generator/prompts/node_builder_prompts.py
Normal file
@ -0,0 +1,138 @@
|
||||
"""Compact prompts for parallel, per-node workflow configuration.
|
||||
|
||||
Each call produces only the semantic ``data`` fields for one planned node.
|
||||
Canvas wrappers, shared labels, topology, layout, and edge defaults are owned
|
||||
by ``WorkflowGenerator`` so completion length scales with node configuration
|
||||
rather than with the full ReactFlow graph.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from core.workflow.generator.prompts.builder_prompts import get_node_config_snippet
|
||||
|
||||
_CONTAINER_CONFIG_SNIPPETS = {
|
||||
"iteration": """- iteration:
|
||||
{"iterator_selector": ["<src>", "<list-var>"],
|
||||
"output_selector": ["<last-child>", "<out-var>"],
|
||||
"is_parallel": false, "parallel_nums": 10,
|
||||
"error_handle_mode": "terminated", "flatten_output": true}
|
||||
The runner supplies start_node_id, child wrappers, and the synthetic start node.""",
|
||||
"loop": """- loop:
|
||||
{"break_conditions": [{"id": "c1",
|
||||
"variable_selector": ["<child>", "<var>"],
|
||||
"comparison_operator": "is",
|
||||
"value": "<value>"}],
|
||||
"loop_count": 10, "logical_operator": "and"}
|
||||
The runner supplies start_node_id, child wrappers, and the synthetic start node.""",
|
||||
}
|
||||
|
||||
_NODE_BUILDER_HEAD = """You configure exactly ONE node in a Dify workflow.
|
||||
|
||||
Return one JSON object with exactly this shape: {"config": {...}}.
|
||||
``config`` contains only node-type-specific ``data`` fields. Do NOT repeat id,
|
||||
type, title, desc, selected, position, wrapper fields, edges, or viewport.
|
||||
|
||||
Rules:
|
||||
- Use only ids from the supplied normalized plan.
|
||||
- Placeholder strings use ``{{#node_id.variable#}}``; selector fields use
|
||||
``["node_id", "variable"]``. Never invent an upstream output.
|
||||
- Use the selected model verbatim for llm, question-classifier, and
|
||||
parameter-extractor nodes.
|
||||
- Keep prompts/code concise but complete for the user's requested behavior.
|
||||
- Emit strict JSON only: no prose, Markdown, comments, or trailing commas.
|
||||
|
||||
# Target node schema
|
||||
|
||||
"""
|
||||
|
||||
|
||||
NODE_BUILDER_USER_PROMPT = """# Target node
|
||||
|
||||
id={node_id}, type={node_type}, label={label!r}
|
||||
purpose={purpose}
|
||||
|
||||
# User instruction
|
||||
|
||||
{instruction}
|
||||
|
||||
{ideal_output_section}{mode_section}{model_section}{tool_catalogue_section}{start_inputs_section}{existing_config_section}\
|
||||
# Normalized plan and topology
|
||||
|
||||
{plan_json}
|
||||
|
||||
Return {{"config": {{...}}}} for target node {node_id} now.
|
||||
"""
|
||||
|
||||
|
||||
def get_node_builder_system_prompt(node_type: str) -> str:
|
||||
"""Build a one-node prompt containing only that node's semantic schema."""
|
||||
snippet = _CONTAINER_CONFIG_SNIPPETS.get(node_type) or get_node_config_snippet(node_type)
|
||||
return _NODE_BUILDER_HEAD + (snippet or f"- {node_type}: emit the minimum valid config fields.")
|
||||
|
||||
|
||||
def format_parallel_plan(
|
||||
plan_nodes: list[dict[str, Any]],
|
||||
plan_edges: list[dict[str, Any]],
|
||||
start_inputs: list[dict[str, Any]] | None = None,
|
||||
) -> str:
|
||||
"""Serialize the shared plan compactly so every node call has graph context.
|
||||
|
||||
``start_inputs`` rides along so downstream builders reference the declared
|
||||
``{{#<start-id>.<variable>#}}`` names instead of guessing them from prose
|
||||
— a guessed name gets auto-injected as a spurious form input later.
|
||||
"""
|
||||
payload: dict[str, Any] = {"nodes": plan_nodes, "edges": plan_edges}
|
||||
if start_inputs:
|
||||
payload["start_inputs"] = start_inputs
|
||||
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
||||
|
||||
|
||||
def format_mode_section(mode: str) -> str:
|
||||
"""Tell each builder which app mode it is configuring for.
|
||||
|
||||
Matters most in advanced-chat, where ``sys.query`` / ``sys.files`` are the
|
||||
sanctioned way to reference the user's message — without this the model
|
||||
invents start-node variables that postprocess then materializes as
|
||||
spurious form inputs.
|
||||
"""
|
||||
if mode == "advanced-chat":
|
||||
return (
|
||||
"# App mode\n\n"
|
||||
"advanced-chat: the user's chat message is available as sys.query and uploaded files "
|
||||
'as sys.files — placeholder {{#sys.query#}}, selector ["sys", "query"]. Reference them '
|
||||
"directly; do NOT invent start-node variables for the chat message.\n\n"
|
||||
)
|
||||
return (
|
||||
"# App mode\n\n"
|
||||
"workflow: there are NO automatic system variables; reference user input only through "
|
||||
"the start node's declared variables.\n\n"
|
||||
)
|
||||
|
||||
|
||||
def format_start_inputs_section(start_inputs: list[dict[str, Any]]) -> str:
|
||||
"""Render planner-declared inputs for the start-node builder only."""
|
||||
if not start_inputs:
|
||||
return ""
|
||||
lines = ["# Start inputs (copy each entry verbatim into start.data.variables)", ""]
|
||||
for input_ in start_inputs:
|
||||
variable = str(input_.get("variable") or "").strip()
|
||||
if not variable:
|
||||
continue
|
||||
label = str(input_.get("label") or "").strip()
|
||||
type_ = str(input_.get("type") or "paragraph").strip()
|
||||
lines.append(f"- variable={variable!r} label={label!r} type={type_!r}")
|
||||
lines.append("")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def format_tool_catalogue_section(catalogue_text: str) -> str:
|
||||
"""Render exact tool identifiers for a tool-node builder only."""
|
||||
if not catalogue_text.strip():
|
||||
return ""
|
||||
return (
|
||||
"# Available tools (use these exact provider/tool identifiers — "
|
||||
"set provider_id and provider_name to the provider portion and "
|
||||
"tool_name to the tool portion)\n\n"
|
||||
f"{catalogue_text}\n\n"
|
||||
)
|
||||
@ -1,13 +1,14 @@
|
||||
"""
|
||||
Planner prompts.
|
||||
|
||||
The planner is the lightweight first step in the slim planner→builder pipeline.
|
||||
The planner is the lightweight first step in the slim planner→node-builders pipeline.
|
||||
It receives the user's natural-language instruction and emits a high-level
|
||||
node plan in JSON. The builder later turns that plan into the final graph.
|
||||
node and edge plan in JSON. Node builders later produce configs that the runner
|
||||
assembles into the final graph.
|
||||
|
||||
We keep the planner deliberately short — the heavy lifting (config schemas,
|
||||
edge wiring, default values) belongs in the builder. The planner only commits
|
||||
to the *which-node-types* decision so the builder gets a tight scaffold.
|
||||
default values) belongs in the builders. The planner commits to the minimum
|
||||
topology and node types so every builder gets a tight scaffold.
|
||||
"""
|
||||
|
||||
PLANNER_SYSTEM_PROMPT = """You are a Dify workflow planner.
|
||||
@ -39,6 +40,8 @@ minimum set of Dify workflow nodes needed to fulfil it, in execution order.
|
||||
mutually-exclusive paths before "end" / "answer".
|
||||
- "list-operator" — filter / sort / slice an array variable (e.g. the items
|
||||
fed into or produced by an "iteration").
|
||||
- "assigner" — update an existing conversation or loop variable.
|
||||
- "human-input" — pause for a person to review, approve, or enter data.
|
||||
|
||||
# Rules
|
||||
|
||||
@ -97,22 +100,45 @@ minimum set of Dify workflow nodes needed to fulfil it, in execution order.
|
||||
variables are automatic — downstream nodes may reference them without
|
||||
a ``start_inputs`` entry. In Workflow mode there is NO automatic
|
||||
variable; everything the user supplies must be in ``start_inputs``.
|
||||
11. Output strictly the JSON object — no prose, no Markdown, no code fences.
|
||||
11. Give every node a unique runtime-safe ``id`` using only letters, digits,
|
||||
and underscores. In create mode use ``node1``, ``node2``, ... in node-list
|
||||
order. In refine mode preserve the existing id for every retained node.
|
||||
12. Emit the target graph's edges in ``edges``. Each edge is
|
||||
``{"source": "<id>", "target": "<id>"}``; add ``source_handle`` only
|
||||
for branch nodes: if-else case id, question-classifier class id, or
|
||||
human-input action id. Container children reference the container id in
|
||||
their ``parent`` field; do not emit the synthetic iteration/loop start node.
|
||||
13. In refine mode add ``action`` to every retained target node:
|
||||
``"keep"`` when its data config is unchanged, ``"update"`` when the user
|
||||
asked to change its config, and ``"add"`` for a new node. Removed nodes are
|
||||
omitted. Edge-only rewiring does not require changing a node's action.
|
||||
14. Output strictly the JSON object — no prose, no Markdown, no code fences.
|
||||
15. Echo the app mode in the ``mode`` output field — exactly "workflow" or
|
||||
"advanced-chat". When the ``# Mode`` section says auto, YOU decide:
|
||||
"workflow" for one-shot automations (run once with form inputs, return a
|
||||
result), "advanced-chat" for conversational multi-turn assistants. The
|
||||
terminal node must match the chosen mode (rule 2): "end" for workflow,
|
||||
"answer" for advanced-chat.
|
||||
|
||||
# Output schema
|
||||
|
||||
{
|
||||
"title": "<≤ 40-char title of the workflow>",
|
||||
"description": "<one-sentence summary>",
|
||||
"mode": "workflow | advanced-chat",
|
||||
"app_name": "<≤ 30-char product-style name, e.g. 'URL Summarizer'>",
|
||||
"icon": "<single emoji that captures the workflow's purpose, e.g. '📰'>",
|
||||
"start_inputs": [
|
||||
{"variable": "url", "label": "URL", "type": "text-input"}
|
||||
],
|
||||
"nodes": [
|
||||
{"label": "Start", "node_type": "start", "purpose": "..."},
|
||||
{"label": "Summarize", "node_type": "llm", "purpose": "..."},
|
||||
{"label": "End", "node_type": "end", "purpose": "..."}
|
||||
{"id": "node1", "label": "Start", "node_type": "start", "purpose": "..."},
|
||||
{"id": "node2", "label": "Summarize", "node_type": "llm", "purpose": "..."},
|
||||
{"id": "node3", "label": "End", "node_type": "end", "purpose": "..."}
|
||||
],
|
||||
"edges": [
|
||||
{"source": "node1", "target": "node2"},
|
||||
{"source": "node2", "target": "node3"}
|
||||
]
|
||||
}
|
||||
"""
|
||||
@ -140,7 +166,8 @@ def format_existing_graph_section(current_graph: dict | None) -> str:
|
||||
|
||||
We pass only ids / node-types / titles + edge endpoints here — the planner
|
||||
decides *which nodes* exist, so it needs the shape, not the per-node config.
|
||||
The builder gets the full graph JSON to preserve untouched node config.
|
||||
Node builders receive only the config of a node marked ``update``;
|
||||
configs marked ``keep`` are reused directly.
|
||||
"""
|
||||
if not current_graph:
|
||||
return ""
|
||||
@ -156,7 +183,13 @@ def format_existing_graph_section(current_graph: dict | None) -> str:
|
||||
for edge in edges:
|
||||
if not isinstance(edge, dict):
|
||||
continue
|
||||
edge_lines.append(f"- {edge.get('source', '')} -> {edge.get('target', '')}")
|
||||
# Branch wiring (if-else case ids, classifier class ids, human-input
|
||||
# action ids) lives in ``sourceHandle``. The planner is the only
|
||||
# source of edges for the rebuilt graph, so the real handle must be
|
||||
# surfaced here or refine silently rewires branches.
|
||||
handle = str(edge.get("sourceHandle") or "")
|
||||
handle_suffix = f" (source_handle={handle!r})" if handle and handle != "source" else ""
|
||||
edge_lines.append(f"- {edge.get('source', '')} -> {edge.get('target', '')}{handle_suffix}")
|
||||
nodes_block = "\n".join(node_lines) or "(none)"
|
||||
edges_block = "\n".join(edge_lines) or "(none)"
|
||||
return (
|
||||
@ -166,7 +199,9 @@ def format_existing_graph_section(current_graph: dict | None) -> str:
|
||||
"node list to reflect that change while keeping everything the "
|
||||
"instruction does not mention — preserve existing nodes, their order, "
|
||||
"and their labels wherever the change leaves them untouched. Only add, "
|
||||
"remove, or rename nodes the requested change actually requires.\n\n"
|
||||
"remove, or rename nodes the requested change actually requires. "
|
||||
"For every retained edge, copy its source_handle verbatim from the "
|
||||
"list below — branch wiring must survive the refine unchanged.\n\n"
|
||||
f"Current nodes:\n{nodes_block}\n\n"
|
||||
f"Current edges:\n{edges_block}\n\n"
|
||||
)
|
||||
|
||||
@ -1,14 +1,14 @@
|
||||
"""
|
||||
Workflow generator runner.
|
||||
|
||||
Slim planner→builder pipeline. Pure domain logic; the model instance is
|
||||
Slim planner→parallel-node-builder pipeline. Pure domain logic; the model instance is
|
||||
injected by ``WorkflowGeneratorService`` so this module stays cleanly
|
||||
separated from the infrastructure layer.
|
||||
|
||||
Pipeline:
|
||||
|
||||
1. PLANNER — short LLM call producing a high-level node list.
|
||||
2. BUILDER — structured-output LLM call producing the full graph JSON.
|
||||
2. BUILDERS — bounded concurrent LLM calls producing compact node configs.
|
||||
3. POSTPROC — fill safe defaults, lay nodes out left-to-right, dedupe
|
||||
edge ids, and run a final structural sanity check.
|
||||
|
||||
@ -20,7 +20,8 @@ Intentionally NOT here (deferred to a future iteration):
|
||||
- Tool / model catalogue filtering
|
||||
|
||||
If quality regresses below product threshold we add those back; for now the
|
||||
single planner+builder pair shipped behind cmd+k `/create` is enough.
|
||||
planner and bounded parallel node builders shipped behind cmd+k `/create` are
|
||||
enough.
|
||||
"""
|
||||
|
||||
import json
|
||||
@ -28,17 +29,22 @@ import logging
|
||||
import re
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from copy import deepcopy
|
||||
from typing import Any, ClassVar, cast
|
||||
|
||||
import json_repair
|
||||
|
||||
from core.workflow.generator.prompts.builder_prompts import (
|
||||
BUILDER_USER_PROMPT,
|
||||
format_builder_existing_graph_section,
|
||||
format_builder_tool_catalogue_section,
|
||||
format_plan_block,
|
||||
from configs import dify_config
|
||||
from core.workflow.generator.prompts.node_builder_prompts import (
|
||||
NODE_BUILDER_USER_PROMPT,
|
||||
format_mode_section,
|
||||
format_parallel_plan,
|
||||
format_start_inputs_section,
|
||||
get_builder_system_prompt,
|
||||
get_node_builder_system_prompt,
|
||||
)
|
||||
from core.workflow.generator.prompts.node_builder_prompts import (
|
||||
format_tool_catalogue_section as format_node_tool_catalogue_section,
|
||||
)
|
||||
from core.workflow.generator.prompts.planner_prompts import (
|
||||
PLANNER_SYSTEM_PROMPT,
|
||||
@ -55,6 +61,7 @@ from core.workflow.generator.types import (
|
||||
WorkflowGenerateErrorDict,
|
||||
WorkflowGenerateResultDict,
|
||||
WorkflowGenerationMode,
|
||||
WorkflowGenerationModeRequest,
|
||||
)
|
||||
from graphon.enums import BuiltinNodeTypes
|
||||
from graphon.model_runtime.entities.llm_entities import LLMResult
|
||||
@ -96,11 +103,31 @@ _DEFAULT_FILE_UPLOAD_METHODS = ("local_file", "remote_url")
|
||||
|
||||
# Token ceiling for the planner call when the caller didn't pin one. The plan
|
||||
# is a short JSON node list (a handful of nodes with labels/purposes), so this
|
||||
# is generous headroom while still bounding a runaway response. The builder is
|
||||
# left on the caller's budget — it emits the full graph and genuinely needs it.
|
||||
# is generous headroom while still bounding a runaway response. Builder calls
|
||||
# keep the caller's budget so complex node configs are not truncated.
|
||||
_PLANNER_DEFAULT_MAX_TOKENS = 4096
|
||||
|
||||
|
||||
# Per-node calls trade a larger request count for a shorter critical path.
|
||||
# The cap comes from ``WORKFLOW_GENERATOR_NODE_BUILDER_MAX_WORKERS`` (default 6,
|
||||
# enough to run the planner's recommended 3–6-node plans as a single wave);
|
||||
# provider rate-limit bursts are absorbed by ``_invoke_with_retry``'s bounded
|
||||
# backoff, and operators can dial the env var back down if their provider is
|
||||
# stricter. Read at call time so tests (and live config reloads) can adjust it
|
||||
# without re-importing.
|
||||
def _node_builder_max_workers() -> int:
|
||||
return dify_config.WORKFLOW_GENERATOR_NODE_BUILDER_MAX_WORKERS
|
||||
|
||||
|
||||
_MODEL_NODE_TYPES = frozenset(
|
||||
{
|
||||
BuiltinNodeTypes.LLM,
|
||||
BuiltinNodeTypes.QUESTION_CLASSIFIER,
|
||||
BuiltinNodeTypes.PARAMETER_EXTRACTOR,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# Appended as a trailing user message on the SECOND (and only) attempt when
|
||||
# the first response wasn't parseable as JSON. Keep this terse — the model
|
||||
# already has its full instructions in the original system message; this is
|
||||
@ -110,6 +137,12 @@ _JSON_RETRY_HINT = (
|
||||
"Do not include any prose, markdown code fences, comments, or trailing commas."
|
||||
)
|
||||
|
||||
_PLANNER_SCHEMA_RETRY_HINT = (
|
||||
"Your plan did not match the required topology schema. Return the complete plan again with "
|
||||
"a unique non-empty id on every node and a non-empty edges array whose source and target "
|
||||
"reference those ids. Return ONLY the JSON object."
|
||||
)
|
||||
|
||||
|
||||
# Provider hiccups we retry: a dropped connection, a 5xx, or a rate-limit are
|
||||
# all transient — the same request usually succeeds moments later. We do NOT
|
||||
@ -189,14 +222,58 @@ def _result_with_errors(
|
||||
def _with_mode(result: WorkflowGenerateResultDict, mode: WorkflowGenerationMode) -> WorkflowGenerateResultDict:
|
||||
"""Stamp the resolved concrete ``mode`` onto a result envelope.
|
||||
|
||||
``mode="auto"`` requests are resolved to a concrete mode before planning;
|
||||
echoing it back lets the frontend pick the right app type to create. It's
|
||||
present for explicit modes too so the response shape stays uniform.
|
||||
``mode="auto"`` requests are resolved to a concrete mode from the planner
|
||||
output; echoing it back lets the frontend pick the right app type to
|
||||
create. It's present for explicit modes too so the response shape stays
|
||||
uniform.
|
||||
"""
|
||||
result["mode"] = mode
|
||||
return result
|
||||
|
||||
|
||||
def _fallback_mode(mode: WorkflowGenerationModeRequest) -> WorkflowGenerationMode:
|
||||
"""Concrete mode for envelopes emitted before the planner resolved one.
|
||||
|
||||
``auto`` maps to the conversational default — the same never-fail fallback
|
||||
the old standalone classifier used — so ``result.mode`` never leaks the
|
||||
``auto`` sentinel to the frontend.
|
||||
"""
|
||||
return "advanced-chat" if mode == "auto" else mode
|
||||
|
||||
|
||||
def _planner_prompt_mode(mode: WorkflowGenerationModeRequest) -> str:
|
||||
"""Mode string interpolated into the planner user prompt.
|
||||
|
||||
For ``auto`` the value is self-describing so the planner knows the choice
|
||||
is delegated to it (system-prompt rule 15).
|
||||
"""
|
||||
return "auto (choose workflow or advanced-chat)" if mode == "auto" else mode
|
||||
|
||||
|
||||
def _resolve_generation_mode(
|
||||
requested: WorkflowGenerationModeRequest, plan: PlannerResultDict
|
||||
) -> WorkflowGenerationMode:
|
||||
"""Resolve the request mode into the concrete generation mode.
|
||||
|
||||
An explicit request always wins — a contradictory planner ``mode`` field is
|
||||
ignored. For ``auto``: trust the planner's echoed ``mode``, else infer from
|
||||
the plan's terminal node type (the structural source of truth the graph is
|
||||
validated against), else fall back to the conversational default. Lenient
|
||||
on purpose — a bad ``mode`` value must never fail the plan.
|
||||
"""
|
||||
if requested != "auto":
|
||||
return requested
|
||||
planner_mode = str(plan.get("mode") or "").strip().lower()
|
||||
if planner_mode in ("workflow", "advanced-chat"):
|
||||
return cast(WorkflowGenerationMode, planner_mode)
|
||||
node_types = {str(node.get("node_type") or "") for node in plan.get("nodes") or [] if isinstance(node, dict)}
|
||||
if BuiltinNodeTypes.ANSWER in node_types:
|
||||
return "advanced-chat"
|
||||
if BuiltinNodeTypes.END in node_types:
|
||||
return "workflow"
|
||||
return "advanced-chat"
|
||||
|
||||
|
||||
def _build_plan_event(
|
||||
*,
|
||||
plan: PlannerResultDict,
|
||||
@ -255,7 +332,7 @@ class WorkflowGenerator:
|
||||
provider: str,
|
||||
model_name: str,
|
||||
model_mode: str,
|
||||
mode: WorkflowGenerationMode,
|
||||
mode: WorkflowGenerationModeRequest,
|
||||
instruction: str,
|
||||
ideal_output: str = "",
|
||||
tool_catalogue_text: str = "",
|
||||
@ -263,20 +340,26 @@ class WorkflowGenerator:
|
||||
current_graph: dict[str, Any] | None = None,
|
||||
) -> WorkflowGenerateResultDict:
|
||||
"""
|
||||
Run planner → builder → postprocess and return a graph payload.
|
||||
Run planner → node builders → postprocess and return a graph payload.
|
||||
|
||||
``mode`` accepts the ``"auto"`` sentinel — the planner then chooses the
|
||||
concrete mode itself (echoed in its ``mode`` output field) so no extra
|
||||
classification call is needed; the resolution is stamped onto the
|
||||
result envelope.
|
||||
|
||||
``current_graph`` switches the pipeline from create mode to REFINE
|
||||
mode: the existing draft graph is injected into both the planner
|
||||
(compact node/edge summary) and the builder (full JSON) so the LLM
|
||||
amends the graph the user is editing instead of inventing a new one.
|
||||
``None`` (the default) is plain create-from-scratch behaviour.
|
||||
mode: the existing draft graph is summarized for the planner. Node
|
||||
builders receive only the config of the node they update, while configs
|
||||
marked ``keep`` are reused without an LLM call. ``None`` (the default)
|
||||
is plain create-from-scratch behaviour.
|
||||
|
||||
``tool_catalogue_text`` is the formatted list of installed tools for
|
||||
the calling tenant (see ``tool_catalogue.build_tool_catalogue`` /
|
||||
``format_tool_catalogue``). It's injected into both the planner and
|
||||
builder prompts so the LLM can pick concrete ``provider/tool``
|
||||
identifiers instead of inventing names; an empty string skips the
|
||||
section entirely (useful for unit tests).
|
||||
identifiers instead of inventing names; node builders receive it
|
||||
only for tool nodes. An empty string skips the section entirely (useful
|
||||
for unit tests).
|
||||
|
||||
``installed_tools`` is the structural sibling — a set of
|
||||
``(provider_name, tool_name)`` pairs the validator consults to reject
|
||||
@ -316,7 +399,7 @@ class WorkflowGenerator:
|
||||
# The event generator always emits exactly one result envelope; this
|
||||
# fallback only guards against a future refactor that forgets to.
|
||||
if result is None:
|
||||
result = _with_mode(_empty_result(), mode)
|
||||
result = _with_mode(_empty_result(), _fallback_mode(mode))
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
@ -328,7 +411,7 @@ class WorkflowGenerator:
|
||||
provider: str,
|
||||
model_name: str,
|
||||
model_mode: str,
|
||||
mode: WorkflowGenerationMode,
|
||||
mode: WorkflowGenerationModeRequest,
|
||||
instruction: str,
|
||||
ideal_output: str = "",
|
||||
tool_catalogue_text: str = "",
|
||||
@ -368,7 +451,7 @@ class WorkflowGenerator:
|
||||
provider: str,
|
||||
model_name: str,
|
||||
model_mode: str,
|
||||
mode: WorkflowGenerationMode,
|
||||
mode: WorkflowGenerationModeRequest,
|
||||
instruction: str,
|
||||
ideal_output: str = "",
|
||||
tool_catalogue_text: str = "",
|
||||
@ -376,7 +459,7 @@ class WorkflowGenerator:
|
||||
current_graph: dict[str, Any] | None = None,
|
||||
) -> Iterator[tuple[str, dict[str, Any]]]:
|
||||
"""
|
||||
Drive planner → builder → postprocess and yield generation events.
|
||||
Drive planner → node builders → postprocess and yield generation events.
|
||||
|
||||
Shared core for both ``generate_workflow_graph`` (keeps only the final
|
||||
``result``) and ``generate_workflow_graph_stream`` (streams every
|
||||
@ -402,11 +485,16 @@ class WorkflowGenerator:
|
||||
),
|
||||
)
|
||||
if plan_err is not None:
|
||||
yield "result", cast(dict[str, Any], _with_mode(_result_with_errors(_empty_result(), [plan_err]), mode))
|
||||
failed = _with_mode(_result_with_errors(_empty_result(), [plan_err]), _fallback_mode(mode))
|
||||
yield "result", cast(dict[str, Any], failed)
|
||||
return
|
||||
|
||||
# The lambda return is non-None when no error fired — narrow it for type-checkers.
|
||||
plan = cast(PlannerResultDict, plan)
|
||||
# ``auto`` requests resolve here — the planner echoed its mode choice
|
||||
# (or we infer it from the plan's terminal node). Explicit modes pass
|
||||
# through unchanged. Everything downstream uses the concrete mode.
|
||||
resolved_mode = _resolve_generation_mode(mode, plan)
|
||||
plan_nodes: list[dict[str, Any]] = cast(list[dict[str, Any]], plan.get("nodes", []))
|
||||
if not plan_nodes:
|
||||
empty_plan = _with_mode(
|
||||
@ -414,15 +502,12 @@ class WorkflowGenerator:
|
||||
_empty_result(),
|
||||
[_err(WorkflowGenerateErrorCode.EMPTY_PLAN, "Planner returned no nodes")],
|
||||
),
|
||||
mode,
|
||||
resolved_mode,
|
||||
)
|
||||
yield "result", cast(dict[str, Any], empty_plan)
|
||||
return
|
||||
|
||||
# A single LLM cannot select multiple retrieval outputs as context.
|
||||
# Make the required template fan-in explicit in the plan so the
|
||||
# builder receives its schema and assigns stable sequential ids.
|
||||
cls._insert_multi_retrieval_template_plan(plan_nodes)
|
||||
plan_edges = [cast(dict[str, Any], edge) for edge in (plan.get("edges") or []) if isinstance(edge, dict)]
|
||||
|
||||
# Planner-supplied user-input declarations. The builder uses these to
|
||||
# populate ``start.data.variables`` so downstream ``{#start.<var>#}``
|
||||
@ -436,34 +521,48 @@ class WorkflowGenerator:
|
||||
|
||||
# First event the stream sees: the high-level plan, before the slower
|
||||
# builder call. Non-streaming callers ignore it.
|
||||
yield "plan", _build_plan_event(plan=plan, plan_nodes=plan_nodes, start_inputs=start_inputs, mode=mode)
|
||||
yield "plan", _build_plan_event(plan=plan, plan_nodes=plan_nodes, start_inputs=start_inputs, mode=resolved_mode)
|
||||
|
||||
# ── 2. BUILDER ────────────────────────────────────────────────────
|
||||
graph, build_err = cls._run_stage(
|
||||
stage="Builder",
|
||||
failure_fallback_message="Failed to build workflow graph",
|
||||
run=lambda: cls._run_builder(
|
||||
builder_started_at = time.monotonic()
|
||||
|
||||
def build_graph() -> GraphDict:
|
||||
return cls._run_parallel_node_builders(
|
||||
model_instance=model_instance,
|
||||
model_parameters=model_parameters,
|
||||
provider=provider,
|
||||
model_name=model_name,
|
||||
model_mode=model_mode,
|
||||
mode=mode,
|
||||
mode=resolved_mode,
|
||||
instruction=instruction,
|
||||
ideal_output=ideal_output,
|
||||
plan_nodes=plan_nodes,
|
||||
plan_edges=plan_edges,
|
||||
tool_catalogue_text=tool_catalogue_text,
|
||||
start_inputs=start_inputs,
|
||||
current_graph=current_graph,
|
||||
),
|
||||
)
|
||||
|
||||
graph, build_err = cls._run_stage(
|
||||
stage="Builder",
|
||||
failure_fallback_message="Failed to build workflow graph",
|
||||
run=build_graph,
|
||||
)
|
||||
logger.info(
|
||||
"Workflow generator: node builders completed nodes=%s elapsed_ms=%.1f",
|
||||
len(plan_nodes),
|
||||
(time.monotonic() - builder_started_at) * 1000,
|
||||
)
|
||||
if build_err is not None:
|
||||
yield "result", cast(dict[str, Any], _with_mode(_result_with_errors(_empty_result(), [build_err]), mode))
|
||||
yield (
|
||||
"result",
|
||||
cast(dict[str, Any], _with_mode(_result_with_errors(_empty_result(), [build_err]), resolved_mode)),
|
||||
)
|
||||
return
|
||||
graph = cast(GraphDict, graph)
|
||||
|
||||
# ── 3. POSTPROC + VALIDATE ────────────────────────────────────────
|
||||
graph = cls._postprocess_graph(graph=graph, mode=mode)
|
||||
graph = cls._postprocess_graph(graph=graph, mode=resolved_mode)
|
||||
|
||||
# ``app_name`` / ``icon`` are planner display metadata; both default
|
||||
# to "" when the LLM omits them — the FE owns the fallback.
|
||||
@ -475,13 +574,13 @@ class WorkflowGenerator:
|
||||
"error": "",
|
||||
"errors": [],
|
||||
}
|
||||
_with_mode(result, mode)
|
||||
_with_mode(result, resolved_mode)
|
||||
|
||||
# Final structural sanity check — fail closed if start/end shape is
|
||||
# wrong, container topology is broken, a tool was hallucinated, or a
|
||||
# variable reference points at a node that won't expose it. We still
|
||||
# return the partial graph so the caller can debug or salvage it.
|
||||
structural_errors = cls._validate_structure(graph=graph, mode=mode, installed_tools=installed_tools)
|
||||
structural_errors = cls._validate_structure(graph=graph, mode=resolved_mode, installed_tools=installed_tools)
|
||||
if structural_errors:
|
||||
logger.warning("Workflow generator: structural validation failed: %s", structural_errors)
|
||||
yield "result", cast(dict[str, Any], _result_with_errors(result, structural_errors))
|
||||
@ -622,14 +721,14 @@ class WorkflowGenerator:
|
||||
*,
|
||||
model_instance,
|
||||
model_parameters: dict[str, Any],
|
||||
mode: WorkflowGenerationMode,
|
||||
mode: WorkflowGenerationModeRequest,
|
||||
instruction: str,
|
||||
ideal_output: str,
|
||||
tool_catalogue_text: str,
|
||||
current_graph: dict[str, Any] | None = None,
|
||||
) -> PlannerResultDict:
|
||||
user_prompt = PLANNER_USER_PROMPT.format(
|
||||
mode=mode,
|
||||
mode=_planner_prompt_mode(mode),
|
||||
instruction=instruction.strip(),
|
||||
existing_graph_section=format_existing_graph_section(current_graph),
|
||||
ideal_output_section=format_ideal_output_section(ideal_output),
|
||||
@ -639,59 +738,65 @@ class WorkflowGenerator:
|
||||
SystemPromptMessage(content=PLANNER_SYSTEM_PROMPT),
|
||||
UserPromptMessage(content=user_prompt),
|
||||
]
|
||||
clamped_parameters = _clamp_for_planner(model_parameters)
|
||||
parsed = cls._invoke_and_parse_json(
|
||||
model_instance=model_instance,
|
||||
messages=messages,
|
||||
model_parameters=_clamp_for_planner(model_parameters),
|
||||
model_parameters=clamped_parameters,
|
||||
stage="Planner",
|
||||
)
|
||||
try:
|
||||
return cls._validate_planner_schema(parsed)
|
||||
except _StageSchemaError:
|
||||
logger.info("Workflow generator: planner schema invalid; retrying once")
|
||||
parsed = cls._invoke_and_parse_json(
|
||||
model_instance=model_instance,
|
||||
messages=[*messages, UserPromptMessage(content=_PLANNER_SCHEMA_RETRY_HINT)],
|
||||
model_parameters=clamped_parameters,
|
||||
stage="Planner",
|
||||
)
|
||||
return cls._validate_planner_schema(parsed)
|
||||
|
||||
@staticmethod
|
||||
def _validate_planner_schema(parsed: dict[str, Any]) -> PlannerResultDict:
|
||||
"""Require the single planner contract consumed by node builders."""
|
||||
nodes = parsed.get("nodes")
|
||||
if not isinstance(nodes, list):
|
||||
raise _StageSchemaError("Planner", "missing 'nodes' array")
|
||||
if not nodes:
|
||||
return cast(PlannerResultDict, parsed)
|
||||
|
||||
node_ids: set[str] = set()
|
||||
for node in nodes:
|
||||
if not isinstance(node, dict) or "node_type" not in node:
|
||||
if not isinstance(node, dict) or not node.get("node_type"):
|
||||
raise _StageSchemaError("Planner", f"malformed node entry: {node!r}")
|
||||
node_id = node.get("id")
|
||||
if not isinstance(node_id, str) or not node_id.strip():
|
||||
raise _StageSchemaError("Planner", f"node missing non-empty id: {node!r}")
|
||||
if node_id in node_ids:
|
||||
raise _StageSchemaError("Planner", f"duplicate node id: {node_id!r}")
|
||||
node_ids.add(node_id)
|
||||
|
||||
edges = parsed.get("edges")
|
||||
if not isinstance(edges, list) or not edges:
|
||||
raise _StageSchemaError("Planner", "missing non-empty 'edges' array")
|
||||
for edge in edges:
|
||||
if not isinstance(edge, dict):
|
||||
raise _StageSchemaError("Planner", f"malformed edge entry: {edge!r}")
|
||||
source = edge.get("source")
|
||||
target = edge.get("target")
|
||||
if not isinstance(source, str) or not isinstance(target, str):
|
||||
raise _StageSchemaError("Planner", f"edge missing source or target: {edge!r}")
|
||||
if source not in node_ids or target not in node_ids:
|
||||
raise _StageSchemaError("Planner", f"edge references unknown node: {edge!r}")
|
||||
|
||||
return cast(PlannerResultDict, parsed)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Plan normalization
|
||||
# ------------------------------------------------------------------
|
||||
@staticmethod
|
||||
def _insert_multi_retrieval_template_plan(plan_nodes: list[dict[str, Any]]) -> None:
|
||||
"""Insert the unambiguous multi-retrieval template step when omitted.
|
||||
|
||||
Multiple LLM nodes make ownership ambiguous, so that case remains in
|
||||
the planner's hands. With exactly one LLM, every independent retrieval
|
||||
result can safely fan into one template immediately before that LLM.
|
||||
"""
|
||||
node_types = [str(node.get("node_type") or "") for node in plan_nodes]
|
||||
if node_types.count(BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL) < 2:
|
||||
return
|
||||
if node_types.count(BuiltinNodeTypes.LLM) != 1:
|
||||
return
|
||||
if BuiltinNodeTypes.TEMPLATE_TRANSFORM in node_types:
|
||||
return
|
||||
|
||||
llm_index = node_types.index(BuiltinNodeTypes.LLM)
|
||||
retrievals_before_llm = node_types[:llm_index].count(BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL)
|
||||
if retrievals_before_llm < 2:
|
||||
return
|
||||
plan_nodes.insert(
|
||||
llm_index,
|
||||
{
|
||||
"label": "Combine Knowledge",
|
||||
"node_type": BuiltinNodeTypes.TEMPLATE_TRANSFORM,
|
||||
"purpose": "Combine every knowledge retrieval result into one labelled context for the LLM.",
|
||||
},
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Builder
|
||||
# ------------------------------------------------------------------
|
||||
@classmethod
|
||||
def _run_builder(
|
||||
def _run_parallel_node_builders(
|
||||
cls,
|
||||
*,
|
||||
model_instance,
|
||||
@ -703,53 +808,275 @@ class WorkflowGenerator:
|
||||
instruction: str,
|
||||
ideal_output: str,
|
||||
plan_nodes: list[dict[str, Any]],
|
||||
plan_edges: list[dict[str, Any]],
|
||||
tool_catalogue_text: str,
|
||||
start_inputs: list[dict[str, Any]] | None = None,
|
||||
current_graph: dict[str, Any] | None = None,
|
||||
start_inputs: list[dict[str, Any]],
|
||||
current_graph: dict[str, Any] | None,
|
||||
) -> GraphDict:
|
||||
user_prompt = BUILDER_USER_PROMPT.format(
|
||||
"""Build changed node configs concurrently and expand them into a graph.
|
||||
|
||||
Refine plans can mark existing nodes as ``keep``; those nodes bypass
|
||||
the model entirely and retain their full data config. Every other node
|
||||
gets one compact call, with at most ``_node_builder_max_workers()``
|
||||
calls in flight. Any fragment failure aborts the graph, preserving the
|
||||
generator's existing fail-closed contract.
|
||||
"""
|
||||
existing_by_id = {
|
||||
str(node.get("id")): node
|
||||
for node in ((current_graph or {}).get("nodes") or [])
|
||||
if isinstance(node, dict) and node.get("id")
|
||||
}
|
||||
existing_edges = [edge for edge in ((current_graph or {}).get("edges") or []) if isinstance(edge, dict)]
|
||||
nodes_to_build = [
|
||||
node for node in plan_nodes if not (node.get("action") == "keep" and str(node.get("id")) in existing_by_id)
|
||||
]
|
||||
|
||||
# Shared across every builder call in this request — compute once.
|
||||
plan_json = format_parallel_plan(plan_nodes, plan_edges, start_inputs)
|
||||
mode_section = format_mode_section(mode)
|
||||
|
||||
configs_by_id: dict[str, dict[str, Any]] = {}
|
||||
if nodes_to_build:
|
||||
max_workers = min(_node_builder_max_workers(), len(nodes_to_build))
|
||||
with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="workflow-node-builder") as executor:
|
||||
futures = {
|
||||
executor.submit(
|
||||
cls._run_node_builder,
|
||||
model_instance=model_instance,
|
||||
model_parameters=model_parameters,
|
||||
provider=provider,
|
||||
model_name=model_name,
|
||||
model_mode=model_mode,
|
||||
mode_section=mode_section,
|
||||
instruction=instruction,
|
||||
ideal_output=ideal_output,
|
||||
target_node=node,
|
||||
plan_json=plan_json,
|
||||
tool_catalogue_text=tool_catalogue_text,
|
||||
start_inputs=start_inputs,
|
||||
existing_node=existing_by_id.get(str(node.get("id"))),
|
||||
): str(node.get("id"))
|
||||
for node in nodes_to_build
|
||||
}
|
||||
try:
|
||||
for future in as_completed(futures):
|
||||
node_id = futures[future]
|
||||
configs_by_id[node_id] = future.result()
|
||||
except BaseException:
|
||||
# Fail fast: one failed fragment aborts the whole graph, so
|
||||
# queued builder calls would only burn quota and delay the
|
||||
# error envelope. In-flight calls cannot be interrupted;
|
||||
# they finish while the pool shuts down.
|
||||
for pending in futures:
|
||||
pending.cancel()
|
||||
raise
|
||||
|
||||
return cls._assemble_parallel_graph(
|
||||
plan_nodes=plan_nodes,
|
||||
plan_edges=plan_edges,
|
||||
configs_by_id=configs_by_id,
|
||||
existing_by_id=existing_by_id,
|
||||
existing_edges=existing_edges,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _run_node_builder(
|
||||
cls,
|
||||
*,
|
||||
model_instance,
|
||||
model_parameters: dict[str, Any],
|
||||
provider: str,
|
||||
model_name: str,
|
||||
model_mode: str,
|
||||
mode_section: str,
|
||||
instruction: str,
|
||||
ideal_output: str,
|
||||
target_node: dict[str, Any],
|
||||
plan_json: str,
|
||||
tool_catalogue_text: str,
|
||||
start_inputs: list[dict[str, Any]],
|
||||
existing_node: dict[str, Any] | None,
|
||||
) -> dict[str, Any]:
|
||||
"""Generate only the semantic config for one normalized plan node."""
|
||||
node_id = str(target_node.get("id") or "")
|
||||
node_type = str(target_node.get("node_type") or "")
|
||||
model_section = ""
|
||||
if node_type in _MODEL_NODE_TYPES:
|
||||
model_section = (
|
||||
f"# Selected model (copy verbatim)\n\nprovider={provider}, name={model_name}, mode={model_mode}\n\n"
|
||||
)
|
||||
existing_config_section = ""
|
||||
if existing_node:
|
||||
existing_data = existing_node.get("data") if isinstance(existing_node.get("data"), dict) else {}
|
||||
existing_config_section = (
|
||||
"# Existing config to preserve unless the instruction changes it\n\n"
|
||||
f"{json.dumps(existing_data, ensure_ascii=False, separators=(',', ':'))}\n\n"
|
||||
)
|
||||
user_prompt = NODE_BUILDER_USER_PROMPT.format(
|
||||
node_id=node_id,
|
||||
node_type=node_type,
|
||||
label=str(target_node.get("label") or ""),
|
||||
purpose=str(target_node.get("purpose") or ""),
|
||||
instruction=instruction.strip(),
|
||||
ideal_output_section=format_ideal_output_section(ideal_output),
|
||||
existing_graph_section=format_builder_existing_graph_section(current_graph),
|
||||
provider=provider,
|
||||
name=model_name,
|
||||
mode_label=model_mode,
|
||||
plan_block=format_plan_block(plan_nodes),
|
||||
tool_catalogue_section=format_builder_tool_catalogue_section(tool_catalogue_text),
|
||||
start_inputs_section=format_start_inputs_section(start_inputs or []),
|
||||
mode_section=mode_section,
|
||||
model_section=model_section,
|
||||
tool_catalogue_section=(
|
||||
format_node_tool_catalogue_section(tool_catalogue_text) if node_type == BuiltinNodeTypes.TOOL else ""
|
||||
),
|
||||
start_inputs_section=(
|
||||
format_start_inputs_section(start_inputs) if node_type == BuiltinNodeTypes.START else ""
|
||||
),
|
||||
existing_config_section=existing_config_section,
|
||||
plan_json=plan_json,
|
||||
)
|
||||
# Scope the builder cheatsheet to exactly the node types the planner
|
||||
# chose, so the prompt carries each type's FULL schema (e.g. a file
|
||||
# start variable's required ``allowed_file_types``) without dragging in
|
||||
# config for unrelated node types.
|
||||
plan_node_types = {
|
||||
str(node.get("node_type") or "").strip() for node in plan_nodes if str(node.get("node_type") or "").strip()
|
||||
}
|
||||
messages = [
|
||||
SystemPromptMessage(content=get_builder_system_prompt(mode, plan_node_types)),
|
||||
UserPromptMessage(content=user_prompt),
|
||||
]
|
||||
parsed = cls._invoke_and_parse_json(
|
||||
model_instance=model_instance,
|
||||
messages=messages,
|
||||
messages=[
|
||||
SystemPromptMessage(content=get_node_builder_system_prompt(node_type)),
|
||||
UserPromptMessage(content=user_prompt),
|
||||
],
|
||||
model_parameters=model_parameters,
|
||||
stage="Builder",
|
||||
stage=f"Builder {node_id}",
|
||||
)
|
||||
config = parsed.get("config")
|
||||
if not isinstance(config, dict):
|
||||
raise _StageSchemaError(f"Builder {node_id}", "missing 'config' object")
|
||||
return cast(dict[str, Any], config)
|
||||
|
||||
nodes = parsed.get("nodes")
|
||||
edges = parsed.get("edges")
|
||||
if not isinstance(nodes, list) or not isinstance(edges, list):
|
||||
raise _StageSchemaError("Builder", "graph missing 'nodes' or 'edges' arrays")
|
||||
@classmethod
|
||||
def _assemble_parallel_graph(
|
||||
cls,
|
||||
*,
|
||||
plan_nodes: list[dict[str, Any]],
|
||||
plan_edges: list[dict[str, Any]],
|
||||
configs_by_id: dict[str, dict[str, Any]],
|
||||
existing_by_id: dict[str, dict[str, Any]],
|
||||
existing_edges: list[dict[str, Any]] | None = None,
|
||||
) -> GraphDict:
|
||||
"""Expand compact node configs and planner topology into graph JSON.
|
||||
|
||||
viewport = parsed.get("viewport") or _DEFAULT_VIEWPORT
|
||||
return cast(
|
||||
GraphDict,
|
||||
{
|
||||
"nodes": nodes,
|
||||
"edges": edges,
|
||||
"viewport": viewport,
|
||||
},
|
||||
)
|
||||
``existing_edges`` (refine only) preserves wiring the planner cannot
|
||||
express: the synthetic ``<container>start`` entry edge keeps its
|
||||
existing target instead of being re-pointed at whichever child the
|
||||
planner happened to list first.
|
||||
"""
|
||||
label_to_id = {
|
||||
str(node.get("label")): str(node.get("id")) for node in plan_nodes if node.get("label") and node.get("id")
|
||||
}
|
||||
type_by_id = {str(node.get("id")): str(node.get("node_type") or "") for node in plan_nodes}
|
||||
children_by_parent: dict[str, list[str]] = {}
|
||||
nodes: list[dict[str, Any]] = []
|
||||
|
||||
for planned in plan_nodes:
|
||||
node_id = str(planned.get("id") or "")
|
||||
node_type = str(planned.get("node_type") or "")
|
||||
existing = existing_by_id.get(node_id)
|
||||
node: dict[str, Any]
|
||||
if planned.get("action") == "keep" and existing is not None:
|
||||
node = deepcopy(existing)
|
||||
else:
|
||||
config = dict(configs_by_id.get(node_id) or {})
|
||||
for shared_key in ("type", "title", "desc", "selected"):
|
||||
config.pop(shared_key, None)
|
||||
data: dict[str, Any] = {
|
||||
"type": node_type,
|
||||
"title": str(planned.get("label") or node_id),
|
||||
"desc": str(planned.get("purpose") or ""),
|
||||
**config,
|
||||
}
|
||||
node = deepcopy(existing) if existing is not None else {"id": node_id}
|
||||
node["id"] = node_id
|
||||
node["data"] = data
|
||||
|
||||
parent_ref = str(planned.get("parent") or "")
|
||||
parent_id = label_to_id.get(parent_ref, parent_ref)
|
||||
if not parent_id and str(node.get("parentId") or "") in type_by_id:
|
||||
# Kept nodes rarely re-state containment — recover the parent
|
||||
# from the deepcopied wrapper so entry-edge synthesis still
|
||||
# counts this child.
|
||||
parent_id = str(node["parentId"])
|
||||
if parent_id:
|
||||
child_index = len(children_by_parent.get(parent_id, []))
|
||||
node["parentId"] = parent_id
|
||||
node.setdefault("position", {"x": 240 + 260 * child_index, "y": 60})
|
||||
node.setdefault("data", {})
|
||||
parent_type = type_by_id.get(parent_id)
|
||||
if parent_type == BuiltinNodeTypes.ITERATION:
|
||||
node["data"].setdefault("isInIteration", True)
|
||||
node["data"].setdefault("iteration_id", parent_id)
|
||||
elif parent_type == BuiltinNodeTypes.LOOP:
|
||||
node["data"].setdefault("isInLoop", True)
|
||||
node["data"].setdefault("loop_id", parent_id)
|
||||
children_by_parent.setdefault(parent_id, []).append(node_id)
|
||||
elif node.get("parentId"):
|
||||
# The container was dropped from the plan: strip the stale
|
||||
# containment markers so the kept node rejoins the top level
|
||||
# (and its auto-layout) instead of pointing at a deleted parent.
|
||||
for wrapper_key in ("parentId", "extent", "zIndex", "position", "positionAbsolute"):
|
||||
node.pop(wrapper_key, None)
|
||||
if isinstance(node.get("data"), dict):
|
||||
for marker_key in ("isInIteration", "iteration_id", "isInLoop", "loop_id"):
|
||||
node["data"].pop(marker_key, None)
|
||||
|
||||
nodes.append(node)
|
||||
if node_type in cls._CONTAINER_TYPES:
|
||||
start_id = f"{node_id}start"
|
||||
node.setdefault("data", {})["start_node_id"] = start_id
|
||||
node.setdefault("width", 808)
|
||||
node.setdefault("height", 204)
|
||||
node.setdefault("zIndex", 1)
|
||||
is_iteration = node_type == BuiltinNodeTypes.ITERATION
|
||||
nodes.append(
|
||||
{
|
||||
"id": start_id,
|
||||
"type": "custom-iteration-start" if is_iteration else "custom-loop-start",
|
||||
"parentId": node_id,
|
||||
"extent": "parent",
|
||||
"draggable": False,
|
||||
"selectable": False,
|
||||
"zIndex": 1002,
|
||||
"position": {"x": 60, "y": 78},
|
||||
"data": {
|
||||
"type": "iteration-start" if is_iteration else "loop-start",
|
||||
"title": "",
|
||||
"desc": "",
|
||||
"selected": False,
|
||||
"isInIteration" if is_iteration else "isInLoop": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
edges: list[dict[str, Any]] = []
|
||||
for planned_edge in plan_edges:
|
||||
edge: dict[str, Any] = {
|
||||
"source": str(planned_edge.get("source") or ""),
|
||||
"target": str(planned_edge.get("target") or ""),
|
||||
}
|
||||
source_handle = planned_edge.get("source_handle") or planned_edge.get("sourceHandle")
|
||||
target_handle = planned_edge.get("target_handle") or planned_edge.get("targetHandle")
|
||||
if source_handle:
|
||||
edge["sourceHandle"] = str(source_handle)
|
||||
if target_handle:
|
||||
edge["targetHandle"] = str(target_handle)
|
||||
edges.append(edge)
|
||||
|
||||
# Synthesize each container's entry edge. Refine keeps the existing
|
||||
# entry target when it is still a child — the planner's node listing
|
||||
# order says nothing about execution order inside a kept container.
|
||||
planned_sources = {str(edge.get("source") or "") for edge in edges}
|
||||
existing_entry_targets = {
|
||||
str(edge.get("source") or ""): str(edge.get("target") or "") for edge in (existing_edges or [])
|
||||
}
|
||||
for parent_id, child_ids in children_by_parent.items():
|
||||
start_id = f"{parent_id}start"
|
||||
if not child_ids or start_id in planned_sources:
|
||||
continue
|
||||
preferred = existing_entry_targets.get(start_id, "")
|
||||
entry_target = preferred if preferred in child_ids else child_ids[0]
|
||||
edges.append({"source": start_id, "target": entry_target})
|
||||
|
||||
return cast(GraphDict, {"nodes": nodes, "edges": edges, "viewport": _DEFAULT_VIEWPORT})
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Postprocessing
|
||||
@ -1225,6 +1552,11 @@ class WorkflowGenerator:
|
||||
return var == "output"
|
||||
if node_type == BuiltinNodeTypes.LIST_OPERATOR:
|
||||
return var in {"result", "first_record", "last_record"}
|
||||
if node_type == BuiltinNodeTypes.HUMAN_INPUT:
|
||||
return any(
|
||||
isinstance(item, dict) and item.get("output_variable_name") == var
|
||||
for item in (data.get("inputs") or [])
|
||||
)
|
||||
# Other node types (if-else, iteration-start, loop-start, ...) don't
|
||||
# produce outputs of their own.
|
||||
return False
|
||||
@ -1247,6 +1579,15 @@ class WorkflowGenerator:
|
||||
if isinstance(parameter, dict) and isinstance(parameter.get("name"), str)
|
||||
]
|
||||
return parameters[0] if len(parameters) == 1 else None
|
||||
if node_type == BuiltinNodeTypes.HUMAN_INPUT:
|
||||
human_outputs: list[str] = []
|
||||
for item in data.get("inputs") or []:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
output_name = item.get("output_variable_name")
|
||||
if isinstance(output_name, str):
|
||||
human_outputs.append(output_name)
|
||||
return human_outputs[0] if len(human_outputs) == 1 else None
|
||||
if not isinstance(node_type, str):
|
||||
return None
|
||||
single_output_by_type: dict[str, str] = {
|
||||
@ -1533,6 +1874,12 @@ class WorkflowGenerator:
|
||||
for klass in (data.get("classes") or [])
|
||||
if isinstance(klass, dict) and klass.get("id")
|
||||
]
|
||||
elif node_type == BuiltinNodeTypes.HUMAN_INPUT:
|
||||
branch_handles = [
|
||||
str(action["id"])
|
||||
for action in (data.get("user_actions") or [])
|
||||
if isinstance(action, dict) and action.get("id")
|
||||
]
|
||||
else:
|
||||
continue
|
||||
|
||||
@ -1941,8 +2288,8 @@ class WorkflowGenerator:
|
||||
"""
|
||||
Validate iteration / loop topology:
|
||||
|
||||
* every container has at least one child whose ``parentId``
|
||||
points at it;
|
||||
* every container has at least one executable child whose
|
||||
``parentId`` points at it;
|
||||
* every non-container node with a ``parentId`` points at a real
|
||||
container, not at a non-container node;
|
||||
* no cycles in the parent chain (a node cannot be its own
|
||||
@ -1959,7 +2306,9 @@ class WorkflowGenerator:
|
||||
if not isinstance(parent, str) or not parent:
|
||||
continue
|
||||
if parent in container_ids:
|
||||
children_by_parent.setdefault(parent, []).append(n.get("id", ""))
|
||||
node_type = (n.get("data") or {}).get("type")
|
||||
if node_type not in {"iteration-start", "loop-start"}:
|
||||
children_by_parent.setdefault(parent, []).append(n.get("id", ""))
|
||||
elif parent in by_id:
|
||||
# Parent exists but isn't a container — that's a topology bug.
|
||||
out.append(
|
||||
|
||||
@ -1,10 +1,10 @@
|
||||
"""
|
||||
Typed payloads for workflow generation.
|
||||
|
||||
These TypedDicts describe the shape that the planner and builder LLM calls are
|
||||
required to return after ``json_repair`` parsing. They mirror the runtime
|
||||
``graph`` shape consumed by ``WorkflowService.sync_draft_workflow`` so the output
|
||||
can be written straight into a draft workflow without further translation.
|
||||
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
|
||||
be written straight into a draft workflow.
|
||||
"""
|
||||
|
||||
from enum import StrEnum
|
||||
@ -12,11 +12,10 @@ from typing import Literal, NotRequired, TypedDict
|
||||
|
||||
WorkflowGenerationMode = Literal["workflow", "advanced-chat"]
|
||||
|
||||
# The mode accepted at the API boundary. ``auto`` is a sentinel that asks the
|
||||
# service to classify the instruction into a concrete ``WorkflowGenerationMode``
|
||||
# (one tiny LLM call) BEFORE planning — see
|
||||
# ``WorkflowGeneratorService._resolve_mode`` and
|
||||
# ``LLMGenerator.classify_workflow_mode``.
|
||||
# The mode accepted at the API boundary. ``auto`` is a sentinel that delegates
|
||||
# the choice to the planner: it echoes a concrete mode in its ``mode`` output
|
||||
# field (falling back to terminal-node inference, then ``advanced-chat``) —
|
||||
# see ``runner._resolve_generation_mode``. No extra LLM call is involved.
|
||||
WorkflowGenerationModeRequest = Literal["workflow", "advanced-chat", "auto"]
|
||||
|
||||
|
||||
@ -58,9 +57,21 @@ class WorkflowGenerateErrorDict(TypedDict):
|
||||
class PlannerNodeDict(TypedDict):
|
||||
"""One node from the planner's high-level plan."""
|
||||
|
||||
id: NotRequired[str]
|
||||
label: str
|
||||
node_type: str
|
||||
purpose: str
|
||||
parent: NotRequired[str]
|
||||
action: NotRequired[Literal["keep", "update", "add"]]
|
||||
|
||||
|
||||
class PlannerEdgeDict(TypedDict):
|
||||
"""Compact topology emitted by the planner for parallel node building."""
|
||||
|
||||
source: str
|
||||
target: str
|
||||
source_handle: NotRequired[str]
|
||||
target_handle: NotRequired[str]
|
||||
|
||||
|
||||
class PlannerStartInputDict(TypedDict):
|
||||
@ -82,10 +93,15 @@ class PlannerResultDict(TypedDict):
|
||||
|
||||
title: str
|
||||
description: str
|
||||
# Concrete mode the planner chose ("workflow" / "advanced-chat"). Parsed
|
||||
# leniently — an ``auto`` request infers the mode from the terminal node
|
||||
# when this is missing or invalid, so a bad value never fails the plan.
|
||||
mode: NotRequired[str]
|
||||
app_name: NotRequired[str]
|
||||
icon: NotRequired[str]
|
||||
start_inputs: NotRequired[list[PlannerStartInputDict]]
|
||||
nodes: list[PlannerNodeDict]
|
||||
edges: NotRequired[list[PlannerEdgeDict]]
|
||||
|
||||
|
||||
class GraphNodePositionDict(TypedDict):
|
||||
|
||||
@ -16,13 +16,11 @@ from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
from core.app.app_config.entities import ModelConfig
|
||||
from core.llm_generator.llm_generator import LLMGenerator
|
||||
from core.model_manager import ModelInstance, ModelManager
|
||||
from core.workflow.generator import WorkflowGenerator
|
||||
from core.workflow.generator.tool_catalogue import build_tool_catalogue, format_tool_catalogue, installed_tool_keys
|
||||
from core.workflow.generator.types import (
|
||||
WorkflowGenerateResultDict,
|
||||
WorkflowGenerationMode,
|
||||
WorkflowGenerationModeRequest,
|
||||
)
|
||||
from graphon.model_runtime.entities.model_entities import ModelType
|
||||
@ -52,11 +50,9 @@ class WorkflowGeneratorService:
|
||||
"""
|
||||
Resolve a model instance for the tenant and run the generator.
|
||||
|
||||
``mode`` accepts the ``"auto"`` sentinel — when set, the instruction is
|
||||
classified into a concrete ``workflow`` / ``advanced-chat`` mode (one
|
||||
tiny LLM call) before planning so the rest of the pipeline runs against
|
||||
a concrete mode. The resolved mode is echoed back under the result's
|
||||
``mode`` key.
|
||||
``mode`` accepts the ``"auto"`` sentinel — the planner itself picks the
|
||||
concrete ``workflow`` / ``advanced-chat`` mode (no extra LLM call) and
|
||||
the resolution is echoed back under the result's ``mode`` key.
|
||||
|
||||
``current_graph`` is the existing draft graph for the cmd+k `/refine`
|
||||
flow — when present the generator refines it instead of creating a new
|
||||
@ -66,9 +62,6 @@ class WorkflowGeneratorService:
|
||||
controller can map them to existing HTTP error envelopes (same
|
||||
envelope as ``/rule-generate``).
|
||||
"""
|
||||
resolved_mode = cls._resolve_mode(
|
||||
tenant_id=tenant_id, mode=mode, instruction=instruction, model_config=model_config
|
||||
)
|
||||
model_instance, model_parameters, tool_catalogue_text, installed_tools = cls._resolve_generation_context(
|
||||
tenant_id=tenant_id, model_config=model_config
|
||||
)
|
||||
@ -79,7 +72,7 @@ class WorkflowGeneratorService:
|
||||
provider=model_config.provider,
|
||||
model_name=model_config.name,
|
||||
model_mode=model_config.mode.value,
|
||||
mode=resolved_mode,
|
||||
mode=mode,
|
||||
instruction=instruction,
|
||||
ideal_output=ideal_output,
|
||||
tool_catalogue_text=tool_catalogue_text,
|
||||
@ -101,16 +94,13 @@ class WorkflowGeneratorService:
|
||||
"""
|
||||
Streaming sibling of ``generate_workflow_graph``.
|
||||
|
||||
Resolves the same model instance / tool catalogue / concrete mode, then
|
||||
delegates to ``WorkflowGenerator.generate_workflow_graph_stream`` and
|
||||
yields its ``(event_name, payload)`` tuples through to the controller's
|
||||
SSE writer. Provider-init / invoke errors raised while resolving the
|
||||
model instance propagate to the caller (the controller emits them as a
|
||||
Resolves the same model instance / tool catalogue, then delegates to
|
||||
``WorkflowGenerator.generate_workflow_graph_stream`` and yields its
|
||||
``(event_name, payload)`` tuples through to the controller's SSE
|
||||
writer. Provider-init / invoke errors raised while resolving the model
|
||||
instance propagate to the caller (the controller emits them as a
|
||||
single ``result`` SSE event).
|
||||
"""
|
||||
resolved_mode = cls._resolve_mode(
|
||||
tenant_id=tenant_id, mode=mode, instruction=instruction, model_config=model_config
|
||||
)
|
||||
model_instance, model_parameters, tool_catalogue_text, installed_tools = cls._resolve_generation_context(
|
||||
tenant_id=tenant_id, model_config=model_config
|
||||
)
|
||||
@ -121,7 +111,7 @@ class WorkflowGeneratorService:
|
||||
provider=model_config.provider,
|
||||
model_name=model_config.name,
|
||||
model_mode=model_config.mode.value,
|
||||
mode=resolved_mode,
|
||||
mode=mode,
|
||||
instruction=instruction,
|
||||
ideal_output=ideal_output,
|
||||
tool_catalogue_text=tool_catalogue_text,
|
||||
@ -129,28 +119,6 @@ class WorkflowGeneratorService:
|
||||
current_graph=current_graph,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _resolve_mode(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
mode: WorkflowGenerationModeRequest,
|
||||
instruction: str,
|
||||
model_config: ModelConfig,
|
||||
) -> WorkflowGenerationMode:
|
||||
"""Resolve the request mode into a concrete generation mode.
|
||||
|
||||
``"auto"`` triggers a one-word LLM classification using the model the
|
||||
user already picked; everything else passes through unchanged. The
|
||||
classifier never raises (defaults to ``advanced-chat``), so ``auto``
|
||||
never blocks generation.
|
||||
"""
|
||||
if mode == "auto":
|
||||
return LLMGenerator.classify_workflow_mode(
|
||||
tenant_id=tenant_id, instruction=instruction, model_config=model_config
|
||||
)
|
||||
return mode
|
||||
|
||||
@classmethod
|
||||
def _resolve_generation_context(
|
||||
cls,
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
import sys
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from core.app.app_config.entities import ModelConfig
|
||||
from core.llm_generator.llm_generator import LLMGenerator, _parse_string_list
|
||||
|
||||
|
||||
@ -112,37 +111,6 @@ class TestBuildSuggestionContext:
|
||||
assert LLMGenerator._build_suggestion_context("tenant") == ""
|
||||
|
||||
|
||||
class TestClassifyWorkflowMode:
|
||||
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
|
||||
def test_model_error(self, mock_for_tenant):
|
||||
mock_for_tenant.return_value.get_model_instance.side_effect = Exception("API error")
|
||||
|
||||
model_config = ModelConfig(provider="test", name="test", mode="chat")
|
||||
assert LLMGenerator.classify_workflow_mode("tenant", "instruction", model_config) == "advanced-chat"
|
||||
|
||||
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
|
||||
def test_workflow_match(self, mock_for_tenant):
|
||||
mock_model = MagicMock()
|
||||
mock_model.invoke_llm.return_value = MagicMock()
|
||||
mock_model.invoke_llm.return_value.message.get_text_content.return_value = " workflow "
|
||||
|
||||
mock_for_tenant.return_value.get_model_instance.return_value = mock_model
|
||||
|
||||
model_config = ModelConfig(provider="test", name="test", mode="chat")
|
||||
assert LLMGenerator.classify_workflow_mode("tenant", "instruction", model_config) == "workflow"
|
||||
|
||||
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
|
||||
def test_other_match(self, mock_for_tenant):
|
||||
mock_model = MagicMock()
|
||||
mock_model.invoke_llm.return_value = MagicMock()
|
||||
mock_model.invoke_llm.return_value.message.get_text_content.return_value = "chatflow"
|
||||
|
||||
mock_for_tenant.return_value.get_model_instance.return_value = mock_model
|
||||
|
||||
model_config = ModelConfig(provider="test", name="test", mode="chat")
|
||||
assert LLMGenerator.classify_workflow_mode("tenant", "instruction", model_config) == "advanced-chat"
|
||||
|
||||
|
||||
class TestWorkflowServiceInterface:
|
||||
def test_protocol_methods(self):
|
||||
# Just to cover the 'pass' statements in the Protocol definition
|
||||
|
||||
@ -1,25 +1,31 @@
|
||||
"""
|
||||
Unit tests for the planner / builder prompt format helpers.
|
||||
"""Unit tests for compact planner and per-node builder prompt helpers."""
|
||||
|
||||
These helpers are pure string-shaping functions that wrap conditional sections
|
||||
into the LLM prompts. We assert they (1) emit empty strings when the source
|
||||
data is empty so the prompt stays tight, (2) include the relevant header text
|
||||
when data is present, and (3) round-trip the raw catalogue text unchanged.
|
||||
"""
|
||||
import json
|
||||
|
||||
from core.workflow.generator.prompts.builder_prompts import (
|
||||
BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT,
|
||||
BUILDER_SYSTEM_PROMPT_WORKFLOW,
|
||||
compact_graph_for_builder,
|
||||
format_builder_existing_graph_section,
|
||||
format_builder_tool_catalogue_section,
|
||||
format_plan_block,
|
||||
get_builder_system_prompt,
|
||||
from core.workflow.generator.prompts.node_builder_prompts import (
|
||||
format_mode_section,
|
||||
format_parallel_plan,
|
||||
format_start_inputs_section,
|
||||
get_node_builder_system_prompt,
|
||||
)
|
||||
from core.workflow.generator.prompts.node_builder_prompts import (
|
||||
format_tool_catalogue_section as format_node_tool_catalogue_section,
|
||||
)
|
||||
from core.workflow.generator.prompts.planner_prompts import (
|
||||
PLANNER_SYSTEM_PROMPT,
|
||||
format_existing_graph_section,
|
||||
format_ideal_output_section,
|
||||
format_tool_catalogue_section,
|
||||
)
|
||||
from core.workflow.generator.prompts.planner_prompts import (
|
||||
format_tool_catalogue_section as format_planner_tool_catalogue_section,
|
||||
)
|
||||
|
||||
|
||||
class TestPlannerSystemPrompt:
|
||||
def test_documents_the_mode_output_field(self):
|
||||
"""Auto-mode resolution rides on the planner echoing its mode choice."""
|
||||
assert '"mode": "workflow | advanced-chat"' in PLANNER_SYSTEM_PROMPT
|
||||
assert "When the ``# Mode`` section says auto, YOU decide" in PLANNER_SYSTEM_PROMPT
|
||||
|
||||
|
||||
class TestFormatIdealOutputSection:
|
||||
@ -29,269 +35,129 @@ class TestFormatIdealOutputSection:
|
||||
|
||||
def test_wraps_content_in_a_labelled_section(self):
|
||||
out = format_ideal_output_section("A short summary.")
|
||||
|
||||
assert out.startswith("# Ideal output")
|
||||
assert "A short summary." in out
|
||||
assert out.endswith("\n\n")
|
||||
|
||||
|
||||
class TestPlannerCatalogueSection:
|
||||
def test_returns_empty_when_catalogue_is_blank(self):
|
||||
# No installed tools — the planner shouldn't see an "Available tools"
|
||||
# heading at all; an empty string keeps the prompt tight.
|
||||
assert format_tool_catalogue_section("") == ""
|
||||
assert format_tool_catalogue_section(" ") == ""
|
||||
class TestToolCatalogueSections:
|
||||
def test_planner_returns_empty_when_catalogue_is_blank(self):
|
||||
assert format_planner_tool_catalogue_section("") == ""
|
||||
assert format_planner_tool_catalogue_section(" ") == ""
|
||||
|
||||
def test_planner_includes_catalogue(self):
|
||||
out = format_planner_tool_catalogue_section("- google/search — Search.")
|
||||
|
||||
def test_emits_a_planner_facing_header_with_the_catalogue(self):
|
||||
out = format_tool_catalogue_section("- google/search — Search.")
|
||||
assert "# Available tools" in out
|
||||
assert "planner" in out.lower()
|
||||
assert "- google/search — Search." in out
|
||||
|
||||
def test_node_builder_returns_empty_when_catalogue_is_blank(self):
|
||||
assert format_node_tool_catalogue_section("") == ""
|
||||
|
||||
class TestBuilderCatalogueSection:
|
||||
def test_returns_empty_when_catalogue_is_blank(self):
|
||||
assert format_builder_tool_catalogue_section("") == ""
|
||||
def test_node_builder_requires_exact_provider_and_tool_ids(self):
|
||||
out = format_node_tool_catalogue_section("- google/search — Search.")
|
||||
|
||||
def test_includes_strict_provider_tool_guidance(self):
|
||||
out = format_builder_tool_catalogue_section("- google/search — Search.")
|
||||
# The builder must be told to use the *exact* identifiers — hallucinated
|
||||
# tools fail at sync time.
|
||||
assert "exact" in out.lower()
|
||||
assert "provider_id" in out
|
||||
assert "tool_name" in out
|
||||
assert "- google/search — Search." in out
|
||||
|
||||
|
||||
class TestFormatPlanBlock:
|
||||
def test_renders_one_line_per_node(self):
|
||||
out = format_plan_block(
|
||||
[
|
||||
{"label": "Start", "node_type": "start", "purpose": "Take input"},
|
||||
{"label": "Summarize", "node_type": "llm", "purpose": "Summarize"},
|
||||
]
|
||||
)
|
||||
lines = out.split("\n")
|
||||
# Two nodes → 4 lines (each entry takes id-line + purpose-line).
|
||||
assert any(line.startswith("1.") and "node1" in line for line in lines)
|
||||
assert any(line.startswith("2.") and "node2" in line for line in lines)
|
||||
assert "purpose: Take input" in out
|
||||
assert "purpose: Summarize" in out
|
||||
class TestNodeBuilderPrompt:
|
||||
def test_only_includes_target_node_schema_and_compact_output_contract(self):
|
||||
prompt = get_node_builder_system_prompt("llm")
|
||||
|
||||
def test_handles_missing_fields_gracefully(self):
|
||||
out = format_plan_block([{"node_type": "llm"}])
|
||||
# Missing label/purpose must not raise — they degrade to empty strings.
|
||||
assert "node1" in out
|
||||
assert "type=llm" in out
|
||||
|
||||
|
||||
class TestGetBuilderSystemPrompt:
|
||||
def test_returns_workflow_prompt_for_workflow_mode(self):
|
||||
# The two prompts are structurally similar but differ in their
|
||||
# mode-specific rules block.
|
||||
prompt = get_builder_system_prompt("workflow")
|
||||
assert prompt is BUILDER_SYSTEM_PROMPT_WORKFLOW
|
||||
assert 'exactly one "end" node' in prompt
|
||||
|
||||
def test_returns_advanced_chat_prompt_for_advanced_chat_mode(self):
|
||||
prompt = get_builder_system_prompt("advanced-chat")
|
||||
assert prompt is BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT
|
||||
assert 'exactly one "answer" node' in prompt
|
||||
|
||||
def test_scopes_cheatsheet_to_planned_node_types(self):
|
||||
# When the runner pins the plan's node-type set, the builder prompt
|
||||
# carries ONLY those types' schemas — no schema for unrelated nodes.
|
||||
prompt = get_builder_system_prompt("workflow", {"start", "llm", "end"})
|
||||
assert "- start:" in prompt
|
||||
assert '"config"' in prompt
|
||||
assert "- llm:" in prompt
|
||||
assert "- if-else:" not in prompt
|
||||
assert "- tool" not in prompt
|
||||
assert "## Containers" not in prompt
|
||||
# Still a valid, mode-correct prompt.
|
||||
assert 'exactly one "end" node' in prompt
|
||||
assert '"viewport":' not in prompt
|
||||
assert '"positionAbsolute":' not in prompt
|
||||
|
||||
def test_scoped_prompt_pulls_in_containers_for_iteration(self):
|
||||
prompt = get_builder_system_prompt("workflow", {"start", "iteration", "llm", "end"})
|
||||
assert "## Containers" in prompt
|
||||
def test_supports_main_human_input_and_assigner_contracts(self):
|
||||
human_input = get_node_builder_system_prompt("human-input")
|
||||
assigner = get_node_builder_system_prompt("assigner")
|
||||
|
||||
def test_scoped_prompt_is_smaller_than_full(self):
|
||||
# The whole point of dynamic assembly: a small plan ships a smaller
|
||||
# builder prompt than the full cheatsheet.
|
||||
scoped = get_builder_system_prompt("workflow", {"start", "llm", "end"})
|
||||
assert len(scoped) < len(BUILDER_SYSTEM_PROMPT_WORKFLOW)
|
||||
assert "delivery_methods" in human_input
|
||||
assert "user_actions" in human_input
|
||||
assert '"version": "2"' in assigner
|
||||
assert "variable_selector" in assigner
|
||||
|
||||
def test_documents_multi_retrieval_fan_in(self):
|
||||
prompt = get_builder_system_prompt(
|
||||
"workflow",
|
||||
{"start", "knowledge-retrieval", "llm", "end"},
|
||||
def test_common_node_prompts_stay_small(self):
|
||||
sizes = [len(get_node_builder_system_prompt(node_type)) for node_type in ("start", "llm", "end")]
|
||||
|
||||
assert max(sizes) < 3000
|
||||
|
||||
def test_unknown_node_type_gets_minimal_fallback(self):
|
||||
prompt = get_node_builder_system_prompt("future-node")
|
||||
|
||||
assert "future-node" in prompt
|
||||
assert "minimum valid config fields" in prompt
|
||||
|
||||
|
||||
class TestNodeBuilderUserSections:
|
||||
def test_formats_start_inputs(self):
|
||||
out = format_start_inputs_section(
|
||||
[{"variable": "url", "label": "URL", "type": "text-input"}, {"variable": "", "label": "Ignored"}]
|
||||
)
|
||||
|
||||
assert "context.variable_selector accepts only one selector" in prompt
|
||||
assert 'value_selector: ["node2", "result"]' in prompt
|
||||
assert 'value_selector: ["node3", "result"]' in prompt
|
||||
assert "edge from EACH retrieval node to the template" in prompt
|
||||
assert 'template\'s ``["<template-node-id>", "output"]``' in prompt
|
||||
assert "variable='url'" in out
|
||||
assert "type='text-input'" in out
|
||||
assert "Ignored" not in out
|
||||
|
||||
def test_empty_start_inputs_are_omitted(self):
|
||||
assert format_start_inputs_section([]) == ""
|
||||
|
||||
class TestBuildNodeConfigCheatsheet:
|
||||
def test_none_returns_full_cheatsheet(self):
|
||||
from core.workflow.generator.prompts.builder_prompts import (
|
||||
NODE_CONFIG_CHEATSHEET,
|
||||
build_node_config_cheatsheet,
|
||||
def test_parallel_plan_is_compact_and_preserves_topology(self):
|
||||
rendered = format_parallel_plan(
|
||||
[{"id": "node1", "node_type": "start"}, {"id": "node2", "node_type": "end"}],
|
||||
[{"source": "node1", "target": "node2"}],
|
||||
)
|
||||
|
||||
full = build_node_config_cheatsheet(None)
|
||||
assert full == NODE_CONFIG_CHEATSHEET
|
||||
# Full cheatsheet documents every node type + containers.
|
||||
assert "- tool" in full
|
||||
assert "- if-else:" in full
|
||||
assert "## Containers" in full
|
||||
assert " " not in rendered
|
||||
assert json.loads(rendered)["edges"] == [{"source": "node1", "target": "node2"}]
|
||||
assert "start_inputs" not in json.loads(rendered)
|
||||
|
||||
def test_always_includes_start_even_when_omitted(self):
|
||||
# Every workflow has a start node; the assembler force-includes it so
|
||||
# the builder can always declare input variables.
|
||||
from core.workflow.generator.prompts.builder_prompts import build_node_config_cheatsheet
|
||||
|
||||
out = build_node_config_cheatsheet({"llm", "end"})
|
||||
assert "- start:" in out
|
||||
|
||||
def test_start_snippet_documents_file_upload_schema(self):
|
||||
# The bug this fixes: a file start variable needs allowed_file_types,
|
||||
# which the builder never knew about. The snippet must now teach it.
|
||||
from core.workflow.generator.prompts.builder_prompts import build_node_config_cheatsheet
|
||||
|
||||
out = build_node_config_cheatsheet({"start", "document-extractor", "llm", "end"})
|
||||
assert "allowed_file_types" in out
|
||||
assert "allowed_file_upload_methods" in out
|
||||
assert "supported file types" in out # the exact Studio error wording
|
||||
|
||||
|
||||
class TestFormatPlanBlockParentHints:
|
||||
def test_resolves_parent_label_to_node_id(self):
|
||||
# The planner emits parent="Per Item" as a hint; the builder needs the
|
||||
# resolved id ("node-N") to set parentId on the inner node.
|
||||
from core.workflow.generator.prompts.builder_prompts import format_plan_block
|
||||
|
||||
out = format_plan_block(
|
||||
[
|
||||
{"label": "Start", "node_type": "start", "purpose": "x"},
|
||||
{"label": "Per Item", "node_type": "iteration", "purpose": "iterate"},
|
||||
{"label": "Sum Item", "node_type": "llm", "purpose": "summarize one", "parent": "Per Item"},
|
||||
]
|
||||
def test_parallel_plan_carries_declared_start_inputs(self):
|
||||
rendered = format_parallel_plan(
|
||||
[{"id": "node1", "node_type": "start"}],
|
||||
[],
|
||||
[{"variable": "url", "label": "URL", "type": "text-input"}],
|
||||
)
|
||||
# The inner line should mention parent=node2 (the iteration node).
|
||||
assert "parent=node2" in out
|
||||
# Top-level nodes must not have a parent clause.
|
||||
first_line = out.splitlines()[0]
|
||||
assert "parent=" not in first_line
|
||||
|
||||
def test_omits_parent_clause_when_label_is_unknown(self):
|
||||
# A typo / unknown parent label should degrade to quoting the raw
|
||||
# label string rather than fabricating a node id.
|
||||
from core.workflow.generator.prompts.builder_prompts import format_plan_block
|
||||
assert json.loads(rendered)["start_inputs"] == [{"variable": "url", "label": "URL", "type": "text-input"}]
|
||||
|
||||
out = format_plan_block(
|
||||
[
|
||||
{"label": "Start", "node_type": "start", "purpose": "x"},
|
||||
{"label": "Step", "node_type": "code", "purpose": "x", "parent": "Ghost Container"},
|
||||
]
|
||||
|
||||
class TestModeSection:
|
||||
def test_advanced_chat_documents_system_variables(self):
|
||||
out = format_mode_section("advanced-chat")
|
||||
|
||||
assert "sys.query" in out
|
||||
assert '["sys", "query"]' in out
|
||||
assert "do NOT invent start-node variables" in out
|
||||
|
||||
def test_workflow_mode_forbids_system_variables(self):
|
||||
out = format_mode_section("workflow")
|
||||
|
||||
assert "NO automatic system variables" in out
|
||||
|
||||
|
||||
class TestExistingGraphSection:
|
||||
def test_edge_lines_surface_branch_source_handles(self):
|
||||
out = format_existing_graph_section(
|
||||
{
|
||||
"nodes": [{"id": "node1", "data": {"type": "if-else", "title": "Branch"}}],
|
||||
"edges": [
|
||||
{"source": "node1", "target": "node2", "sourceHandle": "case-uuid-1"},
|
||||
{"source": "node2", "target": "node3", "sourceHandle": "source"},
|
||||
],
|
||||
}
|
||||
)
|
||||
assert "parent='Ghost Container'" in out
|
||||
|
||||
assert "- node1 -> node2 (source_handle='case-uuid-1')" in out
|
||||
assert "- node2 -> node3\n" in out
|
||||
assert "copy its source_handle verbatim" in out
|
||||
|
||||
class TestCompactGraphForBuilder:
|
||||
"""
|
||||
The refine-mode existing-graph JSON is the single biggest token sink in
|
||||
the pipeline — and the builder echoes untouched nodes back, doubling the
|
||||
cost. The compactor must drop canvas noise (recomputed in postprocess)
|
||||
while keeping everything the builder genuinely has to preserve.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _graph() -> dict:
|
||||
return {
|
||||
"nodes": [
|
||||
{
|
||||
"id": "node1",
|
||||
"type": "custom",
|
||||
"position": {"x": 80, "y": 282},
|
||||
"positionAbsolute": {"x": 80, "y": 282},
|
||||
"width": 244,
|
||||
"height": 100,
|
||||
"sourcePosition": "right",
|
||||
"targetPosition": "left",
|
||||
"selected": True,
|
||||
"data": {"type": "start", "title": "Start", "variables": []},
|
||||
},
|
||||
{
|
||||
"id": "iter1",
|
||||
"type": "custom",
|
||||
"position": {"x": 400, "y": 282},
|
||||
"width": 808,
|
||||
"height": 204,
|
||||
"data": {"type": "iteration", "title": "Per Item", "start_node_id": "iter1start"},
|
||||
},
|
||||
{
|
||||
"id": "iter1start",
|
||||
"type": "custom-iteration-start",
|
||||
"parentId": "iter1",
|
||||
"position": {"x": 60, "y": 78},
|
||||
"positionAbsolute": {"x": 460, "y": 360},
|
||||
"data": {"type": "iteration-start", "title": ""},
|
||||
},
|
||||
],
|
||||
"edges": [
|
||||
{
|
||||
"id": "node1-source-iter1-target",
|
||||
"source": "node1",
|
||||
"target": "iter1",
|
||||
"sourceHandle": "source",
|
||||
"targetHandle": "target",
|
||||
"type": "custom",
|
||||
"zIndex": 0,
|
||||
"data": {"sourceType": "start", "targetType": "iteration", "isInIteration": False},
|
||||
}
|
||||
],
|
||||
"viewport": {"x": 0, "y": 0, "zoom": 0.7},
|
||||
}
|
||||
|
||||
def test_drops_canvas_noise_from_top_level_nodes(self):
|
||||
compact = compact_graph_for_builder(self._graph())
|
||||
start = next(n for n in compact["nodes"] if n["id"] == "node1")
|
||||
for key in ("position", "positionAbsolute", "width", "height", "sourcePosition", "targetPosition", "selected"):
|
||||
assert key not in start
|
||||
# Semantics survive.
|
||||
assert start["data"]["type"] == "start"
|
||||
assert start["type"] == "custom"
|
||||
|
||||
def test_keeps_container_size_but_not_position(self):
|
||||
compact = compact_graph_for_builder(self._graph())
|
||||
container = next(n for n in compact["nodes"] if n["id"] == "iter1")
|
||||
assert container["width"] == 808
|
||||
assert container["height"] == 204
|
||||
assert "position" not in container
|
||||
|
||||
def test_keeps_child_relative_position(self):
|
||||
compact = compact_graph_for_builder(self._graph())
|
||||
child = next(n for n in compact["nodes"] if n["id"] == "iter1start")
|
||||
assert child["position"] == {"x": 60, "y": 78}
|
||||
assert child["parentId"] == "iter1"
|
||||
assert child["type"] == "custom-iteration-start"
|
||||
assert "positionAbsolute" not in child
|
||||
|
||||
def test_edges_keep_only_topology_fields(self):
|
||||
compact = compact_graph_for_builder(self._graph())
|
||||
assert compact["edges"] == [
|
||||
{"source": "node1", "target": "iter1", "sourceHandle": "source", "targetHandle": "target"}
|
||||
]
|
||||
|
||||
def test_viewport_is_dropped(self):
|
||||
assert "viewport" not in compact_graph_for_builder(self._graph())
|
||||
|
||||
def test_existing_graph_section_embeds_the_compact_graph(self):
|
||||
section = format_builder_existing_graph_section(self._graph())
|
||||
assert "Existing graph to refine" in section
|
||||
assert "positionAbsolute" not in section
|
||||
assert '"start_node_id":"iter1start"' in section
|
||||
|
||||
def test_existing_graph_section_empty_for_create_mode(self):
|
||||
assert format_builder_existing_graph_section(None) == ""
|
||||
def test_create_mode_renders_nothing(self):
|
||||
assert format_existing_graph_section(None) == ""
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -200,24 +200,21 @@ class TestWorkflowGeneratorService:
|
||||
call_kwargs = mock_workflow_generator.generate_workflow_graph.call_args.kwargs
|
||||
assert call_kwargs["current_graph"] is None
|
||||
|
||||
@patch("services.workflow_generator_service.LLMGenerator")
|
||||
@patch("services.workflow_generator_service.WorkflowGenerator")
|
||||
@patch("services.workflow_generator_service.ModelManager")
|
||||
@patch("services.workflow_generator_service.build_tool_catalogue")
|
||||
@patch("services.workflow_generator_service.format_tool_catalogue")
|
||||
def test_auto_mode_resolves_via_classifier(
|
||||
def test_auto_mode_forwards_sentinel_to_runner(
|
||||
self,
|
||||
mock_format_catalogue: MagicMock,
|
||||
mock_build_catalogue: MagicMock,
|
||||
mock_model_manager: MagicMock,
|
||||
mock_workflow_generator: MagicMock,
|
||||
mock_llm_generator: MagicMock,
|
||||
):
|
||||
"""Task 3: ``mode="auto"`` is classified before planning; the concrete mode reaches the runner."""
|
||||
"""``mode="auto"`` passes straight through — the planner resolves it, no extra LLM call."""
|
||||
mock_model_manager.for_tenant.return_value.get_model_instance.return_value = MagicMock()
|
||||
mock_build_catalogue.return_value = []
|
||||
mock_format_catalogue.return_value = ""
|
||||
mock_llm_generator.classify_workflow_mode.return_value = "workflow"
|
||||
mock_workflow_generator.generate_workflow_graph.return_value = {
|
||||
"graph": {"nodes": [], "edges": [], "viewport": {"x": 0, "y": 0, "zoom": 0.7}},
|
||||
"message": "",
|
||||
@ -231,26 +228,22 @@ class TestWorkflowGeneratorService:
|
||||
model_config=_model_config(),
|
||||
)
|
||||
|
||||
mock_llm_generator.classify_workflow_mode.assert_called_once()
|
||||
classify_kwargs = mock_llm_generator.classify_workflow_mode.call_args.kwargs
|
||||
assert classify_kwargs["tenant_id"] == "t-1"
|
||||
assert classify_kwargs["instruction"] == "Summarize a URL"
|
||||
assert mock_workflow_generator.generate_workflow_graph.call_args.kwargs["mode"] == "workflow"
|
||||
assert mock_workflow_generator.generate_workflow_graph.call_args.kwargs["mode"] == "auto"
|
||||
# the model registry is consulted exactly once — no classifier resolution
|
||||
mock_model_manager.for_tenant.return_value.get_model_instance.assert_called_once()
|
||||
|
||||
@patch("services.workflow_generator_service.LLMGenerator")
|
||||
@patch("services.workflow_generator_service.WorkflowGenerator")
|
||||
@patch("services.workflow_generator_service.ModelManager")
|
||||
@patch("services.workflow_generator_service.build_tool_catalogue")
|
||||
@patch("services.workflow_generator_service.format_tool_catalogue")
|
||||
def test_explicit_mode_skips_classifier(
|
||||
def test_explicit_mode_passes_through_unchanged(
|
||||
self,
|
||||
mock_format_catalogue: MagicMock,
|
||||
mock_build_catalogue: MagicMock,
|
||||
mock_model_manager: MagicMock,
|
||||
mock_workflow_generator: MagicMock,
|
||||
mock_llm_generator: MagicMock,
|
||||
):
|
||||
"""A concrete mode passes through unchanged without an extra classification call."""
|
||||
"""A concrete mode reaches the runner verbatim."""
|
||||
mock_model_manager.for_tenant.return_value.get_model_instance.return_value = MagicMock()
|
||||
mock_build_catalogue.return_value = []
|
||||
mock_format_catalogue.return_value = ""
|
||||
@ -267,7 +260,6 @@ class TestWorkflowGeneratorService:
|
||||
model_config=_model_config(),
|
||||
)
|
||||
|
||||
mock_llm_generator.classify_workflow_mode.assert_not_called()
|
||||
assert mock_workflow_generator.generate_workflow_graph.call_args.kwargs["mode"] == "advanced-chat"
|
||||
|
||||
@patch("services.workflow_generator_service.WorkflowGenerator")
|
||||
|
||||
@ -141,6 +141,7 @@ API_SENTRY_PROFILES_SAMPLE_RATE=1.0
|
||||
WEB_SENTRY_DSN=
|
||||
AMPLITUDE_API_KEY=
|
||||
TEXT_GENERATION_TIMEOUT_MS=60000
|
||||
WORKFLOW_GENERATION_TIMEOUT_MS=180000
|
||||
CSP_WHITELIST=
|
||||
ALLOW_EMBED=false
|
||||
ALLOW_UNSAFE_DATA_SCHEME=false
|
||||
|
||||
@ -391,6 +391,7 @@ services:
|
||||
NEXT_TELEMETRY_DISABLED: ${NEXT_TELEMETRY_DISABLED:-0}
|
||||
EXPERIMENTAL_ENABLE_VINEXT: ${EXPERIMENTAL_ENABLE_VINEXT:-false}
|
||||
TEXT_GENERATION_TIMEOUT_MS: ${TEXT_GENERATION_TIMEOUT_MS:-60000}
|
||||
WORKFLOW_GENERATION_TIMEOUT_MS: ${WORKFLOW_GENERATION_TIMEOUT_MS:-180000}
|
||||
CSP_WHITELIST: ${CSP_WHITELIST:-}
|
||||
ALLOW_EMBED: ${ALLOW_EMBED:-false}
|
||||
ALLOW_UNSAFE_DATA_SCHEME: ${ALLOW_UNSAFE_DATA_SCHEME:-false}
|
||||
|
||||
@ -397,6 +397,7 @@ services:
|
||||
NEXT_TELEMETRY_DISABLED: ${NEXT_TELEMETRY_DISABLED:-0}
|
||||
EXPERIMENTAL_ENABLE_VINEXT: ${EXPERIMENTAL_ENABLE_VINEXT:-false}
|
||||
TEXT_GENERATION_TIMEOUT_MS: ${TEXT_GENERATION_TIMEOUT_MS:-60000}
|
||||
WORKFLOW_GENERATION_TIMEOUT_MS: ${WORKFLOW_GENERATION_TIMEOUT_MS:-180000}
|
||||
CSP_WHITELIST: ${CSP_WHITELIST:-}
|
||||
ALLOW_EMBED: ${ALLOW_EMBED:-false}
|
||||
ALLOW_UNSAFE_DATA_SCHEME: ${ALLOW_UNSAFE_DATA_SCHEME:-false}
|
||||
|
||||
@ -181,6 +181,8 @@ WORKFLOW_MAX_EXECUTION_STEPS=500
|
||||
WORKFLOW_MAX_EXECUTION_TIME=1200
|
||||
WORKFLOW_CALL_MAX_DEPTH=5
|
||||
MAX_VARIABLE_SIZE=204800
|
||||
WORKFLOW_GENERATOR_NODE_BUILDER_MAX_WORKERS=6
|
||||
WORKFLOW_GENERATION_TIMEOUT_MS=180000
|
||||
WORKFLOW_FILE_UPLOAD_LIMIT=10
|
||||
GRAPH_ENGINE_MIN_WORKERS=3
|
||||
GRAPH_ENGINE_MAX_WORKERS=10
|
||||
|
||||
@ -47,6 +47,11 @@ NEXT_PUBLIC_TEXT_GENERATION_TIMEOUT_MS=60000
|
||||
# Used by web/docker/entrypoint.sh to overwrite/export NEXT_PUBLIC_TEXT_GENERATION_TIMEOUT_MS at container startup (Docker only)
|
||||
TEXT_GENERATION_TIMEOUT_MS=60000
|
||||
|
||||
# The timeout for the cmd+k workflow generation in millisecond
|
||||
NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS=180000
|
||||
# Used by web/docker/entrypoint.sh to overwrite/export NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS at container startup (Docker only)
|
||||
WORKFLOW_GENERATION_TIMEOUT_MS=180000
|
||||
|
||||
# CSP https://developer.mozilla.org/en-US/docs/Web/HTTP/CSP
|
||||
NEXT_PUBLIC_CSP_WHITELIST=
|
||||
# Default is not allow to embed into iframe to prevent Clickjacking: https://owasp.org/www-community/attacks/Clickjacking
|
||||
|
||||
@ -34,6 +34,7 @@ import { ModelTypeEnum } from '@/app/components/header/account-setting/model-pro
|
||||
import { useModelListAndDefaultModelAndCurrentProviderAndModel } from '@/app/components/header/account-setting/model-provider-page/hooks'
|
||||
import ModelParameterModal from '@/app/components/header/account-setting/model-provider-page/model-parameter-modal'
|
||||
import WorkflowPreview from '@/app/components/workflow/workflow-preview'
|
||||
import { WORKFLOW_GENERATION_TIMEOUT_MS } from '@/config'
|
||||
import { systemFeaturesQueryOptions } from '@/features/system-features/client'
|
||||
import { useRouter } from '@/next/navigation'
|
||||
import { fetchWorkflowDraft } from '@/service/workflow'
|
||||
@ -56,12 +57,6 @@ import {
|
||||
import { useWorkflowGeneratorStore } from './store'
|
||||
import useGenGraph from './use-gen-graph'
|
||||
|
||||
// Hard ceiling before we abort a hung request. Generous on purpose: the
|
||||
// backend runs two sequential LLM calls and may retry a transient provider
|
||||
// error (bounded backoff) or an unparseable response (one extra call), so a
|
||||
// slow-but-succeeding generation can legitimately pass the one-minute mark.
|
||||
// Aborting work that would have landed is the worse failure mode.
|
||||
const FE_TIMEOUT_MS = 90_000
|
||||
// Mirrors the backend's instruction/ideal-output cap on /workflow-generate —
|
||||
// keeping the limit client-side turns an opaque 400 into a visible input stop.
|
||||
const MAX_INSTRUCTION_LENGTH = 10_000
|
||||
@ -251,7 +246,7 @@ function WorkflowGeneratorModal() {
|
||||
|
||||
// Holds the AbortController of the in-flight ``/workflow-generate`` request
|
||||
// so we can cancel it on (a) modal close, (b) a second Generate click
|
||||
// while loading, (c) the hard 60 s frontend timeout, or (d) the user
|
||||
// while loading, (c) the hard frontend timeout, or (d) the user
|
||||
// pressing Cancel. Without this an in-flight request outlives the modal
|
||||
// and can race a future Generate call.
|
||||
const abortRef = useRef<AbortController | null>(null)
|
||||
@ -349,13 +344,17 @@ function WorkflowGeneratorModal() {
|
||||
setLoadingTrue()
|
||||
|
||||
// Hard frontend timeout — aborts the request and surfaces a localised toast
|
||||
// instead of a perpetual spinner if the backend hangs.
|
||||
// instead of a perpetual spinner if the backend hangs. Generous on purpose
|
||||
// (NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS, default 180s): the backend
|
||||
// runs a planner call plus parallel builder calls and may retry transient
|
||||
// errors, so aborting a slow-but-succeeding generation is the worse
|
||||
// failure mode.
|
||||
timeoutRef.current = setTimeout(() => {
|
||||
abortRef.current?.abort()
|
||||
abortRef.current = null
|
||||
toast.error(t(($) => $['workflowGenerator.errors.timeout']))
|
||||
setLoadingFalse()
|
||||
}, FE_TIMEOUT_MS)
|
||||
}, WORKFLOW_GENERATION_TIMEOUT_MS)
|
||||
|
||||
// Refine mode: pull the current draft so the backend amends it instead of
|
||||
// starting from scratch. The modal mounts outside the Studio's ReactFlow
|
||||
|
||||
@ -269,6 +269,7 @@ export const JSON_SCHEMA_MAX_DEPTH = 10
|
||||
export const MAX_TOOLS_NUM = env.NEXT_PUBLIC_MAX_TOOLS_NUM
|
||||
export const MAX_PARALLEL_LIMIT = env.NEXT_PUBLIC_MAX_PARALLEL_LIMIT
|
||||
export const TEXT_GENERATION_TIMEOUT_MS = env.NEXT_PUBLIC_TEXT_GENERATION_TIMEOUT_MS
|
||||
export const WORKFLOW_GENERATION_TIMEOUT_MS = env.NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS
|
||||
export const LOOP_NODE_MAX_COUNT = env.NEXT_PUBLIC_LOOP_NODE_MAX_COUNT
|
||||
export const MAX_ITERATIONS_NUM = env.NEXT_PUBLIC_MAX_ITERATIONS_NUM
|
||||
export const MAX_TREE_DEPTH = env.NEXT_PUBLIC_MAX_TREE_DEPTH
|
||||
|
||||
@ -46,6 +46,7 @@ export NEXT_PUBLIC_ENABLE_LEARN_APP=${NEXT_PUBLIC_ENABLE_LEARN_APP:-${ENABLE_LEA
|
||||
export NEXT_PUBLIC_RBAC_ENABLED=${NEXT_PUBLIC_RBAC_ENABLED:-${RBAC_ENABLED}}
|
||||
|
||||
export NEXT_PUBLIC_TEXT_GENERATION_TIMEOUT_MS=${TEXT_GENERATION_TIMEOUT_MS}
|
||||
export NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS=${NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS:-${WORKFLOW_GENERATION_TIMEOUT_MS}}
|
||||
export NEXT_PUBLIC_CSP_WHITELIST=${CSP_WHITELIST}
|
||||
export NEXT_PUBLIC_ALLOW_EMBED=${ALLOW_EMBED}
|
||||
export NEXT_PUBLIC_ALLOW_INLINE_STYLES=${ALLOW_INLINE_STYLES:-false}
|
||||
|
||||
11
web/env.ts
11
web/env.ts
@ -164,6 +164,10 @@ const clientSchema = {
|
||||
*/
|
||||
NEXT_PUBLIC_UPLOAD_IMAGE_AS_ICON: coercedBoolean.default(false),
|
||||
NEXT_PUBLIC_WEB_PREFIX: z.url().optional(),
|
||||
/**
|
||||
* The timeout for the cmd+k workflow generation in millisecond
|
||||
*/
|
||||
NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS: coercedNumber.default(180000),
|
||||
NEXT_PUBLIC_ZENDESK_FIELD_ID_EMAIL: z.string().optional(),
|
||||
NEXT_PUBLIC_ZENDESK_FIELD_ID_ENVIRONMENT: z.string().optional(),
|
||||
NEXT_PUBLIC_ZENDESK_FIELD_ID_PLAN: z.string().optional(),
|
||||
@ -189,6 +193,10 @@ export const env = createEnv({
|
||||
* The timeout for the text generation in millisecond
|
||||
*/
|
||||
TEXT_GENERATION_TIMEOUT_MS: coercedNumber.default(60000),
|
||||
/**
|
||||
* The timeout for the cmd+k workflow generation in millisecond
|
||||
*/
|
||||
WORKFLOW_GENERATION_TIMEOUT_MS: coercedNumber.default(180000),
|
||||
},
|
||||
client: clientSchema,
|
||||
experimental__runtimeEnv: {
|
||||
@ -354,6 +362,9 @@ export const env = createEnv({
|
||||
NEXT_PUBLIC_WEB_PREFIX: isServer
|
||||
? process.env.NEXT_PUBLIC_WEB_PREFIX
|
||||
: getRuntimeEnvFromBody('webPrefix'),
|
||||
NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS: isServer
|
||||
? process.env.NEXT_PUBLIC_WORKFLOW_GENERATION_TIMEOUT_MS
|
||||
: getRuntimeEnvFromBody('workflowGenerationTimeoutMs'),
|
||||
NEXT_PUBLIC_ZENDESK_FIELD_ID_EMAIL: isServer
|
||||
? process.env.NEXT_PUBLIC_ZENDESK_FIELD_ID_EMAIL
|
||||
: getRuntimeEnvFromBody('zendeskFieldIdEmail'),
|
||||
|
||||
Loading…
Reference in New Issue
Block a user