"""将简单修改或新增规划转换为动作列表""" import json from langchain_core.exceptions import OutputParserException from langchain_core.messages import HumanMessage, SystemMessage from pydantic import BaseModel, ValidationError from app.actions.plan import assemble_plan_actions, plan_block_reason, plan_materials from app.clients.llm import get_llm from app.processor.base import BaseNode from app.processor.state import QueryState from app.prompt.action import ACTION_SYSTEM_PROMPT from app.schemas.action_validation import ActionValidationError, schema_errors from app.schemas.actions import ActionBatch, AddAction def can_generate_actions(state: QueryState) -> bool: if state.get("screen_plan") is not None: return plan_block_reason(state) is None requirement = state["requirement"] scope = state["change_scope"] # 当前只接入三种已有组件操作，后续能力逐步扩展 return ( requirement.intent == "modify_screen" and not requirement.needs_clarification and not scope.needs_planning and not scope.resource_needs and bool(requirement.changes) and all( change.target_scope == "nodes" and change.operation in ( "move", "resize", "update", ) for change in requirement.changes ) ) class ActionNode(BaseNode): name = "action_node" async def process(self, state: QueryState) -> QueryState: updated_state: QueryState = state.copy() if state.get("screen_plan") is not None and ( reason := plan_block_reason(state) ): updated_state.update( actions=None, validation_passed=False, validation_errors=[], simulated_screen=None, failure_reason=reason, status="failed", ) return updated_state if not can_generate_actions(state): raise ValueError( "ActionNode 当前只支持无需规划和资源的 move、resize、update 需求" ) payload = { "input_context": state["input_context"].model_dump(mode="json"), "requirement": state["requirement"].model_dump(mode="json"), "change_scope": state["change_scope"].model_dump(mode="json"), } if state.get("screen_plan") is not None: payload["screen_plan"] = state["screen_plan"].model_dump(mode="json") payload["component_materials"] = plan_materials(state) if state.get("repair_plan") is not None: payload.update( repair_plan=state["repair_plan"].model_dump(mode="json"), previous_actions=[ action.model_dump(mode="json") for action in state.get("actions") or [] ], validation_errors=[ error.model_dump() for error in state.get("validation_errors", []) ], ) # 配置和网络错误继续抛出，仅模型输出错误进入内容修正循环 llm_client = get_llm(ActionBatch) updated_state["actions"] = None updated_state["validation_passed"] = None updated_state["validation_errors"] = [] updated_state["simulated_screen"] = None updated_state["failure_reason"] = None updated_state["status"] = "running" try: result = await llm_client.ainvoke( [ SystemMessage(content=ACTION_SYSTEM_PROMPT), HumanMessage(content=json.dumps(payload, ensure_ascii=False)), ] ) raw = result.model_dump() if isinstance(result, BaseModel) else result if isinstance(result, ActionBatch) and state.get("screen_plan") is not None: for source, target in zip(result.actions, raw["actions"]): if isinstance(source, AddAction): target["node"] = source.node.model_dump(exclude_unset=True) batch = ActionBatch.model_validate(raw) if state.get("screen_plan") is not None: batch = assemble_plan_actions(state, batch) except ValidationError as error: updated_state["validation_passed"] = False updated_state["validation_errors"] = schema_errors(error) return updated_state except OutputParserException: updated_state["validation_passed"] = False updated_state["validation_errors"] = [ ActionValidationError( code="generation_error", message="模型输出无法解析为 ActionBatch，请按协议重新生成", ) ] return updated_state updated_state["actions"] = batch.actions return updated_state if __name__ == "__main__": import asyncio import sys from pydantic import BaseModel from app.processor.nodes.component_resource_node import ComponentResourceNode from app.processor.nodes.input_node import InputNode from app.processor.nodes.validator_node import ValidatorNode from app.schemas.change_scope import ChangeScope from app.schemas.requirement import Requirement from app.schemas.screen_plan import ScreenPlan async def main(): state = InputNode().process( { "request": { "prompt": "将销售额统计向右移动 50px", "screen": { "canvas": {"width": 1920, "height": 1080}, "nodes": [ { "id": "sales-chart", "type": "chart", "layout": { "x": 100, "y": 200, "width": 400, "height": 300, }, } ], }, } } ) state["requirement"] = Requirement( intent="modify_screen", goal="向右移动 50px", needs_clarification=False, changes=[ { "target_scope": "nodes", "target_ids": ["sales-chart"], "operation": "move", "parameters": {"delta_x": 50}, } ], ) state["change_scope"] = ChangeScope( scope="local", target_ids=["sales-chart"], needs_planning=False, reason="明确的组件位移", ) # 加 --add 测试规划到新增动作，默认仍测试简单移动 if "--add" in sys.argv: state = InputNode().process( { "request": { "prompt": "保留已有组件，在顶部新增销售总览标题", "screen": state["input_context"].screen.model_dump(mode="json"), } } ) state["requirement"] = Requirement( intent="modify_screen", goal="新增标题", needs_clarification=False, changes=[{"target_scope": "screen", "operation": "add"}], ) state["change_scope"] = ChangeScope( scope="local", needs_planning=True, reason="顶部新增标题", ) state = ComponentResourceNode().process(state) state["screen_plan"] = ScreenPlan( goal="新增标题", design_summary="顶部标题", layout_description="保留原图表", components=[ { "plan_id": "heading", "disposition": "add", "component_type": "text", "purpose": "显示销售总览", "design": {"text": "销售总览"}, "layout": {"x": 100, "y": 40, "width": 500, "height": 60}, } ], ) node = ActionNode() final_state = await node.process(state) if "--add" in sys.argv and final_state.get("actions") is not None: final_state = ValidatorNode().process(final_state) print( json.dumps( final_state, ensure_ascii=False, indent=2, default=lambda value: ( value.model_dump(mode="json") if isinstance(value, BaseModel) else str(value) ), ) ) asyncio.run(main())