"""校验动作范围，再按顺序修改大屏 JSON 副本，检查整批动作能否执行""" from pydantic import ValidationError from app.actions.plan import validate_plan_actions, validate_plan_result from app.actions.registry import apply_action from app.processor.base import BaseNode from app.processor.state import QueryState from app.schemas.action_validation import ActionValidationError, schema_errors from app.schemas.actions import ActionBatch, AddAction, UpdateCanvasAction class ValidatorNode(BaseNode): name = "validator_node" def process(self, state: QueryState) -> QueryState: # 浅拷贝状态 updated_state: QueryState = state.copy() # 校验失败仍可能进入修正循环，最终失败由 Fallback 标记 updated_state["status"] = "running" # 获取本轮待校验的动作，先检查结构，再检查范围和模拟执行结果 actions = state["actions"] try: # 转回字典重新校验，避免赋值后绕过字段校验 batch = ActionBatch.model_validate( {"actions": [action.model_dump() for action in actions]} ) except ValidationError as error: # 将结构异常转换成统一错误列表，供后续修正流程定位问题 errors = schema_errors(error) # 标记本轮未通过并清空旧模拟快照，避免后续读取过期结果 updated_state["validation_passed"] = False updated_state["validation_errors"] = errors updated_state["simulated_screen"] = None return updated_state requirement = state["requirement"] # 如果已有规划，检查动作是否覆盖规划目标，以及是否修改了规划外的内容 if state.get("screen_plan") is not None: errors = validate_plan_actions(state, batch) # 如果规划动作检查失败，记录错误并返回，不继续模拟执行 if errors: updated_state["validation_passed"] = False updated_state["validation_errors"] = errors updated_state["simulated_screen"] = None return updated_state else: # 此分支没有规划，只校验简单修改，新增动作需要走上面的规划分支 if any(isinstance(action, AddAction) for action in batch.actions): # 为每个新增动作记录下标和范围错误，便于修正时逐项定位 updated_state["validation_passed"] = False updated_state["validation_errors"] = [ ActionValidationError( action_index=index, code="scope_error", message="新增动作需要先经过规划，不能作为简单修改直接校验", ) for index, action in enumerate(batch.actions) if isinstance(action, AddAction) ] updated_state["simulated_screen"] = None return updated_state # 将需求允许的节点操作表示为三元组：作用域、目标 ID、操作名称 allowed = { ("nodes", target, change.operation) for change in requirement.changes if change.target_scope == "nodes" for target in change.target_ids } # 将需求中的画布更新和尺寸调整统一映射为 update_canvas 动作 allowed.update( ("canvas", "", "update_canvas") for change in requirement.changes if change.target_scope == "canvas" and change.operation in ("update", "resize", "update_canvas") ) # 画布没有组件 ID，使用独立作用域避免与同名组件混淆 targets = [ ("canvas", "", action.type) if isinstance(action, UpdateCanvasAction) else ("nodes", action.target_id, action.type) for action in batch.actions ] # 去重后的实际操作集合用于检查遗漏，原 targets 列表保留动作顺序和下标 actual = set(targets) # 逐项找出需求未允许的目标或操作，防止模型扩大修改范围 errors = [ ActionValidationError( action_index=index, code="scope_error", message=f"动作超出需求范围: {' / '.join(target)}", ) for index, target in enumerate(targets) if target not in allowed ] # 如果需求允许的目标操作没有对应动作，记录遗漏，避免只执行部分需求 if missing := allowed - actual: errors.append( ActionValidationError( code="scope_error", message=f"动作遗漏了需求中的目标或操作: {sorted(missing)}", ) ) # 如果存在越界或遗漏，返回全部范围错误，并清空旧模拟结果 if errors: updated_state["validation_passed"] = False updated_state["validation_errors"] = errors updated_state["simulated_screen"] = None return updated_state # 在副本上试执行，不改变用户传入的大屏，也不操作前端画布 screen = state["input_context"].screen.model_copy(deep=True) # 按动作顺序逐项执行，每一步都使用上一步更新后的大屏副本 for index, action in enumerate(batch.actions): try: # 根据动作类型调用对应处理逻辑，检查目标和参数并返回更新后的大屏 screen = apply_action(screen, action) except ValueError as error: # 首个失败即停止，后续动作可能依赖前序结果，不返回部分成功快照 updated_state["validation_passed"] = False updated_state["validation_errors"] = [ ActionValidationError( action_index=index, code="execution_error", message=str(error) ) ] updated_state["simulated_screen"] = None return updated_state # 如果已有规划，还需核对整批执行后的大屏是否满足规划，不能只看单步执行成功 if state.get("screen_plan") is not None: errors = validate_plan_result(state, screen) # 如果最终结果与规划不符，记录错误并丢弃模拟快照，交给后续修正流程 if errors: updated_state["validation_passed"] = False updated_state["validation_errors"] = errors updated_state["simulated_screen"] = None return updated_state # 全部检查通过后，保存重新校验过的动作对象，统一后续读取的数据结构 updated_state["actions"] = batch.actions # 标记整批动作校验通过，并清空之前可能残留的错误 updated_state["validation_passed"] = True updated_state["validation_errors"] = [] # 保存整批动作执行后的 JSON，供后续检查，不代表前端已经完成更新 updated_state["simulated_screen"] = screen # 仅表示后端动作已通过校验，前端执行结果需另行反馈 updated_state["status"] = "completed" # 返回动作、校验结果和完整模拟快照，供后续输出使用 return updated_state if __name__ == "__main__": import json from app.processor.nodes.input_node import InputNode from app.schemas.actions import MoveAction, ResizeAction, UpdateAction from app.schemas.requirement import Requirement # 1. 创建 state，不需要调用模型 state = InputNode().process( { "request": { "prompt": "移动、调整尺寸并修改标题", "screen": { "canvas": {"width": 1920, "height": 1080}, "nodes": [ { "id": "sales-chart", "type": "chart", "layout": {"x": 100, "y": 200, "width": 400, "height": 300}, "props": {"title": "销售额"}, } ], }, } } ) state["requirement"] = Requirement( intent="modify_screen", goal="移动、调整尺寸并修改标题", needs_clarification=False, changes=[ {"target_scope": "nodes", "target_ids": ["sales-chart"], "operation": op} for op in ("move", "resize", "update") ], ) state["actions"] = [ MoveAction(target_id="sales-chart", delta_x=50), ResizeAction(target_id="sales-chart", width=500), UpdateAction( target_id="sales-chart", section="props", path=["title"], value="月度销售额" ), ] # 2. 执行节点并打印完整 state node = ValidatorNode() final_state = node.process(state) print( json.dumps( final_state, ensure_ascii=False, indent=2, default=lambda value: value.model_dump(mode="json"), ) ) # 3. 在第二个动作制造错误，验证整批失败且不返回部分结果 failed_state = { **final_state, "requirement": Requirement( intent="modify_screen", goal="移动两个组件", needs_clarification=False, changes=[ {"target_scope": "nodes", "target_ids": [target], "operation": "move"} for target in ("sales-chart", "missing") ], ), "actions": [ state["actions"][0], MoveAction(target_id="missing", delta_x=50), ], } failed_result = node.process(failed_state) print("失败校验:", failed_result["validation_passed"]) print("错误:", [error.model_dump() for error in failed_result["validation_errors"]]) print("失败快照:", failed_result["simulated_screen"])