"""校验动作范围，再按顺序修改大屏 JSON 副本，检查整批动作能否执行""" from pydantic import BaseModel, 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() if isinstance(action, BaseModel) else action for action in actions ] if isinstance(actions, list) else 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.get("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 elif requirement is not None: # 此分支没有规划，只校验简单修改，新增动作需要走上面的规划分支 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 # 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["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, "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"])