"""校验动作范围，再按顺序修改大屏 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 allowed = { ("nodes", target, change.operation) for change in requirement.changes if change.target_scope == "nodes" for target in change.target_ids } 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 ] 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"])