Spaces:
Runtime error
Runtime error
File size: 5,000 Bytes
595d60e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import re
from typing import List, Set, Dict, Any
from ..schemas.agent_spec import AgentSpec
from ..schemas.conversation import ValidationError, ValidationResult
class WorkflowValidator:
def validate(self, spec: AgentSpec) -> ValidationResult:
errors = []
steps_dict = {s.id: s for s in spec.steps}
step_ids = list(steps_dict.keys())
# 1. Duplicate step IDs
all_ids = [s.id for s in spec.steps]
if len(all_ids) != len(set(all_ids)):
errors.append(ValidationError(
code="DUPLICATE_STEP_ID",
message="Duplicate step IDs found in the workflow",
path="steps"
))
# 2. Depends_on exists
for step in spec.steps:
for dep in step.depends_on:
if dep not in steps_dict:
errors.append(ValidationError(
code="MISSING_DEPENDENCY",
message=f"Step '{step.id}' depends on missing step '{dep}'",
path=f"steps.{step.id}.depends_on"
))
# 3. Circular dependency detection (Cycle Detection using DFS)
visited = {} # state: 0=unvisited, 1=visiting, 2=visited
for s_id in step_ids:
visited[s_id] = 0
def dfs(s_id: str, path: List[str]) -> bool:
visited[s_id] = 1 # visiting
step = steps_dict[s_id]
for dep in step.depends_on:
if dep not in steps_dict:
continue
if visited[dep] == 1:
cycle_path = " -> ".join(path + [dep])
errors.append(ValidationError(
code="CIRCULAR_DEPENDENCY",
message=f"Circular dependency detected: {cycle_path}",
path=f"steps.{s_id}.depends_on"
))
return True
elif visited[dep] == 0:
if dfs(dep, path + [dep]):
return True
visited[s_id] = 2 # visited
return False
for s_id in step_ids:
if visited[s_id] == 0:
dfs(s_id, [s_id])
# 4. Helper to find all ancestors of a step
def get_all_ancestors(s_id: str, memo: Dict[str, Set[str]]) -> Set[str]:
if s_id in memo:
return memo[s_id]
ancestors = set()
step = steps_dict.get(s_id)
if step:
for dep in step.depends_on:
ancestors.add(dep)
ancestors.update(get_all_ancestors(dep, memo))
memo[s_id] = ancestors
return ancestors
ancestors_memo = {}
# 5. Data flow validation (regex check for $step_id.field references)
ref_pattern = re.compile(r"^\$(\w+)(?:\.(\w+))?$")
def scan_for_refs(val: Any) -> List[tuple]:
refs = []
if isinstance(val, str):
match = ref_pattern.match(val)
if match:
refs.append((match.group(1), match.group(2)))
elif isinstance(val, dict):
for k, v in val.items():
refs.extend(scan_for_refs(v))
elif isinstance(val, list):
for item in val:
refs.extend(scan_for_refs(item))
return refs
for step in spec.steps:
refs = scan_for_refs(step.input)
for ref_step_id, ref_field in refs:
if ref_step_id not in steps_dict:
errors.append(ValidationError(
code="INVALID_DATA_FLOW",
message=f"Step '{step.id}' references input from non-existent step '{ref_step_id}'",
path=f"steps.{step.id}.input"
))
continue
ancestors = get_all_ancestors(step.id, ancestors_memo)
if ref_step_id != step.id and ref_step_id not in ancestors:
errors.append(ValidationError(
code="INVALID_DATA_FLOW",
message=f"Step '{step.id}' references output of step '{ref_step_id}' but does not depend on it",
path=f"steps.{step.id}.depends_on"
))
ref_step = steps_dict[ref_step_id]
if ref_field and ref_step.output:
if ref_field not in ref_step.output:
errors.append(ValidationError(
code="INVALID_DATA_FLOW",
message=f"Step '{step.id}' references field '{ref_field}' which is not defined in step '{ref_step_id}' outputs",
path=f"steps.{step.id}.input"
))
return ValidationResult(valid=len(errors) == 0, errors=errors)
|