From ff8adba01b36866830475d62a11e6100452e7f95 Mon Sep 17 00:00:00 2001 From: Marcus <1922576605@qq.com> Date: Tue, 29 Sep 2026 19:21:29 +0800 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E6=96=B0=E5=A2=9E=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/processor/nodes/action_node.py | 48 ++++++++++++++++++- app/processor/nodes/clarify_node.py | 18 ++++++- .../nodes/component_resource_node.py | 12 ++++- app/processor/nodes/fallback_node.py | 13 ++++- app/processor/nodes/input_node.py | 30 ++++++++++-- app/processor/nodes/intent_node.py | 34 +++++++++++-- app/processor/nodes/planner_node.py | 31 +++++++++++- app/processor/nodes/qa_node.py | 14 +++++- app/processor/nodes/reflection_node.py | 20 ++++++++ app/processor/nodes/requirement_node.py | 21 +++++++- app/processor/nodes/scope_node.py | 19 +++++++- app/processor/nodes/validator_node.py | 35 +++++++++++++- app/resources/component.py | 18 +++++++ 13 files changed, 293 insertions(+), 20 deletions(-) diff --git a/app/processor/nodes/action_node.py b/app/processor/nodes/action_node.py index ca4be53..6276a44 100644 --- a/app/processor/nodes/action_node.py +++ b/app/processor/nodes/action_node.py @@ -16,25 +16,34 @@ from app.schemas.actions import ActionBatch, AddAction def can_generate_actions(state: QueryState) -> bool: + # 没有阻塞原因时才允许生成动作，路由和节点入口复用同一套能力检查 return action_block_reason(state) is None def action_block_reason(state: QueryState) -> str | None: - # 路由和失败说明共用能力判断，避免支持范围不一致 + # 如果已有大屏规划，按规划检查资源、目标和操作是否受支持 if state.get("screen_plan") is not None: return plan_block_reason(state) + # 没有规划时，读取需求和范围，检查是否可以直接生成简单修改动作 requirement = state["requirement"] scope = state["change_scope"] + # 如果需求尚未澄清完整，不能猜测用户意图并生成动作 if requirement.needs_clarification: return "需求尚未澄清完整，不能生成动作" + # 如果需要规划或不是修改已有大屏，则必须先完成规划流程 if scope.needs_planning or requirement.intent != "modify_screen": return "当前需求需要先完成规划，不能直接生成动作" + # 如果仍需查询额外资源，当前直接操作流程无法满足需求 if scope.resource_needs: return "尚未接入所需资源查询：" + "；".join(scope.resource_needs) + # 如果没有具体变更，就没有可转换为动作的需求 if not requirement.changes: return "需求中没有可执行的变更" + + # 收集不支持的目标范围与操作组合，防止只执行需求中受支持的部分 + # 直接操作仅支持下列已有节点修改和画布属性修改 unsupported = [ f"{change.target_scope}/{change.operation}" for change in requirement.changes @@ -45,8 +54,11 @@ def action_block_reason(state: QueryState) -> str | None: and change.operation in ("update", "resize", "update_canvas") ) ] + + # 如果存在不支持的操作，按出现顺序去重后返回原因，便于展示失败信息 if unsupported: return "当前尚未支持的直接操作：" + "、".join(dict.fromkeys(unsupported)) + # 所有检查通过，返回 None 表示没有阻塞原因 return None @@ -54,10 +66,14 @@ 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, @@ -67,21 +83,30 @@ class ActionNode(BaseNode): status="failed", ) return updated_state + + # 如果未通过动作能力检查，抛出异常，阻止不受支持的需求进入生成流程 if not can_generate_actions(state): raise ValueError( "ActionNode 当前只支持无需规划和资源的 move、resize、update、delete、update_canvas 需求" ) + + # 将输入上下文、结构化需求和修改范围转成字典，作为本次模型输入 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 [] @@ -90,9 +115,10 @@ class ActionNode(BaseNode): 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"] = [] @@ -100,7 +126,9 @@ class ActionNode(BaseNode): updated_state["failure_reason"] = None updated_state["status"] = "running" + # 捕获输出结构和解析错误，转换成统一校验结果供后续修正流程使用 try: + # 异步调用模型，根据需求、规划及是否正在修正选择对应的系统规则 result = await llm_client.ainvoke( [ SystemMessage( @@ -110,22 +138,35 @@ class ActionNode(BaseNode): repairing=state.get("repair_plan") is not None, ) ), + # 将本次具体资料放入用户消息，保留中文以便阅读和排查 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) + + # 统一校验整批动作的字段和结构，格式不合法时抛出 ValidationError 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: + # 如果模型输出无法解析，记录生成错误，提示后续按 ActionBatch 协议重新生成 updated_state["validation_passed"] = False updated_state["validation_errors"] = [ ActionValidationError( @@ -135,7 +176,10 @@ class ActionNode(BaseNode): ] return updated_state + # 保存生成的动作，实际目标、规划一致性及模拟执行结果仍由后续校验节点检查 updated_state["actions"] = batch.actions + + # 返回待校验状态，此时 validation_passed 仍为 None，不代表动作已执行或通过校验 return updated_state diff --git a/app/processor/nodes/clarify_node.py b/app/processor/nodes/clarify_node.py index 2061ea9..10fbe77 100644 --- a/app/processor/nodes/clarify_node.py +++ b/app/processor/nodes/clarify_node.py @@ -17,30 +17,37 @@ class ClarifyNode(BaseNode): name = "clarify_node" async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态，保存澄清问题和等待状态时不直接修改传入 state 的顶层字段 updated_state: QueryState = state.copy() + # 获取上游确认的意图，区分意图不明确和具体需求缺少信息两种澄清场景 intent = state["intent"] + # 整理意图和输入上下文，让模型结合用户原话、大屏及已有澄清记录生成问题 payload = { "intent": intent, "input_context": state["input_context"].model_dump(mode="json"), } + # 如果意图已明确，补充需求节点的结果，围绕尚缺的信息提问 if intent != "unclear": # 以需求节点的问题为依据补齐选项，不重新判断意图 payload["requirement"] = state["requirement"].model_dump(mode="json") llm_client = get_llm(Clarification) + # 系统消息提供稳定的澄清规则，用户消息承载本次需要分析的具体资料 messages = [ SystemMessage(content=CLARIFY_SYSTEM_PROMPT), HumanMessage(content=json.dumps(payload, ensure_ascii=False)), ] - # 格式错误重试一次 + # 最多调用两次，仅在输出结构或解析失败时重试一次，其他异常继续向上抛出 for attempt in range(2): try: + # 异步生成澄清问题和候选项，成功后立即退出重试循环 clarification = await llm_client.ainvoke(messages) break except (ValidationError, OutputParserException): + # 如果首次生成格式不合法，追加明确的格式要求，再调用一次模型 if attempt == 0: # 要求重新生成完整问题，避免截断选项改变原意 messages.append( @@ -52,17 +59,24 @@ class ClarifyNode(BaseNode): ) ) ) + # 进入下一次尝试，使用包含格式纠正要求的消息列表重新生成 continue - # 连续格式失败时退回自由回答，需求阶段沿用已确定的问题 + + # 意图不明确时不读取需求，避免使用尚未生成或不适用的需求结果 requirement = state.get("requirement") if intent != "unclear" else None + # 优先使用需求节点已有的澄清问题，没有可用问题时使用通用询问 question = ( requirement.clarification_question if requirement else None ) or "你希望我对当前大屏做什么？可以描述想修改的内容，或想了解的问题" + # 只构建问题，不提供候选项，让用户自由补充，避免继续依赖模型格式修正 clarification = Clarification(question=question) + # 保存模型生成或兜底构建的澄清内容，供后续暂停流程展示给用户 updated_state["clarification"] = clarification # 在进入 interrupt 前保存等待状态，暂停中的节点尚不会返回更新 updated_state["status"] = "waiting_for_input" + + # 返回澄清内容和等待状态，实际暂停由后续调用 interrupt 的节点完成 return updated_state diff --git a/app/processor/nodes/component_resource_node.py b/app/processor/nodes/component_resource_node.py index 10312e1..6844f78 100644 --- a/app/processor/nodes/component_resource_node.py +++ b/app/processor/nodes/component_resource_node.py @@ -10,17 +10,27 @@ class ComponentResourceNode(BaseNode): name = "component_resource_node" def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() - # 当前目录较小，先提供全量，避免关键词筛选遗漏可用组件 + # 空查询不设置筛选条件，由提供器读取默认组件目录并返回查询结果 result = ComponentProvider().query(ComponentQuery()) + # 保存组件定义、目录版本和查询状态，供后续规划和动作生成使用 updated_state["component_resources"] = result + # 清空旧规划，后续需要根据本次查询到的组件能力重新规划 updated_state["screen_plan"] = None + + # 如果目录未配置或没有查到组件，标记失败，避免基于未知组件能力继续规划 if result.status != "resolved": updated_state["status"] = "failed" + # 沿用查询结果中的说明，保留无法获取组件的具体原因 updated_state["failure_reason"] = result.message else: + # 如果已获取组件定义，保持运行状态，让后续节点继续规划 updated_state["status"] = "running" + # 清空旧的失败原因，避免成功查询后仍携带上一轮错误 updated_state["failure_reason"] = None + + # 返回资源查询结果和处理状态，由图决定继续规划还是进入失败处理 return updated_state diff --git a/app/processor/nodes/fallback_node.py b/app/processor/nodes/fallback_node.py index 4ab9bc0..61ae0ca 100644 --- a/app/processor/nodes/fallback_node.py +++ b/app/processor/nodes/fallback_node.py @@ -8,19 +8,30 @@ class FallbackNode(BaseNode): name = "fallback_node" def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + # 读取上游记录的具体失败原因 reason = state.get("failure_reason") + + # 如果上游没有提供有效原因，则根据校验错误补充失败说明 if not reason: + # 如果仍有校验错误，使用达到自动修正上限的说明，并带上已重试次数 if state.get("validation_errors"): reason = f"动作校验未通过，已达到自动修正上限，共重试 {state.get('retry_count', 0)} 次" else: + # 如果既没有具体原因也没有校验错误，使用通用说明，避免返回空原因 reason = "当前需求未能完成，缺少具体失败原因" - # 失败动作仅保留作诊断，validation_passed=False 表示不可交付执行 + # 将任务标记为最终失败，结束本轮处理 updated_state["status"] = "failed" + # 保存上游原因或本节点补充的说明，供后续输出给用户 updated_state["failure_reason"] = reason + # 显式标记校验未通过，防止残留动作被当作可执行结果 updated_state["validation_passed"] = False + # 清空旧模拟快照，避免失败结果中仍携带之前的成功快照 updated_state["simulated_screen"] = None + + # 返回统一的失败状态，保留原动作和错误信息供排查 return updated_state diff --git a/app/processor/nodes/input_node.py b/app/processor/nodes/input_node.py index 37a3d0a..47c0d72 100644 --- a/app/processor/nodes/input_node.py +++ b/app/processor/nodes/input_node.py @@ -11,16 +11,22 @@ class InputNode(BaseNode): name = "input_node" def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + # 获取待合并的用户澄清回复 reply = state.get("clarification_reply") + # 如果有澄清回复，将补充信息合并到已有上下文后重新处理 if reply is not None: + # 将已有上下文转成独立的字典，避免直接修改原上下文中的嵌套数据 data = state["input_context"].model_dump(mode="json") + # 记录本轮澄清问题和用户回答，供后续节点理解补充需求 turn = { "question": state["clarification"].question, "answer": reply.answer, } + # 如果用户选择了候选项，按选项 ID 查找并保存完整选项内容 if reply.option_id is not None: option = next( option @@ -28,48 +34,64 @@ class InputNode(BaseNode): if option.id == reply.option_id ) turn["selected_option"] = option.model_dump(mode="json") + + # 将本轮问答追加到历史中，保留多轮澄清信息 data["clarification_history"].append(turn) + # 如果回复携带最新大屏快照，用它替换旧快照 if reply.screen is not None: data["screen"] = reply.screen.model_dump(mode="json") - # 新快照不沿用可能已经失效的选中节点 + # 如果没有同时提供交互上下文，清空可能已失效的选中节点和版本号 if reply.context is None: data["context"]["selected_node_ids"] = [] data["context"]["screen_revision"] = None + + # 如果回复携带交互上下文，使用本次提交的选中节点和版本等信息 if reply.context is not None: data["context"] = reply.context.model_dump(mode="json") + # 重新校验合并后的数据并构建上下文，校验失败时向上抛出 ValidationError context = InputContext.model_validate(data) - # 显式清空旧推导结果，LangGraph 的状态合并不会删除省略字段 + # 保存新上下文并清空已消费的回复，避免后续重复合并 updated_state["input_context"] = context updated_state["clarification_reply"] = None + # 清空旧意图、需求和规划，让后续节点根据补充信息重新推导 + # LangGraph 的状态合并不会删除省略字段，因此需要显式赋空值 updated_state["intent_decision"] = None updated_state["intent"] = None updated_state["requirement"] = None updated_state["change_scope"] = None updated_state["screen_plan"] = None updated_state["component_resources"] = None + # 清空旧动作及其校验、模拟结果，避免复用补充需求前的处理结果 updated_state["actions"] = None updated_state["validation_passed"] = None updated_state["validation_errors"] = [] updated_state["simulated_screen"] = None + # 重置自动修正次数和失败信息，让本轮处理重新计算重试次数 updated_state["retry_count"] = 0 updated_state["repair_plan"] = None updated_state["failure_reason"] = None + # 清空上一轮澄清问题和回答，恢复任务处理状态 updated_state["clarification"] = None updated_state["answer"] = None updated_state["status"] = "running" + + # 返回澄清恢复后的状态，不再执行下方首次请求的初始化逻辑 return updated_state - # 深拷贝隔离原始请求，后续修改上下文不会改变提交时的快照 + # 没有澄清回复时，按首次请求校验原始输入，并深拷贝隔离提交时的快照 context = InputContext.model_validate(state["request"]).model_copy(deep=True) + # 保存校验后的上下文，供后续节点统一读取 updated_state["input_context"] = context + # 初始化自动修正次数，并清空修正建议和失败原因 updated_state["retry_count"] = 0 updated_state["repair_plan"] = None updated_state["failure_reason"] = None # 首次请求进入处理状态，恢复澄清时也使用同一状态 updated_state["status"] = "running" + # 返回初始化后的状态，由图接续执行后续节点 return updated_state @@ -106,7 +128,7 @@ if __name__ == "__main__": node = InputNode() final_state = node.process(state) - # 3. 打印完整状态 + # 3. 将上下文模型转成可序列化的字典，再打印完整状态 printable_state = { **final_state, "input_context": final_state["input_context"].model_dump( diff --git a/app/processor/nodes/intent_node.py b/app/processor/nodes/intent_node.py index 4e69afc..0556fd3 100644 --- a/app/processor/nodes/intent_node.py +++ b/app/processor/nodes/intent_node.py @@ -18,25 +18,36 @@ class IntentNode(BaseNode): @staticmethod def build_decision_state(context: InputContext) -> JsonObject: """只整理判断意图需要的摘要，完整 Screen 仍保留在输入上下文""" + + # 收集每个组件的识别信息和布局，帮助判断用户指的是哪个组件 nodes: list[JsonObject] = [] + for node in context.screen.nodes: + # 获取图表配置，组件没有 option 时返回 None option = node.props.get("option") + # 只有字典配置才能读取 title，其他类型按没有标题处理 title = option.get("title") if isinstance(option, dict) else None + # 将单个标题统一包装成列表，兼容图表配置中的多个标题 titles = title if isinstance(title, list) else [title] + + # 保存组件摘要，不传递完整 props，减少意图判断时的无关信息 nodes.append( { "id": node.id, "type": node.type, "name": node.name, + # 仅提取字典中字符串类型的标题文本，忽略缺失或格式不符的标题 "titles": [ item["text"] for item in titles if isinstance(item, dict) and isinstance(item.get("text"), str) ], + # 将布局转成可序列化的字典，只保留显式设置的字段 "layout": node.layout.model_dump(mode="json", exclude_unset=True), } ) + # 汇总用户需求、澄清历史、交互上下文和大屏摘要，作为意图判断的输入 return { "prompt": context.prompt, "clarification_history": context.clarification_history, @@ -47,6 +58,7 @@ class IntentNode(BaseNode): "height": context.screen.canvas.height, }, "nodes": nodes, + # 数据源只保留已有的标识、名称和类型，省略具体数据及其他配置 "data_sources": [ { key: source[key] @@ -56,6 +68,7 @@ class IntentNode(BaseNode): for source in context.screen.dataSource ], }, + # 附件只保留已有的标识、类别和名称，用于判断请求涉及哪些附件 "attachments": [ { key: attachment[key] @@ -67,32 +80,42 @@ class IntentNode(BaseNode): } async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + # 获取输入节点已校验的上下文，包含最新大屏和澄清历史 context = state["input_context"] - # 构建jev请求 + # 构建 Jev 决策请求，将上下文摘要与意图判断问题一起提交 request = DecisionRequest( # 将 InputNode 的上下文整理为意图判断摘要 state=self.build_decision_state(context), + # 使用 intent 作为决策名称，后续通过同名键读取判断结果 decisions={"intent": build_intent_question()}, ) - # 调用 jev + # 获取决策引擎，用于执行本次意图判断 decision_engine = get_decision_engine() - # 获取 jev 结果 + # 异步等待 Jev 返回决策结果，包含选项、置信度和调用信息 decision_res = await decision_engine.run(request) - # 原生 unclear 或低可信时澄清，原始结果完整保留 + # 从响应中取出 intent 对应的判断结果 decision = decision_res.results["intent"] + + # 如果模型无法明确意图，或置信度低于配置阈值，则转入澄清流程 if ( decision.choice == "unclear" or decision.confidence < self.config.intent_min_confidence ): intent: Intent = "unclear" else: + # 置信度达标时沿用模型选项，cast 仅作类型提示，不进行运行时转换或校验 intent = cast(Intent, decision.choice) + # 完整保留原始决策，便于查看模型选项及其置信度 updated_state["intent_decision"] = decision_res + # 保存经过置信度判断的最终意图，供后续路由选择处理分支 updated_state["intent"] = intent + + # 返回更新后的状态，由图继续执行对应的处理节点 return updated_state @@ -131,8 +154,10 @@ if __name__ == "__main__": try: final_state = await node.process(state) finally: + # 无论调用成功还是抛出异常，都关闭示例使用的 Jev 连接 await close_decision_engine() # 3. 打印完整 state，包含选项、概率、置信度和调用信息 + # 将状态中的 Pydantic 模型转成字典，其他值保持原样，以便输出 JSON print( json.dumps( { @@ -146,4 +171,5 @@ if __name__ == "__main__": ) ) + # 启动事件循环，执行上面的异步调试流程 asyncio.run(main()) diff --git a/app/processor/nodes/planner_node.py b/app/processor/nodes/planner_node.py index bb14e3d..c89faaa 100644 --- a/app/processor/nodes/planner_node.py +++ b/app/processor/nodes/planner_node.py @@ -15,10 +15,13 @@ class PlannerNode(BaseNode): name = "planner_node" async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + # 获取已提取的需求和范围，作为规划内容及修改边界的依据 requirement = state["requirement"] scope = state["change_scope"] + # 如果需求仍需澄清、不是新建或修改大屏，或无需规划，则拒绝进入本节点 if ( requirement.needs_clarification or requirement.intent @@ -30,20 +33,28 @@ class PlannerNode(BaseNode): ): raise ValueError("PlannerNode 只处理信息完整且需要规划的新建或修改需求") + # 获取当前大屏和用户输入，供模型结合已有内容安排布局与组件 context = state["input_context"] + # 获取可选的组件查询结果 resources = state.get("component_resources") + + # 如果已有查询结果但目录未就绪，则停止规划，避免忽略资源查询失败 if resources is not None and resources.status != "resolved": raise ValueError("组件目录尚未就绪，不能继续规划") - result = await get_llm(ScreenPlan).ainvoke( + llm_client = get_llm(ScreenPlan) + result = await llm_client.ainvoke( [ + # 系统消息提供稳定的规划规则和输出边界 SystemMessage(content=PLANNER_SYSTEM_PROMPT), + # 用户消息承载本次上下文、需求、范围及组件能力摘要 HumanMessage( content=json.dumps( { "input_context": context.model_dump(mode="json"), "requirement": requirement.model_dump(mode="json"), "change_scope": scope.model_dump(mode="json"), + # 仅提供规划需要的组件摘要，省略完整默认配置，没有目录时传入 None "component_resources": resources.to_planner_context() if resources is not None else None, @@ -54,13 +65,18 @@ class PlannerNode(BaseNode): ] ) + # 将模型结果转成字典后重新校验，确保规划字段及组件关系符合输出协议 plan = ScreenPlan.model_validate(result.model_dump()) + # 如果提供了组件目录，则额外校验规划中的物料类型是否真实受支持 if resources is not None: + # 提取目录支持的前端组件类型，使用集合进行成员检查 supported_types = {component.type for component in resources.components} for component in plan.components: + # 如果新增组件没有指定类型，无法选择对应物料，立即报错 if component.disposition == "add" and component.component_type is None: raise ValueError(f"新增规划组件未指定物料类型: {component.plan_id}") + # 新增或修改组件时，只要明确指定了类型，就必须在目录中存在 if ( component.disposition in ("add", "modify") and component.component_type is not None @@ -70,39 +86,50 @@ class PlannerNode(BaseNode): f"规划使用了目录中不存在的组件类型: {component.component_type}" ) + # 收集当前大屏中真实存在的节点 ID，用于检查规划引用的已有目标 existing_ids = {node.id for node in context.screen.nodes} + # 汇总规划中明确引用的目标 ID，新增组件没有已有目标时不计入 planned_ids = { component.target_id for component in plan.components if component.target_id is not None } + # 如果规划引用了不存在的组件，则抛出异常并列出无效 ID if unknown := planned_ids - existing_ids: raise ValueError(f"规划引用了不存在的组件: {sorted(unknown)}") + # 汇总需求明确指定的目标 ID，避免规划遗漏用户提及的组件 required_ids = { target for change in requirement.changes for target in change.target_ids } + # 如果需求目标未在规划中出现，则停止处理并指出遗漏 if missing := required_ids - planned_ids: raise ValueError(f"规划遗漏了需求中的组件: {sorted(missing)}") + # 如果是局部规划，还需保证对已有组件的修改和删除不超出指定范围 if scope.scope == "local": + # 只统计会改变已有组件的目标，保留和新增组件不纳入此项检查 changed_ids = { component.target_id for component in plan.components if component.disposition in ("modify", "remove") } + # 如果实际修改或删除的目标不在允许范围内，则拒绝扩大修改范围 if outside := changed_ids - set(scope.target_ids): raise ValueError(f"局部规划修改了范围外的组件: {sorted(outside)}") - # 原始约束和假设不能在规划阶段被无声丢弃 + # 合并需求约束和规划约束，按首次出现顺序去重，保留需求约束在前 plan.constraints = list( dict.fromkeys([*requirement.constraints, *plan.constraints]) ) + # 同样合并原始假设与规划补充的假设，避免重复记录或丢失前置条件 plan.assumptions = list( dict.fromkeys([*requirement.assumptions, *plan.assumptions]) ) + # 保存通过校验的规划，供后续动作生成使用 updated_state["screen_plan"] = plan + # 返回更新后的状态，由图继续处理规划对应的动作 return updated_state diff --git a/app/processor/nodes/qa_node.py b/app/processor/nodes/qa_node.py index 43f9009..738c1d6 100644 --- a/app/processor/nodes/qa_node.py +++ b/app/processor/nodes/qa_node.py @@ -13,23 +13,35 @@ class QANode(BaseNode): name = "qa_node" async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + + # 获取普通对话模型，问答只需文本回答，不绑定结构化输出协议 llm_client = get_llm() + # 异步调用模型，结合用户问题、当前大屏和澄清记录生成回答 llm_res = await llm_client.ainvoke( [ + # 系统消息提供稳定的问答规则和回答边界 SystemMessage(content=QA_SYSTEM_PROMPT), + # 将输入上下文序列化为 JSON 放入用户消息，只保留显式设置的字段 HumanMessage( content=state["input_context"].model_dump_json(exclude_unset=True) ), ] ) - # 普通文本输出，同时兼容模型返回的文本内容块 + + # 从模型响应中提取文本，兼容文本内容块，并去除回答首尾的空白 answer = StrOutputParser().invoke(llm_res).strip() + # 如果没有有效文本，抛出异常，避免将空回答标记为处理完成 if not answer: raise ValueError("问答模型未返回文本回答") + + # 保存最终回答，供后续输出给用户 updated_state["answer"] = answer # 回答生成完成，问答分支在此结束 updated_state["status"] = "completed" + + # 返回包含回答和完成状态的更新结果 return updated_state diff --git a/app/processor/nodes/reflection_node.py b/app/processor/nodes/reflection_node.py index 13c4768..4d8baf1 100644 --- a/app/processor/nodes/reflection_node.py +++ b/app/processor/nodes/reflection_node.py @@ -18,44 +18,64 @@ class ReflectionNode(BaseNode): name = "reflection_node" async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + llm_client = get_llm(ActionRepairPlan) + # 汇总原始上下文、需求和失败信息，让模型基于实际错误提出修正建议 payload = { "input_context": state["input_context"].model_dump(mode="json"), "requirement": state["requirement"].model_dump(mode="json"), + # 动作可能在生成或解析阶段就失败，没有动作时传入空列表 "actions": [ action.model_dump(mode="json") for action in state.get("actions") or [] ], + # 将校验错误转成字典，保留错误类型、动作下标和具体原因 "validation_errors": [ error.model_dump() for error in state["validation_errors"] ], + # 如果已有修正建议，一并提供给模型，帮助分析上次修正后仍失败的原因 "previous_repair_plan": state["repair_plan"].model_dump() if state.get("repair_plan") else None, } + # 如果动作来自大屏规划，补充规划和新增组件物料，避免修正时偏离规划或物料定义 if state.get("screen_plan") is not None: payload["screen_plan"] = state["screen_plan"].model_dump(mode="json") payload["component_materials"] = plan_materials(state) + # 只捕获输出结构和解析错误，配置、网络等异常继续向上抛出 try: + # 异步生成修正建议，本节点不直接修改动作 result = await llm_client.ainvoke( [ + # 系统消息提供稳定的失败分析和修正规则 SystemMessage(content=REFLECTION_SYSTEM_PROMPT), + # 用户消息承载本轮待分析的具体资料，序列化时保留中文 HumanMessage(content=json.dumps(payload, ensure_ascii=False)), ] ) + # 将模型对象转成字典后统一校验，非模型返回值直接交给协议校验 plan = ActionRepairPlan.model_validate( result.model_dump() if isinstance(result, BaseModel) else result ) except (ValidationError, OutputParserException): + # 如果未能得到合法建议，清空旧修正计划，避免再次使用已失效的建议 updated_state["repair_plan"] = None + # 记录无法继续修正的原因，交给后续路由和失败处理流程 updated_state["failure_reason"] = "未能生成有效的动作修正建议" + return updated_state + # 保存通过校验的修正建议，供后续路由判断及动作重新生成使用 updated_state["repair_plan"] = plan + # 仅在建议允许重试时将次数加一，缺少历史计数时从零开始，不可重试时保持原次数 updated_state["retry_count"] = state.get("retry_count", 0) + int(plan.retryable) + # 可以重试时清空旧失败原因，否则使用建议中的原因说明为何停止修正 updated_state["failure_reason"] = None if plan.retryable else plan.reason + + # 返回修正建议、重试次数和失败原因，由后续路由决定重新生成动作还是结束 return updated_state diff --git a/app/processor/nodes/requirement_node.py b/app/processor/nodes/requirement_node.py index 1c317e4..5d89585 100644 --- a/app/processor/nodes/requirement_node.py +++ b/app/processor/nodes/requirement_node.py @@ -15,28 +15,41 @@ class RequirementNode(BaseNode): name = "requirement_node" async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + # 获取输入节点已校验的上下文，包含用户需求、当前大屏和澄清记录 context = state["input_context"] + # 获取意图节点经过置信度判断的最终意图，本节点沿用该结果 intent = state["intent"] llm_client = get_llm(Requirement) + # 异步调用模型，结合当前上下文提取目标、变更、约束及需要澄清的信息 llm_res = await llm_client.ainvoke( [ + # 系统消息提供稳定的需求提取规则和输出边界 SystemMessage(content=REQUIREMENT_SYSTEM_PROMPT), + # 用户消息承载本次需求、大屏、澄清记录和已确认的意图等具体资料 HumanMessage(content=build_requirement_input(context, intent)), ] ) - # 沿用门控后的意图，合并后重新校验字段一致性 + # 用上游意图覆盖模型返回的 intent，避免需求提取阶段重新分类 + # 覆盖后重新校验字段一致性，例如 unclear 必须要求澄清，qa 不能包含变更 requirement = Requirement.model_validate( {**llm_res.model_dump(), "intent": intent} ) + # 收集当前大屏中真实存在的节点 ID，用于校验模型提取的变更目标 node_ids = {node.id for node in context.screen.nodes} + # 逐项检查需求变更，防止后续节点基于不存在的组件生成方案或动作 for change in requirement.changes: + # 用集合差集找出不在当前大屏中的目标 ID，非空时保存到 unknown_ids if unknown_ids := set(change.target_ids) - node_ids: + # 如果存在无效目标，立即抛出异常，并排序输出 ID 以便定位问题 raise ValueError(f"需求引用了不存在的节点: {sorted(unknown_ids)}") + # 保存通过结构及目标校验的需求，供后续澄清或范围判断使用 updated_state["requirement"] = requirement + # 返回更新后的状态，由图继续执行后续节点 return updated_state @@ -72,14 +85,19 @@ if __name__ == "__main__": } # 2. 依次执行输入、意图判断和需求理解 try: + # 先将原始请求校验为统一的输入上下文 state = InputNode().process(state) + # 调用 Jev 判断意图，为需求提取提供已确认的分类结果 state = await IntentNode().process(state) + # 基于上下文和意图调用模型，提取结构化需求 node = RequirementNode() final_state = await node.process(state) finally: + # 无论处理成功还是抛出异常，都关闭示例使用的 Jev 连接 await close_decision_engine() # 3. 打印完整 state + # 将状态中的 Pydantic 模型转成字典，其他值保持原样，以便输出 JSON print( json.dumps( { @@ -93,4 +111,5 @@ if __name__ == "__main__": ) ) + # 启动事件循环，执行上面的异步调试流程 asyncio.run(main()) diff --git a/app/processor/nodes/scope_node.py b/app/processor/nodes/scope_node.py index 38506e2..a1a34bc 100644 --- a/app/processor/nodes/scope_node.py +++ b/app/processor/nodes/scope_node.py @@ -16,20 +16,28 @@ class ScopeNode(BaseNode): name = "scope_node" async def process(self, state: QueryState) -> QueryState: + # 浅拷贝状态 updated_state: QueryState = state.copy() + # 获取需求节点提取的结构化需求，作为范围判断的依据 requirement = state["requirement"] + # 如果需求仍需澄清，或意图不是新建、修改大屏，则不能进入范围判断 if requirement.needs_clarification or requirement.intent not in ( "create_screen", "modify_screen", ): raise ValueError("ScopeNode 只处理信息完整的新建或修改需求") + # 获取当前大屏和用户输入，帮助模型结合已有组件判断影响范围 context = state["input_context"] + llm_client = get_llm(ChangeScope) + # 异步调用模型，判断涉及哪些组件、是否需要规划以及需要补充哪些资源 scope = await llm_client.ainvoke( [ + # 系统消息提供稳定的范围、规划和资源判断规则 SystemMessage(content=SCOPE_SYSTEM_PROMPT), + # 用户消息承载本次结构化需求和输入上下文，序列化为 JSON 供模型读取 HumanMessage( content=json.dumps( { @@ -44,23 +52,32 @@ class ScopeNode(BaseNode): ] ) + # 收集当前大屏中真实存在的节点 ID，用于检查模型是否引用了不存在的组件 existing_ids = {node.id for node in context.screen.nodes} + # 如果范围中的目标 ID 不在当前大屏中，则抛出异常并列出无效 ID if unknown_ids := set(scope.target_ids) - existing_ids: raise ValueError(f"范围引用了不存在的节点: {sorted(unknown_ids)}") + # 汇总需求中明确指定的目标 ID，使用集合去除多项变更之间重复引用的节点 required_ids = { node_id for change in requirement.changes for node_id in change.target_ids } + # 如果需求指定的节点未被纳入范围，则抛出异常，避免后续处理遗漏用户要求 if missing_ids := required_ids - set(scope.target_ids): raise ValueError(f"范围遗漏了需求中的节点: {sorted(missing_ids)}") + # 保存校验后的范围，下面的动作能力检查也需要读取该结果 updated_state["change_scope"] = scope - # 规划分支继续获取组件能力，直接操作分支提前记录缺失能力 + # 需要规划时先交给后续规划流程，不在此处判断能否直接生成动作 + # 无需规划时检查资源和操作能力，无法执行则记录阻塞原因，通过则返回 None reason = None if scope.needs_planning else action_block_reason(updated_state) + # 写入本次检查结果，没有阻塞原因时同时清空旧的失败说明 updated_state["failure_reason"] = reason + # 有阻塞原因时标记失败，否则保持运行状态，继续后续处理 updated_state["status"] = "failed" if reason else "running" + # 返回包含范围、失败原因和处理状态的更新结果 return updated_state diff --git a/app/processor/nodes/validator_node.py b/app/processor/nodes/validator_node.py index 2353585..1227347 100644 --- a/app/processor/nodes/validator_node.py +++ b/app/processor/nodes/validator_node.py @@ -14,15 +14,19 @@ 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 ] @@ -31,16 +35,21 @@ class ValidatorNode(BaseNode): } ) 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 @@ -49,6 +58,7 @@ class ValidatorNode(BaseNode): 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( @@ -59,20 +69,26 @@ class ValidatorNode(BaseNode): 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) @@ -80,7 +96,11 @@ class ValidatorNode(BaseNode): else ("nodes", action.target_id, action.type) for action in batch.actions ] + + # 去重后的实际操作集合用于检查遗漏，原 targets 列表保留动作顺序和下标 actual = set(targets) + + # 逐项找出需求未允许的目标或操作，防止模型扩大修改范围 errors = [ ActionValidationError( action_index=index, @@ -90,6 +110,8 @@ class ValidatorNode(BaseNode): for index, target in enumerate(targets) if target not in allowed ] + + # 如果需求允许的目标操作没有对应动作，记录遗漏，避免只执行部分需求 if missing := allowed - actual: errors.append( ActionValidationError( @@ -97,6 +119,8 @@ class ValidatorNode(BaseNode): message=f"动作遗漏了需求中的目标或操作: {sorted(missing)}", ) ) + + # 如果存在越界或遗漏，返回全部范围错误，并清空旧模拟结果 if errors: updated_state["validation_passed"] = False updated_state["validation_errors"] = errors @@ -105,8 +129,11 @@ class ValidatorNode(BaseNode): # 在副本上试执行，不改变用户传入的大屏，也不操作前端画布 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: # 首个失败即停止，后续动作可能依赖前序结果，不返回部分成功快照 @@ -119,21 +146,27 @@ class ValidatorNode(BaseNode): 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 diff --git a/app/resources/component.py b/app/resources/component.py index 159947a..a8c00a4 100644 --- a/app/resources/component.py +++ b/app/resources/component.py @@ -11,36 +11,50 @@ from app.schemas.component_resource import ( class ComponentProvider: def __init__(self, catalog_path: str | Path | None = None): + # 保存可选的目录文件路径，未指定时由查询方法使用项目内的默认目录 self.catalog_path = catalog_path def query(self, query: ComponentQuery) -> ComponentResult: # TODO 以后迁移数据库，保持组件查询接口不变 + # 优先使用调用方指定的路径，否则读取当前模块旁的 catalog/components.json path = ( Path(self.catalog_path) if self.catalog_path is not None else Path(__file__).resolve().parent / "catalog" / "components.json" ) + + # 如果传入的是相对路径，以项目根目录为基准，避免受启动时工作目录影响 if not path.is_absolute(): path = Path(__file__).resolve().parents[2] / path + try: # 每次读取当前文件，避免更新目录后仍使用旧缓存 + # utf-8-sig 同时兼容带 BOM 和不带 BOM 的 UTF-8 文件 content = path.read_text(encoding="utf-8-sig") except FileNotFoundError: + # 如果目录文件不存在，返回未支持状态和原因，由调用方决定失败处理方式 return ComponentResult( status="unsupported", message="尚未配置前端组件目录，请先导出真实组件定义", ) + # 文件损坏或定义不合法直接报错，不能伪装成查无结果 catalog = ComponentCatalog.model_validate_json(content) + # 如果提供关键词，去除首尾空白并统一大小写，未提供时不启用关键词筛选 keyword = ( query.keyword.strip().casefold() if query.keyword is not None else None ) + + # 收集同时满足所有已提供条件的组件，未设置任何条件时返回全量目录 matches = [] for component in catalog.components: + # 如果指定目录编号，则跳过编号不完全匹配的组件 if query.key is not None and component.key != query.key: continue + # 如果指定前端组件类型，则跳过类型不完全匹配的组件 if query.type is not None and component.type != query.type: continue + # 将编号、类型、名称、说明和标签拼成检索文本，并与关键词统一大小写 searchable = " ".join( [ component.key, @@ -50,9 +64,13 @@ class ComponentProvider: *component.tags, ] ).casefold() + # 如果设置了关键词，但检索文本中不包含该关键词，则跳过当前组件 if keyword is not None and keyword not in searchable: continue + # 所有筛选条件通过，保留完整组件定义供后续规划和动作生成使用 matches.append(component) + + # 根据是否有匹配项区分查询成功和未找到，同时保留目录版本、来源及结果说明 return ComponentResult( status="resolved" if matches else "not_found", components=matches,