opensource/dify
3
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-24 23:22:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: Improve workflow generator accessibility and API contracts (#38838)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Crazywoola
2026-07-13 09:15:54 +00:00
committed by GitHub
co-authored by autofix-ci[bot]
parent 8e0d10bf5e
commit 7865ffd429
27 changed files with 1869 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/core/workflow/generator/prompts/builder_prompts.py
+20 -1
View File
@@ -440,7 +440,26 @@ correct.
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.
``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.
"""
api/core/workflow/generator/prompts/planner_prompts.py
+5
View File
@@ -51,6 +51,11 @@ minimum set of Dify workflow nodes needed to fulfil it, in execution order.
"question-classifier" (semantic / intent routing)
- "for each item in a list" → "iteration"
- "keep going until condition" → "loop"
- synthesize multiple independent knowledge sources → one
"knowledge-retrieval" node per source, then one "template-transform"
node that combines every result, then one "llm" node that consumes the
template output as context; the retrievals are parallel inputs to the
template, not mutually exclusive branches
5. PREFER "tool" over "http-request" or "code" whenever an installed tool from the
"Available tools" section below covers the task (e.g. web search, time lookup,
scraping, audio, translation, etc.). Only fall back to "http-request" for
api/core/workflow/generator/runner.py
+393 -31
View File
@@ -15,7 +15,7 @@ Pipeline:
Intentionally NOT here (deferred to a future iteration):
- Mermaid rendering
- Heuristic node/edge auto-repair beyond default fill
- Broad semantic auto-repair when multiple valid graph interpretations exist
- Multi-step validation engine with classification of fixable vs. user-required errors
- Tool / model catalogue filtering
@@ -419,6 +419,11 @@ class WorkflowGenerator:
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)
# Planner-supplied user-input declarations. The builder uses these to
# populate ``start.data.variables`` so downstream ``{#start.<var>#}``
# references resolve at run time. Optional field — older prompts may
@@ -650,6 +655,38 @@ class WorkflowGenerator:
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
# ------------------------------------------------------------------
@@ -719,7 +756,7 @@ class WorkflowGenerator:
# ------------------------------------------------------------------
@classmethod
def _postprocess_graph(cls, *, graph: GraphDict, mode: WorkflowGenerationMode) -> GraphDict:
"""Fill safe defaults, normalise positions and dedupe edges."""
"""Fill safe defaults and apply only deterministic graph repairs."""
# Internally treat nodes/edges as untyped dicts — TypedDicts forbid the
# arbitrary-key setdefault writes we need here, but the caller only sees
@@ -737,6 +774,12 @@ class WorkflowGenerator:
# of the postprocess pass touches them.
cls._sanitize_node_ids(nodes=nodes, edges=edges)
# An LLM context accepts one selector. If the builder wires multiple
# retrieval nodes straight into an LLM but selects only one result,
# insert a template-transform fan-in that renders every result into
# one string and point the context at that output.
cls._insert_multi_retrieval_context_templates(nodes=nodes, edges=edges)
# Container-child nodes carry their own relative positions inside the
# parent and have a special ``type`` (custom-iteration-start /
# custom-loop-start). We must not override their positions or wrapper
@@ -835,7 +878,10 @@ class WorkflowGenerator:
# fails at run time with "variable not found". The dominant failure
# mode is a prompt that references ``{#start.url#}`` when the start
# node has ``variables: []``, so we auto-inject missing start-node
# variables before we surface them as errors.
# variables. We also repair a mistaken output name when its source
# exposes exactly one valid output; ambiguous references still fail
# closed in the structural validator.
cls._normalize_sys_query_references(nodes=nodes, mode=mode)
cls._reconcile_variable_references(nodes=nodes, mode=mode)
# Schema backstop: a "file" / "file-list" start variable MUST carry a
@@ -851,6 +897,104 @@ class WorkflowGenerator:
# Variable-reference reconciliation
# ------------------------------------------------------------------
@classmethod
def _normalize_sys_query_references(
cls,
*,
nodes: list[dict[str, Any]],
mode: WorkflowGenerationMode,
) -> None:
"""Normalize malformed ``sys.query`` references without changing their intent.
Chatflows expose ``sys.query`` directly. Plain workflows do not, so
their query references become a Start-node input and the existing
reconciliation pass declares that input immediately afterwards.
"""
if mode == "advanced-chat":
target_node_id = "sys"
else:
start_node = next(
(node for node in nodes if node.get("data", {}).get("type") == BuiltinNodeTypes.START),
None,
)
start_node_id = start_node.get("id") if start_node else None
if not isinstance(start_node_id, str) or not start_node_id:
return
target_node_id = start_node_id
for node in nodes:
data = node.get("data")
if isinstance(data, dict):
cls._normalize_sys_query_reference_in_data(data, target_node_id=target_node_id)
@classmethod
def _normalize_sys_query_reference_in_data(
cls,
value: Any,
*,
target_node_id: str,
allow_selector: bool = True,
) -> Any:
"""Rewrite query placeholders and selectors at any node-data depth.
Some node schemas store selectors inside another list, for example a
parameter extractor's ``query`` or a variable aggregator's
``variables``. Literal string-list fields opt out so an option list
such as ``["sys", "query"]`` is preserved.
"""
target_placeholder = f"{{{{#{target_node_id}.query#}}}}"
if isinstance(value, str):
return value.replace("{{#sys.query#}}", target_placeholder).replace("{{#sys,query#}}", target_placeholder)
if isinstance(value, dict):
for key, item in list(value.items()):
item_allows_selector = allow_selector and key not in cls._NON_SELECTOR_LIST_KEYS
if item_allows_selector and cls._is_sys_query_selector(item):
value[key] = [target_node_id, "query"]
continue
if (
item_allows_selector
and cls._is_selector_field(key)
and isinstance(item, str)
and cls._is_sys_query_token(item)
):
value[key] = [target_node_id, "query"]
continue
value[key] = cls._normalize_sys_query_reference_in_data(
item,
target_node_id=target_node_id,
allow_selector=item_allows_selector,
)
return value
if isinstance(value, list):
if allow_selector and cls._is_sys_query_selector(value):
return [target_node_id, "query"]
for index, item in enumerate(value):
value[index] = cls._normalize_sys_query_reference_in_data(
item,
target_node_id=target_node_id,
allow_selector=allow_selector,
)
return value
@staticmethod
def _is_sys_query_selector(value: Any) -> bool:
"""Recognize the valid selector and common one-item LLM variants."""
if value == ["sys", "query"]:
return True
if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], str):
return False
return WorkflowGenerator._is_sys_query_token(value[0])
@staticmethod
def _is_sys_query_token(value: str) -> bool:
"""Return whether a string is a compact malformed query selector."""
return value.replace(" ", "") in {"sys.query", "sys,query"}
@staticmethod
def _is_selector_field(key: str) -> bool:
"""Return whether a data field explicitly stores one selector."""
return key == "selector" or key.endswith("_selector")
# Detects ``{{#node_id.var#}}`` placeholders. We match the EXACT regex
# Dify's workflow runtime uses (see
# ``graphon.runtime.variable_pool.VARIABLE_PATTERN``):
@@ -903,9 +1047,13 @@ class WorkflowGenerator:
@classmethod
def _reconcile_variable_references(cls, *, nodes: list[dict[str, Any]], mode: WorkflowGenerationMode) -> None:
"""
Walk every variable reference, ensure it resolves; auto-fix missing
start-node variables (the safe, dominant case) by adding a stub
``paragraph`` entry to ``start.data.variables``.
Apply deterministic repairs to unresolved variable references.
Missing start-node inputs are added as ``paragraph`` variables. For
non-start nodes, a mistaken output name is rewritten only when the
source exposes exactly one declared output. Sources with zero or
multiple outputs remain untouched so validation fails closed instead
of guessing which value the workflow should consume.
For Advanced-Chat mode, ``sys.query`` and ``sys.files`` are always
treated as resolved without any declaration. Tool nodes' parameter
@@ -934,16 +1082,83 @@ class WorkflowGenerator:
continue
if cls._declares_variable(target, var):
continue
# Missing variable. Auto-fix start-node references; let everything
# else fall through and surface in the result's ``error`` field
# via the post-postprocess validator below.
if start_node is not None and target is start_node:
cls._inject_start_variable(start_node, var)
logger.info("Workflow generator: auto-injected missing start variable %r", var)
continue
replacement = cls._sole_declared_variable(target)
if replacement is None:
continue
for node in nodes:
data = node.get("data")
if isinstance(data, dict):
cls._rewrite_variable_reference_in_data(
data,
node_id=node_id,
old_variable=var,
new_variable=replacement,
)
logger.info(
"Workflow generator: rewrote unresolved reference %s.%s to sole output %s.%s",
node_id,
var,
node_id,
replacement,
)
@classmethod
def _collect_refs_in_data(cls, value: Any, out: set[tuple[str, str]]) -> None:
"""Recursively walk a node's ``data`` and harvest every reference."""
def _rewrite_variable_reference_in_data(
cls,
value: Any,
*,
node_id: str,
old_variable: str,
new_variable: str,
allow_selector: bool = True,
) -> Any:
"""Rewrite one exact placeholder or selector at any data depth."""
if isinstance(value, str):
return cls._VAR_REF_RE.sub(
lambda match: (
f"{{{{#{node_id}.{new_variable}#}}}}"
if match.group(1) == node_id and match.group(2) == old_variable
else match.group(0)
),
value,
)
if isinstance(value, dict):
for key, item in list(value.items()):
value[key] = cls._rewrite_variable_reference_in_data(
item,
node_id=node_id,
old_variable=old_variable,
new_variable=new_variable,
allow_selector=allow_selector and key not in cls._NON_SELECTOR_LIST_KEYS,
)
return value
if isinstance(value, list):
if allow_selector and value == [node_id, old_variable]:
return [node_id, new_variable]
for index, item in enumerate(value):
value[index] = cls._rewrite_variable_reference_in_data(
item,
node_id=node_id,
old_variable=old_variable,
new_variable=new_variable,
allow_selector=allow_selector,
)
return value
@classmethod
def _collect_refs_in_data(
cls,
value: Any,
out: set[tuple[str, str]],
*,
allow_selector: bool = True,
) -> None:
"""Recursively harvest placeholders and selectors at any data depth."""
if isinstance(value, str):
for match in cls._VAR_REF_RE.finditer(value):
node_id, var = match.group(1).strip(), match.group(2).strip()
@@ -951,28 +1166,20 @@ class WorkflowGenerator:
out.add((node_id, var))
return
if isinstance(value, dict):
# Known selector shapes: 2-element [node_id, var] lists.
for k, v in value.items():
# ``value_selector`` / ``query_variable_selector`` / etc.: a
# flat 2-element list of strings. Skip keys whose value is a
# plain string list that merely HAPPENS to have two entries —
# a 2-option ``select`` or a file variable's two allowed upload
# methods are NOT ``[node_id, var]`` selectors and must not be
# mistaken for references.
if (
isinstance(v, list)
and len(v) == 2
and all(isinstance(x, str) for x in v)
and k not in cls._NON_SELECTOR_LIST_KEYS
):
node_id, var = v[0].strip(), v[1].strip()
if node_id and var:
out.add((node_id, var))
cls._collect_refs_in_data(v, out)
cls._collect_refs_in_data(
v,
out,
allow_selector=allow_selector and k not in cls._NON_SELECTOR_LIST_KEYS,
)
return
if isinstance(value, list):
if allow_selector and len(value) == 2 and all(isinstance(item, str) for item in value):
node_id, var = value[0].strip(), value[1].strip()
if node_id and var:
out.add((node_id, var))
for item in value:
cls._collect_refs_in_data(item, out)
cls._collect_refs_in_data(item, out, allow_selector=allow_selector)
@classmethod
def _declares_variable(cls, node: dict[str, Any], var: str) -> bool:
@@ -1022,6 +1229,37 @@ class WorkflowGenerator:
# produce outputs of their own.
return False
@classmethod
def _sole_declared_variable(cls, node: dict[str, Any]) -> str | None:
"""Return the only output exposed by ``node``, or ``None`` when ambiguous."""
data = node.get("data") or {}
node_type = data.get("type")
if node_type == BuiltinNodeTypes.LLM:
schema = ((data.get("structured_output") or {}).get("schema") or {}).get("properties") or {}
return "text" if not schema else None
if node_type == BuiltinNodeTypes.CODE:
outputs = [key for key in (data.get("outputs") or {}) if isinstance(key, str)]
return outputs[0] if len(outputs) == 1 else None
if node_type == BuiltinNodeTypes.PARAMETER_EXTRACTOR:
parameters = [
parameter.get("name")
for parameter in (data.get("parameters") or [])
if isinstance(parameter, dict) and isinstance(parameter.get("name"), str)
]
return parameters[0] if len(parameters) == 1 else None
if not isinstance(node_type, str):
return None
single_output_by_type: dict[str, str] = {
BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL: "result",
BuiltinNodeTypes.TEMPLATE_TRANSFORM: "output",
BuiltinNodeTypes.ITERATION: "output",
BuiltinNodeTypes.LOOP: "output",
BuiltinNodeTypes.DOCUMENT_EXTRACTOR: "text",
BuiltinNodeTypes.VARIABLE_AGGREGATOR: "output",
BuiltinNodeTypes.LEGACY_VARIABLE_AGGREGATOR: "output",
}
return single_output_by_type.get(node_type)
@classmethod
def _sanitize_node_ids(cls, *, nodes: list[dict[str, Any]], edges: list[dict[str, Any]]) -> None:
"""
@@ -1137,6 +1375,129 @@ class WorkflowGenerator:
new_id = id_map.get(node_id, node_id)
return f"{{{{#{new_id}.{rest}#}}}}"
@classmethod
def _insert_multi_retrieval_context_templates(
cls,
*,
nodes: list[dict[str, Any]],
edges: list[dict[str, Any]],
) -> None:
"""Fan multiple retrieval inputs into an LLM through one template.
The repair is intentionally narrow: it applies only when at least two
knowledge-retrieval nodes have direct edges into the same LLM and that
LLM currently uses one of those results as its enabled context. This
is enough to repair the builder's lossy single-context output without
guessing about unrelated retrievals or mutually exclusive branches.
"""
nodes_by_id: dict[str, dict[str, Any]] = {
node_id: node for node in nodes if isinstance(node_id := node.get("id"), str)
}
used_ids = set(nodes_by_id)
llm_nodes = [node for node in nodes if node.get("data", {}).get("type") == BuiltinNodeTypes.LLM]
for llm_node in llm_nodes:
llm_id = llm_node.get("id")
if not isinstance(llm_id, str):
continue
incoming_retrieval_edges: list[dict[str, Any]] = []
for edge in edges:
source_id = edge.get("source")
if edge.get("target") != llm_id or not isinstance(source_id, str):
continue
source_node = nodes_by_id.get(source_id)
if source_node and source_node.get("data", {}).get("type") == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
incoming_retrieval_edges.append(edge)
retrieval_ids = list(
dict.fromkeys(
edge["source"] for edge in incoming_retrieval_edges if isinstance(edge.get("source"), str)
)
)
if len(retrieval_ids) < 2:
continue
llm_data = llm_node.get("data")
if not isinstance(llm_data, dict):
continue
context = llm_data.get("context")
if not isinstance(context, dict) or not context.get("enabled"):
continue
selector = context.get("variable_selector")
if selector not in [[retrieval_id, "result"] for retrieval_id in retrieval_ids]:
continue
template_id = cls._next_generated_node_id(prefix="retrieval_context", used_ids=used_ids)
variables = [
{"variable": f"knowledge_{index}", "value_selector": [retrieval_id, "result"]}
for index, retrieval_id in enumerate(retrieval_ids, start=1)
]
sections = [
(
f"## Knowledge source {index}\n"
f"{{% for item in knowledge_{index} %}}{{{{ item.content }}}}\n{{% endfor %}}"
)
for index in range(1, len(retrieval_ids) + 1)
]
nodes.append(
{
"id": template_id,
"type": "custom",
"position": {"x": 0, "y": 0},
"data": {
"type": BuiltinNodeTypes.TEMPLATE_TRANSFORM,
"title": "Combine Knowledge",
"variables": variables,
"template": "\n\n".join(sections),
},
}
)
incoming_edge_objects = {id(edge) for edge in incoming_retrieval_edges}
edges[:] = [edge for edge in edges if id(edge) not in incoming_edge_objects]
edges.extend(
{"source": retrieval_id, "target": template_id, "type": "custom"} for retrieval_id in retrieval_ids
)
edges.append({"source": template_id, "target": llm_id, "type": "custom"})
context["variable_selector"] = [template_id, "output"]
cls._ensure_llm_context_placeholder(llm_data)
logger.info(
"Workflow generator: inserted template %s to combine retrieval inputs for LLM %s",
template_id,
llm_id,
)
@staticmethod
def _next_generated_node_id(*, prefix: str, used_ids: set[str]) -> str:
"""Return a short runtime-safe node id and reserve it in ``used_ids``."""
suffix = 1
candidate = prefix
while candidate in used_ids:
suffix += 1
candidate = f"{prefix}_{suffix}"
used_ids.add(candidate)
return candidate
@staticmethod
def _ensure_llm_context_placeholder(llm_data: dict[str, Any]) -> None:
"""Ensure an enabled LLM context is actually present in its prompt."""
prompt_template = llm_data.get("prompt_template")
if isinstance(prompt_template, list):
messages = [message for message in prompt_template if isinstance(message, dict)]
if any("{{#context#}}" in str(message.get("text") or "") for message in messages):
return
target = next((message for message in reversed(messages) if message.get("role") == "user"), None)
if target is None:
prompt_template.append({"role": "user", "text": "{{#context#}}"})
return
target["text"] = f"{target.get('text') or ''}\n\n{{{{#context#}}}}"
return
if isinstance(prompt_template, dict):
text = str(prompt_template.get("text") or "")
if "{{#context#}}" not in text:
prompt_template["text"] = f"{text}\n\n{{{{#context#}}}}"
@classmethod
def _repair_branch_edge_handles(cls, *, nodes: list[dict[str, Any]], edges: list[dict[str, Any]]) -> None:
"""
@@ -1690,8 +2051,9 @@ class WorkflowGenerator:
"""
Walk every variable reference and flag anything pointing at a node
that doesn't declare it. The postprocess step has already
auto-injected missing start-node variables, so by the time this
runs only NON-start references should ever fail.
auto-injected missing start-node variables and repaired references to
sole outputs, so by the time this runs only ambiguous or impossible
references should fail.
"""
out: list[WorkflowGenerateErrorDict] = []
by_id: dict[str, dict[str, Any]] = {n.get("id", ""): n for n in nodes if n.get("id")}
api/core/workflow/generator/types.py
+18 -17
View File
@@ -7,7 +7,8 @@ required to return after ``json_repair`` parsing. They mirror the runtime
can be written straight into a draft workflow without further translation.
"""
from typing import Final, Literal, NotRequired, TypedDict
from enum import StrEnum
from typing import Literal, NotRequired, TypedDict
WorkflowGenerationMode = Literal["workflow", "advanced-chat"]
@@ -22,22 +23,22 @@ WorkflowGenerationModeRequest = Literal["workflow", "advanced-chat", "auto"]
# Machine-readable error codes returned in ``WorkflowGenerateResultDict.errors``.
# Frontend maps these to localised copy via ``workflow.generator.errors.<code>``
# i18n keys, so any change here MUST be mirrored in the FE i18n map.
class WorkflowGenerateErrorCode:
INVALID_JSON: Final = "INVALID_JSON"
INVALID_SCHEMA: Final = "INVALID_SCHEMA"
EMPTY_INSTRUCTION: Final = "EMPTY_INSTRUCTION"
INSTRUCTION_TOO_LONG: Final = "INSTRUCTION_TOO_LONG"
DUPLICATE_NODE_ID: Final = "DUPLICATE_NODE_ID"
GRAPH_CYCLE: Final = "GRAPH_CYCLE"
EMPTY_PLAN: Final = "EMPTY_PLAN"
UNKNOWN_NODE_REFERENCE: Final = "UNKNOWN_NODE_REFERENCE"
INVALID_CONTAINER: Final = "INVALID_CONTAINER"
UNRESOLVED_REFERENCE: Final = "UNRESOLVED_REFERENCE"
UNKNOWN_TOOL: Final = "UNKNOWN_TOOL"
MISSING_TERMINAL: Final = "MISSING_TERMINAL"
MISSING_START: Final = "MISSING_START"
DANGLING_EDGE: Final = "DANGLING_EDGE"
MODEL_ERROR: Final = "MODEL_ERROR"
class WorkflowGenerateErrorCode(StrEnum):
INVALID_JSON = "INVALID_JSON"
INVALID_SCHEMA = "INVALID_SCHEMA"
EMPTY_INSTRUCTION = "EMPTY_INSTRUCTION"
INSTRUCTION_TOO_LONG = "INSTRUCTION_TOO_LONG"
DUPLICATE_NODE_ID = "DUPLICATE_NODE_ID"
GRAPH_CYCLE = "GRAPH_CYCLE"
EMPTY_PLAN = "EMPTY_PLAN"
UNKNOWN_NODE_REFERENCE = "UNKNOWN_NODE_REFERENCE"
INVALID_CONTAINER = "INVALID_CONTAINER"
UNRESOLVED_REFERENCE = "UNRESOLVED_REFERENCE"
UNKNOWN_TOOL = "UNKNOWN_TOOL"
MISSING_TERMINAL = "MISSING_TERMINAL"
MISSING_START = "MISSING_START"
DANGLING_EDGE = "DANGLING_EDGE"
MODEL_ERROR = "MODEL_ERROR"
class WorkflowGenerateErrorDict(TypedDict):