diff --git a/application/agents/workflows/cel_evaluator.py b/application/agents/workflows/cel_evaluator.py index 5f8bab37..2a7a1748 100644 --- a/application/agents/workflows/cel_evaluator.py +++ b/application/agents/workflows/cel_evaluator.py @@ -1,3 +1,4 @@ +import re from typing import Any, Dict import celpy @@ -32,19 +33,105 @@ def build_activation(state: Dict[str, Any]) -> Dict[str, Any]: return {k: _convert_value(v) for k, v in state.items()} +_MAX_ERROR_CHARS = 200 + + +def _summarize_cel_error(exc: Exception) -> str: + """Reduce a celpy error to a short, value-free description. + + Two problems with the raw text. ``undeclared reference`` errors embed a + repr of the *entire* activation — every binding, thousands of characters, + including the user's own query. Others interpolate the offending **value** + (``invalid literal for int() with base 10: StringType('…')``). Both get + logged and, now that config errors skip ``sanitize_api_error``, shown. So + strip the activation dump, drop anything that looks like a quoted value, + and cap the result. + """ + text = " ".join(str(exc).split()) + marker = " (in activation" + if marker in text: + text = text.split(marker, 1)[0] + text = text.strip().lstrip("(").strip() + # Quoted fragments in celpy messages are either the offending identifier + # (safe, and the useful part) or a state value (not ours to echo). Keep + # bare identifiers, redact anything else. + def _redact(match: "re.Match[str]") -> str: + inner = match.group(1) + return f"'{inner}'" if inner.isidentifier() else "'…'" + + text = re.sub(r"'([^']*)'", _redact, text) + text = text.strip().strip('"').strip() + if len(text) > _MAX_ERROR_CHARS: + text = text[: _MAX_ERROR_CHARS - 1].rstrip() + "…" + return text + + +def _looks_like_template_syntax(expression: str) -> bool: + """Whether the source contains a ``{{ … }}`` substitution. + + Only consulted **after** compilation has already failed: ``{{`` is legal + CEL inside a string literal (``x == "{{y}}"``) and at the head of a nested + map literal (``{{"a": 1}: 2}``), both of which compile fine and must not be + rejected. + """ + return "{{" in expression and "}}" in expression + + +def _template_syntax_error(expression: str) -> "CelEvaluationError": + inner = expression.split("{{", 1)[1].split("}}", 1)[0].strip() + suggestion = inner or "query" + return CelEvaluationError( + "This field takes a CEL expression, not {{ }} template syntax. " + f"Write {suggestion} instead of {{{{{suggestion}}}}}." + ) + + +def _compile(expression: str): + """Compile, translating failures into actionable messages.""" + try: + return celpy.Environment().compile(expression) + except Exception as exc: + # Agent ``prompt_template`` and end ``output_template`` fields *are* + # Jinja2, so ``{{query}}`` is correct one panel over. Here it fails to + # parse with a bare caret, which reads as "the product is broken" + # rather than "wrong syntax for this field". + if _looks_like_template_syntax(expression): + raise _template_syntax_error(expression) from exc + raise CelEvaluationError(f"CEL error: {_summarize_cel_error(exc)}") from exc + + +def validate_cel_expression(expression: str) -> None: + """Compile-check an expression without running it. + + Used at save time. Only syntax is knowable then — workflow state is built + during a run, so unresolved names are *not* an error here. + + Args: + expression: The CEL source to check. + + Raises: + CelEvaluationError: If the expression is empty or does not compile. + """ + if not expression or not expression.strip(): + raise CelEvaluationError("Empty expression") + _compile(expression) + + def evaluate_cel(expression: str, state: Dict[str, Any]) -> Any: if not expression or not expression.strip(): raise CelEvaluationError("Empty expression") + ast = _compile(expression) try: env = celpy.Environment() - ast = env.compile(expression) program = env.program(ast) activation = build_activation(state) result = program.evaluate(activation) except celpy.CELEvalError as exc: - raise CelEvaluationError(f"CEL evaluation error: {exc}") from exc + raise CelEvaluationError( + f"CEL evaluation error: {_summarize_cel_error(exc)}" + ) from exc except Exception as exc: - raise CelEvaluationError(f"CEL error: {exc}") from exc + raise CelEvaluationError(f"CEL error: {_summarize_cel_error(exc)}") from exc return cel_to_python(result) diff --git a/application/agents/workflows/workflow_engine.py b/application/agents/workflows/workflow_engine.py index 0c4f39a3..b4b5c145 100644 --- a/application/agents/workflows/workflow_engine.py +++ b/application/agents/workflows/workflow_engine.py @@ -165,7 +165,19 @@ class WorkflowEngine: self._finalize_log_entry(log_entry, pre_state) self.execution_log.append(log_entry) - user_friendly_error = sanitize_api_error(e) + # A misconfigured expression is the builder's own typo, and the + # message says exactly how to fix it. Routing it through + # sanitize_api_error would match the ``{`` in ``{{query}}`` and + # replace the whole thing with "An error occurred … try again + # later" — advice that sends the user round the same loop. + # Everything else stays sanitized: node internals and upstream + # provider errors are not for end users. + is_config_error = isinstance(e, CelEvaluationError) + user_friendly_error = ( + f"{node.title or node.type.value}: {e}" + if is_config_error + else sanitize_api_error(e) + ) yield { "type": "workflow_step", "node_id": node.id, @@ -175,7 +187,11 @@ class WorkflowEngine: "state_delta": log_entry["state_delta"], "error": user_friendly_error, } - yield {"type": "error", "error": user_friendly_error} + yield { + "type": "error", + "error": user_friendly_error, + "user_facing": is_config_error, + } break self.execution_log.append(log_entry) pre_state = dict(self.state) diff --git a/application/api/user/workflows/routes.py b/application/api/user/workflows/routes.py index 47a9798a..c5e92832 100644 --- a/application/api/user/workflows/routes.py +++ b/application/api/user/workflows/routes.py @@ -5,6 +5,10 @@ from typing import Any, Dict, List, Optional, Set from flask import current_app, request from flask_restx import Namespace, Resource +from application.agents.workflows.cel_evaluator import ( + CelEvaluationError, + validate_cel_expression, +) from application.storage.db.base_repository import looks_like_uuid from application.storage.db.repositories.workflow_edges import WorkflowEdgesRepository from application.storage.db.repositories.workflow_nodes import WorkflowNodesRepository @@ -323,6 +327,14 @@ def validate_workflow_structure( errors.append( f"Condition node '{cnode_title}' case '{handle}' has an outgoing edge but no expression" ) + if has_expression: + try: + validate_cel_expression(raw_expression) + except CelEvaluationError as exc: + errors.append( + f"Condition node '{cnode_title}' case '{handle}' " + f"has an invalid expression: {exc}" + ) for handle, handle_edges in outgoing_by_handle.items(): if not handle: @@ -335,6 +347,55 @@ def validate_workflow_structure( f"must eventually reach an end node" ) + # Set State nodes were validated nowhere. A node whose CEL does not compile + # saves and publishes clean, then aborts the run on first execution — the + # user only finds out as a failed answer, with no pointer to the node. + state_nodes = [n for n in nodes if n.get("type") == "state"] + for snode in state_nodes: + snode_title = snode.get("title", snode.get("id")) + node_data = snode.get("data", {}) or {} + # The builder writes state config under ``data.config`` but other + # payloads keep it flat on ``data``; the engine reads + # ``node.config.get("config", node.config)`` for exactly this reason, + # so accept both rather than silently validating nothing. + nested = node_data.get("config") + source = nested if isinstance(nested, dict) else node_data + operations = source.get("operations", []) + if not isinstance(operations, list): + continue + for index, operation in enumerate(operations): + if not isinstance(operation, dict): + continue + raw_expression = operation.get("expression", "") + target_variable = operation.get("target_variable", "") + has_expression = isinstance(raw_expression, str) and bool( + raw_expression.strip() + ) + has_target = isinstance(target_variable, str) and bool( + target_variable.strip() + ) + # The engine silently skips an operation missing either half + # (workflow_engine._execute_state_node), so downstream nodes read a + # variable that never gets set. Surface it at save time instead. + if has_expression and not has_target: + errors.append( + f"Set State node '{snode_title}' operation {index + 1} " + f"has an expression but no target variable" + ) + if has_target and not has_expression: + errors.append( + f"Set State node '{snode_title}' operation {index + 1} " + f"has a target variable but no expression" + ) + if has_expression: + try: + validate_cel_expression(raw_expression) + except CelEvaluationError as exc: + errors.append( + f"Set State node '{snode_title}' operation {index + 1} " + f"has an invalid expression: {exc}" + ) + agent_nodes = [n for n in nodes if n.get("type") == "agent"] for agent_node in agent_nodes: agent_title = agent_node.get("title", agent_node.get("id", "unknown")) diff --git a/docs/content/Agents/nodes.mdx b/docs/content/Agents/nodes.mdx index 76656742..d0b2eb12 100644 --- a/docs/content/Agents/nodes.mdx +++ b/docs/content/Agents/nodes.mdx @@ -5,9 +5,29 @@ DocsGPT workflows are composed of **Nodes** that are connected to form a process ## The Shared State Every workflow run maintains a state object (a JSON-like dictionary). -- **Initial State**: Contains the user's input query (`{{query}}`) and chat history (`{{chat_history}}`). -- **Accessing Variables**: You can access any variable in the state using the double-curly braces syntax: `{{variable_name}}`. +- **Initial State**: Contains the user's input query (`query`) and chat history (`chat_history`). - **Modifying State**: Nodes read from this state and write their outputs back to it. +- **Node Outputs**: Each node writes its result to `node__output`. + +### Two ways to reference state + +Which syntax a field takes depends on the field, so check this table before +typing `{{ }}` anywhere: + +| Field | Node | Syntax | +| --- | --- | --- | +| Prompt Template | AI Agent | Template — `{{variable_name}}` | +| Output Template | End | Template — `{{variable_name}}` | +| Expression | Set State | [CEL](https://cel.dev/) — bare `variable_name` | +| Expression | Condition | [CEL](https://cel.dev/) — bare `variable_name` | + +**Template fields** substitute `{{variable_name}}` into surrounding text, so +`Analysis: {{analysis_result}}` produces a string. + +**Expression fields** are evaluated as [Common Expression +Language](https://cel.dev/), where a bare name *is* the variable. Write +`query`, not `{{query}}` — the braces are a syntax error there, and saving a +workflow that contains one is rejected with the correction. --- @@ -53,26 +73,65 @@ The **Set State Node** allows you to manipulate variables within the shared stat ### Operations -You can define multiple operations in a single node. Each operation targets a specific **Key** (variable name). +You can define multiple operations in a single node. Each operation has two +parts: -1. **Set**: Assigns a specific value to a variable. - - *Value*: Can be a static string or a template using variables. - - *Example*: Set `current_step` to `1`. - - *Example*: Set `formatted_response` to `Analysis: {{analysis_result}}`. +- **Target Variable**: the state key to write to. +- **Expression**: a [CEL](https://cel.dev/) expression evaluated against the + current state. Its result becomes the variable's value. -2. **Increment**: Increases the value of a numeric variable. - - *Value*: The amount to add (default is 1). - - *Example*: Increment `retry_count` by `1`. +Reference state variables by bare name. `{{ }}` is template syntax and does +**not** work here — an operation must supply both parts, or it is rejected +when you save. -3. **Append**: Adds a value to a list variable. - - *Value*: The item to add to the list. - - *Example*: Append `{{last_result}}` to `history_list`. +| Goal | Target Variable | Expression | +| --- | --- | --- | +| Initialize a counter | `retry_count` | `0` | +| Increment a counter | `retry_count` | `retry_count + 1` | +| Copy a node's output to a stable name | `context` | `node_search_1_output` | +| Build a string | `formatted_response` | `"Analysis: " + analysis_result` | +| Append to a list | `history_list` | `history_list + [last_result]` | +| Derive a boolean | `needs_review` | `size(context) < 100` | + +CEL supports arithmetic, string concatenation with `+`, comparisons, +`&&`/`||`, ternaries (`cond ? a : b`), and built-ins such as `size()`, +`startsWith()`, and `contains()`. ### Usage Examples -- **Loop Counters**: Use a *Set State* node to initialize a counter (`i = 0`) before a loop, and another to increment it inside the loop. -- **Accumulators**: Use *Append* to collect results from multiple parallel branches into a single list. -- **Renaming**: Copy the output of a previous node to a more generic name (e.g., set `context` to `{{search_results}}`) so subsequent nodes can use a standard variable name. +- **Loop Counters**: Initialize `retry_count` to `0` before a loop, then set it + to `retry_count + 1` inside the loop. +- **Accumulators**: Collect results across branches with + `history_list + [last_result]`. +- **Renaming**: Copy a previous node's output to a generic name (target + `context`, expression `node_search_1_output`) so later nodes can use one + standard variable. + +--- + +## Condition Node + +The **Condition Node** branches the workflow. Each case pairs a +[CEL](https://cel.dev/) **Expression** with an outgoing branch; the first case +whose expression is true wins, and execution follows that branch. + +- Expressions use the same syntax as the Set State node: reference state by + bare name, not `{{ }}`. +- Every condition node needs an **else** branch, which is taken when no case + matches. +- Each case with an expression must have an outgoing edge, and every branch + must eventually reach an end node. + +| Goal | Expression | +| --- | --- | +| Route on a previous answer | `node_classify_1_output == "refund"` | +| Guard on retrieved context | `size(context) > 0` | +| Combine checks | `needs_review && retry_count < 3` | +| Match text | `query.contains("invoice")` | + +A case whose expression fails to evaluate at run time is skipped and the next +case is tried, so a workflow that always lands on **else** usually means an +expression is referencing a variable that no earlier node writes. --- diff --git a/frontend/src/agents/workflow/WorkflowBuilder.tsx b/frontend/src/agents/workflow/WorkflowBuilder.tsx index a2dca188..1dc81940 100644 --- a/frontend/src/agents/workflow/WorkflowBuilder.tsx +++ b/frontend/src/agents/workflow/WorkflowBuilder.tsx @@ -2473,11 +2473,13 @@ function WorkflowBuilderInner() { }} className="border-border focus-visible:ring-ring/50 focus-visible:border-ring bg-card dark:bg-accent mb-1 w-full rounded-xl border px-3 py-2 text-sm transition-all outline-none focus-visible:ring-2 dark:text-white" rows={2} - placeholder="input.foo + 1" + placeholder="query" />

Use Common Expression Language to create - a custom expression.{' '} + a custom expression. Reference state by + bare name (query), not{' '} + {'{{query}}'}.{' '}

Use Common Expression Language to - create a custom expression.{' '} + create a custom expression. + Reference state by bare name ( + query), not{' '} + {'{{query}}'}.{' '}