From af84485e8d51223b315623f8c847f54cce72c7cc Mon Sep 17 00:00:00 2001 From: Marcus <1922576605@qq.com> Date: Tue, 29 Sep 2026 19:52:52 +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/actions/add.py | 6 +++ app/actions/base.py | 6 ++- app/actions/delete.py | 3 ++ app/actions/move.py | 8 +++ app/actions/plan.py | 102 +++++++++++++++++++++++++++++++++++ app/actions/registry.py | 6 +++ app/actions/resize.py | 9 ++++ app/actions/update.py | 15 ++++++ app/actions/update_canvas.py | 5 ++ app/processor/graph.py | 53 +++++++++++++++--- app/processor/progress.py | 18 +++++++ 11 files changed, 224 insertions(+), 7 deletions(-) diff --git a/app/actions/add.py b/app/actions/add.py index 3b867d5..21b93dd 100644 --- a/app/actions/add.py +++ b/app/actions/add.py @@ -9,11 +9,17 @@ class AddHandler(BaseActionHandler[AddAction]): def apply(self, screen: Screen, action: AddAction) -> Screen: # 重新校验节点，避免模型创建后被修改成无效内容 node = ScreenNode.model_validate(action.node.model_dump()) + + # 如果新节点 ID 与已有组件重复，拒绝新增，避免后续按 ID 查找时产生歧义 if any(existing.id == node.id for existing in screen.nodes): raise ValueError(f"新增组件 ID 已存在: {node.id}") + + # 深拷贝整个大屏，新增操作只影响返回的副本 result = screen.model_copy(deep=True) # 追加到末尾，不共享动作中的嵌套配置 result.nodes.append(node.model_copy(deep=True)) + + # 返回新增后的完整大屏，供后续动作继续模拟执行 return result diff --git a/app/actions/base.py b/app/actions/base.py index 38ecf3d..11c4310 100644 --- a/app/actions/base.py +++ b/app/actions/base.py @@ -16,9 +16,13 @@ class BaseActionHandler(ABC, Generic[ActionT]): raise NotImplementedError def copy_target(self, screen: Screen, target_id: str) -> tuple[Screen, ScreenNode]: - # 先确认目标存在，再复制，避免修改原始快照 + # 遍历节点 for index, node in enumerate(screen.nodes): + # 如果找到目标，复制完整大屏，并返回副本中同一位置的节点 if node.id == target_id: result = screen.model_copy(deep=True) + # 返回副本内的节点引用，让调用方的修改直接体现在 result 中 return result, result.nodes[index] + + # 遍历后仍未找到目标时明确报错，不能把无效动作当作执行成功 raise ValueError(f"目标组件不存在: {target_id}") diff --git a/app/actions/delete.py b/app/actions/delete.py index 87cad4c..e5d271d 100644 --- a/app/actions/delete.py +++ b/app/actions/delete.py @@ -7,6 +7,9 @@ from app.schemas.screen import Screen class DeleteHandler(BaseActionHandler[DeleteAction]): def apply(self, screen: Screen, action: DeleteAction) -> Screen: + # 确认删除目标存在，并取得与原始输入隔离的大屏副本及目标节点 result, target = self.copy_target(screen, action.target_id) + # 从副本中移除目标组件，保留其余组件及共享数据源 result.nodes = [node for node in result.nodes if node.id != target.id] + # 返回删除后的完整大屏，原始输入保持不变 return result diff --git a/app/actions/move.py b/app/actions/move.py index 19a155a..32c313e 100644 --- a/app/actions/move.py +++ b/app/actions/move.py @@ -7,12 +7,20 @@ from app.schemas.screen import Layout, Screen class MoveHandler(BaseActionHandler[MoveAction]): def apply(self, screen: Screen, action: MoveAction) -> Screen: + # 确认移动目标存在，并取得大屏副本及副本中的目标节点 result, target = self.copy_target(screen, action.target_id) + + # 将现有布局转成字典，保留尺寸和其他布局字段 layout = target.layout.model_dump() + + # 在原坐标上叠加相对位移，不将 delta_x 和 delta_y 当作绝对坐标 layout["x"] += action.delta_x layout["y"] += action.delta_y + # 重新校验计算结果，防止浮点溢出产生无效坐标 target.layout = Layout.model_validate(layout) + + # 返回移动后的大屏副本，供下一条动作继续使用 return result diff --git a/app/actions/plan.py b/app/actions/plan.py index e6b1ef7..dce7fc8 100644 --- a/app/actions/plan.py +++ b/app/actions/plan.py @@ -21,12 +21,20 @@ from app.schemas.screen_plan import PlanComponent def plan_block_reason(state: QueryState) -> str | None: """检查规划是否在当前支持范围内，不支持时返回原因""" + + # 获取当前规划和可选的组件目录，后续分别检查操作范围与新增物料能力 plan = state["screen_plan"] resources = state.get("component_resources") + + # 如果需求尚未明确，不能根据规划直接生成动作 if state["requirement"].needs_clarification: return "需求尚未澄清，不能生成规划动作" + + # 如果规划仍依赖待获取的资源，先阻止动作生成 if plan.resource_needs: return "规划仍有待获取资源，不能直接生成动作" + + # 分别收集新增、删除和修改目标，保留组件不算可执行变更 additions = [item for item in plan.components if item.disposition == "add"] removals = { item.target_id for item in plan.components if item.disposition == "remove" @@ -34,24 +42,38 @@ def plan_block_reason(state: QueryState) -> str | None: modifications = { item.target_id: item for item in plan.components if item.disposition == "modify" } + + # 如果没有任何组件或画布变更，则没有可转换为动作的规划内容 if not additions and not modifications and not removals and not plan.canvas_changes: return "规划未包含可执行的组件或画布修改" + + # 新增组件依赖真实物料目录，仅调整已有组件时不强制要求目录存在 if additions and (resources is None or resources.status != "resolved"): return "组件目录尚未就绪" + + # 收集目录支持的类型，拒绝规划虚构前端未注册的新增组件类型 supported = {item.type for item in resources.components} if resources else set() if any(item.component_type not in supported for item in additions): return "规划中的新增类型未在组件目录中找到" + + # 按 ID 索引原大屏节点，检查已有目标是否存在及类型是否保持一致 nodes = {node.id: node for node in state["input_context"].screen.nodes} for item in plan.components: + # 新增组件没有已有目标，不参与下面的已有节点检查 if item.disposition == "add": continue + + # 保留、修改和删除都必须指向真实存在的节点 if item.target_id not in nodes: return f"规划目标不存在: {item.target_id}" + + # 如果规划显式指定类型，必须与已有节点相同，不能借修改操作更换类型 if ( item.component_type is not None and item.component_type != nodes[item.target_id].type ): return "当前不支持更换已有组件类型" + # 已有目标的修改诉求不能通过 keep 或遗漏目标绕过 supported_operations = {"move", "resize", "update", "restyle", "reflow", "layout"} changes = state["requirement"].changes @@ -64,10 +86,13 @@ def plan_block_reason(state: QueryState) -> str | None: for target in change.target_ids } + # 规划删除集合必须与需求完全一致，既不能多删，也不能漏删 if removals != allowed_removals: return "规划删除目标与需求不一致，不能遗漏或扩大删除范围" + # 逐项核对需求是否被规划覆盖，任一不支持的需求都会阻止整批生成 for change in changes: + # 画布更新需求必须对应非空的画布变更规划 if change.target_scope == "canvas" and change.operation in { "update", "resize", @@ -76,14 +101,17 @@ def plan_block_reason(state: QueryState) -> str | None: if not plan.canvas_changes: return "规划遗漏画布修改需求" elif change.target_scope == "screen" and change.operation == "add": + # 新增需求至少需要一个新增组件规划 if not additions: return "规划遗漏新增需求" elif change.target_scope == "nodes" and change.operation == "delete": + # 删除目标已在上面通过集合相等检查，此处不重复处理 continue elif ( change.target_scope in {"nodes", "screen"} and change.operation in supported_operations ): + # 调整需求必须有修改规划，并覆盖需求中明确指定的全部目标 if not modifications or any( target not in modifications for target in change.target_ids ): @@ -91,6 +119,7 @@ def plan_block_reason(state: QueryState) -> str | None: else: return "需求包含未支持的操作，当前不能只执行部分规划" + # 没有整体调整需求时，已有组件的修改目标只能来自需求明确列出的 ID if ( modifications and changes @@ -99,21 +128,30 @@ def plan_block_reason(state: QueryState) -> str | None: for change in changes ) ): + # 汇总需求目标，检查规划是否额外修改了其他组件 allowed = {target for change in changes for target in change.target_ids} if set(modifications) - allowed: return "规划调整了需求范围外的组件" + + # 所有能力和范围检查通过，None 表示没有阻塞原因 return None def plan_materials(state: QueryState) -> list[dict]: """提取规划需要的完整物料配置，供动作生成和修正使用""" + + # 只收集新增组件所需的类型，使用集合去除重复类型 types = { item.component_type for item in state["screen_plan"].components if item.disposition == "add" } + + # 如果没有新增组件，就不需要向模型提供物料配置 if not types: return [] + + # 按类型提取全部匹配物料，使用协议别名输出 schema 等字段 return [ item.model_dump(mode="json", by_alias=True) for item in state["component_resources"].components @@ -124,31 +162,45 @@ def plan_materials(state: QueryState) -> list[dict]: def merge_config(defaults: dict, overrides: dict) -> dict: # 对象逐层覆盖，数组和标量整体替换，不修改目录或模型输出 result = deepcopy(defaults) + + # 逐项合并覆盖值，双方都是字典时递归，否则用新值的深拷贝整体替换 for key, value in overrides.items(): result[key] = ( merge_config(result[key], value) if isinstance(result.get(key), dict) and isinstance(value, dict) else deepcopy(value) ) + + # 返回独立配置，保留未被覆盖的默认字段 return result def assemble_plan_actions(state: QueryState, batch: ActionBatch) -> ActionBatch: """用物料默认配置补全新增节点，模型显式提供的字段优先""" + + # 按物料编号建立索引，动作通过 component_key 选择具体预设 resources = state.get("component_resources") materials = {item.key: item for item in resources.components} if resources else {} + + # 复制整批动作，补全配置时不改动调用方提供的动作对象 result = batch.model_copy(deep=True) for action in result.actions: + # 非新增动作或找不到物料时不补全，物料合法性由规划动作校验负责 if not isinstance(action, AddAction) or action.component_key not in materials: continue + + # 获取所选物料，仅用模型显式提供的节点字段覆盖默认配置 material = materials[action.component_key] node = merge_config( material.schema_.model_dump(mode="json"), action.node.model_dump(mode="json", exclude_unset=True), ) + + # 对合并后的完整节点重新校验，防止默认配置与覆盖值组合后不合法 action.node = ScreenNode.model_validate(node) # 保守标记物料含示例数据，不能由模型声称已取得真实数据 action.sample_data = material.sample_data + return result @@ -156,9 +208,14 @@ def _validate_canvas_action( action: UpdateCanvasAction, planned_fields: set[str], covered_fields: set[str] ) -> str | None: """检查画布字段范围，通过后记录已覆盖字段，最终值在执行后检查""" + + # 找出动作包含但规划未允许的顶层画布字段 if extra := set(action.properties) - planned_fields: return f"不能修改规划外的画布字段: {sorted(extra)}" + + # 累计本动作覆盖的字段，供整批检查是否遗漏规划内容 covered_fields.update(action.properties) + return None @@ -166,11 +223,17 @@ def _validate_delete_action( action: DeleteAction, planned_ids: set[str], removed_ids: set[str] ) -> str | None: """只允许删除 remove 目标，通过后记录目标以防重复删除""" + # 如果目标未被规划标记为删除，拒绝该动作 if action.target_id not in planned_ids: return f"不能删除非 remove 规划目标: {action.target_id}" + + # 如果前面的动作已经删除该目标，拒绝重复删除 if action.target_id in removed_ids: return f"不能重复删除组件: {action.target_id}" + + # 记录本次已覆盖的删除目标，同时用于后续重复和遗漏检查 removed_ids.add(action.target_id) + return None @@ -180,11 +243,17 @@ def _validate_modify_action( modified_ids: set[str], ) -> str | None: """只允许调整 modify 目标，属性更新还需规划中有设计要求""" + # 修改目标必须在 modify 规划中，不能修改 keep 或未提及的组件 if action.target_id not in planned: return f"不能修改保留或未提及的组件: {action.target_id}" + + # 属性更新还需要明确的设计内容，单纯布局规划不能授权任意属性修改 if isinstance(action, UpdateAction) and not planned[action.target_id].design: return f"规划未要求修改属性: {action.target_id}" + + # 记录已覆盖的修改目标，同一组件可以由多条移动、尺寸或属性动作共同完成 modified_ids.add(action.target_id) + return None @@ -196,29 +265,43 @@ def _validate_add_action( seen_plan_ids: set[str], ) -> str | None: """检查新增规划关联、ID、物料类型、布局及数据标记""" + # 新增动作必须关联有效的新增规划编号 if action.plan_id not in planned: return f"动作未关联有效的新增规划: {action.plan_id}" + + # 同一新增规划只能生成一个组件，防止重复新增 if action.plan_id in seen_plan_ids: return f"同一规划组件不能重复新增: {action.plan_id}" + + # 新组件不能复用原大屏已有的节点 ID if action.node.id in existing_ids: return f"新增组件不能复用原有 ID: {action.node.id}" # 关联有效即记录出现，后续配置错误单独报告，不再重复报遗漏 seen_plan_ids.add(action.plan_id) + + # 获取关联规划与具体物料，核对三者声明的组件类型 item = planned[action.plan_id] material = materials.get(action.component_key) + if ( material is None or material.type != item.component_type or action.node.type != item.component_type ): return "新增节点类型、物料和规划类型不一致" + # 如果规划已明确新增布局，动作必须完整沿用该布局 if item.layout is not None and action.node.layout != item.layout: return f"新增布局与规划不一致: {item.plan_id}" + + # 示例数据标记必须与物料一致，不能将示例内容标记为真实业务数据 if action.sample_data != material.sample_data: return "示例数据标记与物料不一致" + + # 当前没有数据绑定能力，拒绝新增节点携带非空的扩展 dataId if action.node.model_extra.get("dataId"): return "当前尚未接入数据绑定，不能生成 dataId" + return None @@ -229,34 +312,41 @@ def validate_plan_actions( # 先检查规划是否可执行，不支持时直接返回原因 if reason := plan_block_reason(state): return [ActionValidationError(code="scope_error", message=reason)] + # 新增动作通过 plan_id 查找对应规划 expected = { item.plan_id: item for item in state["screen_plan"].components if item.disposition == "add" } + # 修改动作通过已有组件 ID 查找对应规划 modifications = { item.target_id: item for item in state["screen_plan"].components if item.disposition == "modify" } + # 物料按 key 索引，用于核对新增组件的类型和示例数据标记 resources = state.get("component_resources") materials = {item.key: item for item in resources.components} if resources else {} seen: set[str] = set() # 已关联的新增规划编号 modified: set[str] = set() # 已覆盖的修改目标 + # 只有规划标为 remove 的组件才允许删除 removals = { item.target_id for item in state["screen_plan"].components if item.disposition == "remove" } + removed: set[str] = set() # 已覆盖的删除目标 canvas_fields: set[str] = set() # 已覆盖的画布字段 planned_canvas = state[ "screen_plan" ].canvas_changes # 规划要求修改的画布字段及最终值 + + # 累计所有动作及覆盖范围错误，便于一次性反馈给修正流程 errors = [] # 保存原有组件 ID，防止新增组件复用已有 ID @@ -304,6 +394,8 @@ def validate_plan_actions( code="scope_error", message=f"遗漏{label}: {sorted(missing)}" ) ) + + # 返回全部校验问题，空列表表示动作范围和覆盖检查通过 return errors @@ -311,7 +403,10 @@ def validate_plan_result( state: QueryState, screen: Screen ) -> list[ActionValidationError]: """检查整批执行后的组件布局和画布属性是否达到规划要求""" + + # 按节点 ID 索引模拟执行结果，便于读取每个修改目标的最终布局 nodes = {node.id: node for node in screen.nodes} + # 只核对明确提供布局的修改规划，收集最终布局未达到要求的目标 errors = [ ActionValidationError( code="scope_error", message=f"执行后的布局与规划不一致: {item.target_id}" @@ -320,6 +415,7 @@ def validate_plan_result( if item.disposition == "modify" and item.layout is not None and not all( + # 坐标和尺寸允许极小的浮点误差，避免加减计算造成误报 isclose( getattr(nodes[item.target_id].layout, key), value, @@ -327,16 +423,22 @@ def validate_plan_result( abs_tol=1e-9, ) if key in {"x", "y", "width", "height"} + # 其他布局字段按值精确比较，不使用数值容差 else nodes[item.target_id].layout.model_dump().get(key) == value for key, value in item.layout.model_dump().items() ) ] + + # 将执行后的画布转成字典，逐项核对规划要求的最终属性值 canvas = screen.canvas.model_dump() for key, value in state["screen_plan"].canvas_changes.items(): + # 字段缺失或最终值不一致都视为未达到规划要求 if key not in canvas or canvas[key] != value: errors.append( ActionValidationError( code="scope_error", message=f"执行后的画布属性与规划不一致: {key}" ) ) + + # 返回布局及画布结果错误，空列表表示这些最终值符合规划 return errors diff --git a/app/actions/registry.py b/app/actions/registry.py index e0dc44a..1ab2323 100644 --- a/app/actions/registry.py +++ b/app/actions/registry.py @@ -25,7 +25,13 @@ ACTION_HANDLERS: dict[str, BaseActionHandler[Any]] = { def apply_action(screen: Screen, action: Action) -> Screen: """接收已通过 Schema 校验的动作，返回修改后的大屏副本，不保存或更新前端""" + + # 根据动作协议中的 type 查找对应处理器 handler = ACTION_HANDLERS.get(action.type) + + # 如果类型尚未注册，明确报错，避免静默跳过动作 if handler is None: raise ValueError(f"尚未实现动作处理器: {action.type}") + + # 委托处理器校验并修改副本，执行异常交由调用方统一处理 return handler.apply(screen, action) diff --git a/app/actions/resize.py b/app/actions/resize.py index bb8a2a2..6728f49 100644 --- a/app/actions/resize.py +++ b/app/actions/resize.py @@ -7,14 +7,23 @@ from app.schemas.screen import Layout, Screen class ResizeHandler(BaseActionHandler[ResizeAction]): def apply(self, screen: Screen, action: ResizeAction) -> Screen: + # 确认目标存在，并取得大屏副本及副本中的目标节点 result, target = self.copy_target(screen, action.target_id) + # 将原布局转成字典，便于替换尺寸后重新校验 layout = target.layout.model_dump() + # 仅替换提供的尺寸，位置和其他布局字段保持不变 if action.width is not None: layout["width"] = action.width + + # 如果提供了高度，则替换为指定值，否则沿用原高度 if action.height is not None: layout["height"] = action.height + + # 重新校验完整布局，确保更新后的宽高等字段仍满足协议约束 target.layout = Layout.model_validate(layout) + + # 返回调整尺寸后的大屏副本 return result diff --git a/app/actions/update.py b/app/actions/update.py index 19ac9ac..a445625 100644 --- a/app/actions/update.py +++ b/app/actions/update.py @@ -9,32 +9,47 @@ from app.schemas.screen import Screen class UpdateHandler(BaseActionHandler[UpdateAction]): def apply(self, screen: Screen, action: UpdateAction) -> Screen: + # 确认目标存在，后续路径访问和赋值均在大屏副本中进行 result, target = self.copy_target(screen, action.target_id) + # 未设置的 style 视为空对象，只允许直接添加末级字段 if action.section == "style" and target.style is None: target.style = {} + + # 获取动作指定的属性区域，作为路径遍历的起点 current = getattr(target, action.section) + # 逐段访问路径，只有到达末级时才写入新值 for index, segment in enumerate(action.path): + # 标记当前是否为末级，用于区分继续访问和最终赋值 last = index == len(action.path) - 1 + # 保存走到当前段的路径，异常时指出具体失败位置 location = f"{action.section} {action.path[: index + 1]!r}" + # 如果当前容器是字典，路径段必须是字符串键 if isinstance(current, dict): if not isinstance(segment, str): raise ValueError(f"对象路径必须使用字符串键: {location}") + # 中间层必须已存在，只允许在末级新增键，避免凭空构造嵌套结构 if not last and segment not in current: raise ValueError(f"中间路径不存在: {location}") elif isinstance(current, list): + # 数组只接受整数下标，使用精确类型判断排除 bool if type(segment) is not int: raise ValueError(f"数组路径必须使用整数下标: {location}") + # 只允许访问已有元素，不接受负下标，也不通过赋值扩展数组 if not 0 <= segment < len(current): raise ValueError(f"数组下标越界: {location}") else: + # 标量或 None 不能继续按路径访问，停止执行并返回明确错误 raise ValueError(f"路径无法继续访问，当前值不是对象或数组: {location}") if last: # 新值也复制，避免后续修改结果时影响 Action current[segment] = deepcopy(action.value) else: + # 尚未到达末级，进入已校验存在的下一层容器 current = current[segment] + + # 返回属性更新后的大屏副本，原始大屏和动作值均未被修改 return result diff --git a/app/actions/update_canvas.py b/app/actions/update_canvas.py index af11d79..bec2d97 100644 --- a/app/actions/update_canvas.py +++ b/app/actions/update_canvas.py @@ -9,12 +9,17 @@ from app.schemas.screen import Canvas, Screen class UpdateCanvasHandler(BaseActionHandler[UpdateCanvasAction]): def apply(self, screen: Screen, action: UpdateCanvasAction) -> Screen: + # 将现有画布转成字典，保留动作没有指定的字段 properties = screen.canvas.model_dump() # 顶层字段整体替换，嵌套对象不做深度合并 properties.update(deepcopy(action.properties)) + # 校验合并后的画布属性，尺寸等字段不合法时直接抛出异常 canvas = Canvas.model_validate(properties) + # 深拷贝大屏后替换画布，避免修改原始快照 result = screen.model_copy(deep=True) result.canvas = canvas + + # 返回包含新画布属性的完整大屏副本 return result diff --git a/app/processor/graph.py b/app/processor/graph.py index cc84aa5..117d04e 100644 --- a/app/processor/graph.py +++ b/app/processor/graph.py @@ -48,10 +48,11 @@ def route_after_scope(state: QueryState) -> str: def build_graph(checkpointer=None): from app.config import settings + # 最大重试次数 max_retries = settings.max_action_retries def route_failure(state: QueryState) -> str: - # 上限检查放在 Reflection 前，避免最后一次无效模型调用 + # 上限检查，放在 Reflection 前，避免最后一次无效模型调用 return ( "reflection_node" if state.get("retry_count", 0) < max_retries @@ -59,8 +60,13 @@ def build_graph(checkpointer=None): ) def route_after_action(state: QueryState) -> str: + # 如果动作节点已明确标记失败，直接收尾，不再进入内容修正 if state.get("status") == "failed": return "fallback_node" + + # validation_passed 为 False：动作生成时解析或结构校验失败，未达重试上限则进入修正分析，已达上限则失败收尾 + # validation_passed 为 None：动作已生成但尚未模拟执行，进入 validator_node 检查动作范围和执行结果 + # 必须用 is False 区分失败与尚未校验，不能将 None 也当作失败 return ( route_failure(state) if state["validation_passed"] is False @@ -68,16 +74,20 @@ def build_graph(checkpointer=None): ) def route_after_validation(state: QueryState) -> str: + # 整批动作通过校验则结束，否则根据剩余重试次数决定修正或失败收尾 return END if state["validation_passed"] else route_failure(state) def route_after_reflection(state: QueryState) -> str: + # 获取失败分析产生的修正建议，生成建议失败时该字段可能为空 plan = state.get("repair_plan") + + # 只有存在建议且明确允许重试时才重新生成动作，其余情况统一失败收尾 return "action_node" if plan is not None and plan.retryable else "fallback_node" - # 定义工作流 + # 使用 QueryState 定义节点之间共享的状态结构 workflow = StateGraph(QueryState) - # 实例化节点 + # 为本次构建的图创建节点实例，字典键同时作为注册名称和路由目标 nodes = { "input_node": InputNode(), "intent_node": IntentNode(), @@ -96,14 +106,17 @@ def build_graph(checkpointer=None): # 注册节点，异步节点使用 BaseNode 的异步入口 for node_name, node in nodes.items(): + # 根据 process 是否为异步函数选择调用入口，保留 BaseNode 的统一执行封装 action = node._acall if inspect.iscoroutinefunction(node.process) else node + # 将节点名称与实际调用对象绑定，供后面的边引用 workflow.add_node(node_name, action) - # 定义入口 + # 首次执行从输入节点开始，完成上下文校验后再判断意图 workflow.set_entry_point("input_node") - # 添加业务边 + # 输入处理完成后固定进入意图判断 workflow.add_edge("input_node", "intent_node") + # 新建和修改进入需求理解，意图不明进入澄清 workflow.add_conditional_edges( source="intent_node", @@ -115,13 +128,20 @@ def build_graph(checkpointer=None): "unclear": "clarify_node", }, ) + + # 需求信息不足时提问，信息完整时继续判断修改范围 workflow.add_conditional_edges( source="requirement_node", path=route_after_requirement, path_map={True: "clarify_node", False: "scope_node"}, ) + + # 先保存澄清问题和等待状态，再进入调用 interrupt 的节点暂停执行 workflow.add_edge("clarify_node", "wait_for_user_node") + + # 问答生成文本后直接结束，不进入规划和动作流程 workflow.add_edge("qa_node", END) + # 先读取组件能力再规划，规划后的业务数据查询后续接入 workflow.add_conditional_edges( "scope_node", @@ -132,12 +152,18 @@ def build_graph(checkpointer=None): "fallback_node": "fallback_node", }, ) + + # 组件目录查询成功才允许规划，未找到或不支持时进入失败收尾 workflow.add_conditional_edges( "component_resource_node", lambda state: state["component_resources"].status == "resolved", {True: "planner_node", False: "fallback_node"}, ) + + # 规划完成后交给动作节点转换，动作节点还会检查规划是否可执行 workflow.add_edge("planner_node", "action_node") + + # 动作生成后按结果进入校验、修正分析或失败收尾 workflow.add_conditional_edges( "action_node", route_after_action, @@ -147,6 +173,8 @@ def build_graph(checkpointer=None): "fallback_node": "fallback_node", }, ) + + # 校验成功结束，失败则在重试上限内分析修正原因 workflow.add_conditional_edges( "validator_node", route_after_validation, @@ -156,6 +184,8 @@ def build_graph(checkpointer=None): "fallback_node": "fallback_node", }, ) + + # 修正分析允许重试时回到动作生成，形成生成、校验、分析的循环 workflow.add_conditional_edges( "reflection_node", route_after_reflection, @@ -164,14 +194,21 @@ def build_graph(checkpointer=None): "fallback_node": "fallback_node", }, ) + + # 失败节点统一写入最终失败状态后结束流程 workflow.add_edge("fallback_node", END) + + # 用户补充后重新合并输入并判断意图，取消则直接结束 workflow.add_conditional_edges( "wait_for_user_node", route_after_wait, {"input_node": "input_node", END: END}, ) + # TODO 本地默认保存在内存，可传入持久化 checkpointer + # 仅在调用方未提供检查点存储时创建默认实现，外部传入的存储直接复用 if checkpointer is None: + # 允许恢复状态中使用的业务模型类型，使检查点读取后仍可按模型对象访问字段 serde = JsonPlusSerializer( allowed_msgpack_modules=[ ("app.schemas.input_context", "InputContext"), @@ -193,16 +230,20 @@ def build_graph(checkpointer=None): ("app.schemas.clarification_reply", "ClarificationReply"), ] ) + + # 将检查点保存在当前进程内存中，支持暂停恢复，但进程退出后不会保留 checkpointer = InMemorySaver(serde=serde) + # 编译节点和边，并绑定检查点存储，供执行时按 thread_id 保存和恢复状态 return workflow.compile(checkpointer=checkpointer).with_config( { + # 随修正上限增加图执行步数预算，每轮预留分析、生成、校验三步，至少允许 25 步 "recursion_limit": max(25, 12 + 3 * max_retries), } ) -# 创建全局图实例 +# 模块加载时创建共享图实例，复用该实例的调用也共用其检查点存储 query_app = build_graph() diff --git a/app/processor/progress.py b/app/processor/progress.py index b85b18d..f02cdc6 100644 --- a/app/processor/progress.py +++ b/app/processor/progress.py @@ -11,6 +11,7 @@ from langgraph.errors import GraphInterrupt from app.schemas.node_progress import NodeProgress, NodeProgressStatus +# 将内部节点名映射为用户能理解的进度说明，未配置的节点仍发送事件但不附带 label NODE_PROGRESS_MESSAGES = { "requirement_node": "正在理解你的需求", "qa_node": "正在整理回答", @@ -25,42 +26,58 @@ NODE_PROGRESS_MESSAGES = { @contextmanager def node_progress(node: str, state): """每次调用独立计时和编号，同步与异步入口共用""" + try: + # 读取当前图执行配置，用于判断是否处于图上下文，并获取任务标识 config = get_config() except RuntimeError: # 独立调用节点没有图上下文，直接执行业务逻辑 + # 将控制权交给 with 块中的节点逻辑，此分支不发送进度事件 yield + # 节点逻辑正常执行后直接退出，不再获取图的流写入器 return + # 获取当前图的 custom 流写入器，供本次节点执行推送进度事件 writer = get_stream_writer() + # 状态支持键值读取时用于提取标识，其他类型按空字典处理，避免调用 get 出错 values = state if isinstance(state, Mapping) else {} # 以实际检查点配置为准，兼容没有在 state 中填写 ID 的本地示例 thread_id = config.get("configurable", {}).get("thread_id") or values.get( "thread_id" ) + # 为本次节点调用生成唯一编号，区分同一节点的首次执行、重试和恢复执行 execution_id = str(uuid4()) + # 使用单调计时器记录起点，后续计算本次执行耗时 started = perf_counter() def emit(status: NodeProgressStatus): + # 将任务、用户消息和本次执行编号放入同一事件，便于消费方关联进度 event = NodeProgress( thread_id=thread_id, message_id=values.get("message_id"), execution_id=execution_id, node=node, + # 未配置用户说明时返回 None，序列化时省略该字段 label=NODE_PROGRESS_MESSAGES.get(node), status=status, + # 开始事件不携带耗时，其他状态携带从本次调用开始计算的毫秒数 duration_ms=None if status == "started" else round((perf_counter() - started) * 1000, 3), ) + + # 转成可序列化的字典并发送到 custom 流，省略空字段，不写入业务状态 writer(event.model_dump(mode="json", exclude_none=True)) + # 节点业务逻辑开始前，先通知消费方本次执行已经启动 emit("started") try: + # 执行 with 块中的节点逻辑，节点抛出的异常会回到下面对应的异常分支 yield except GraphInterrupt: # interrupt 是正常暂停，恢复后会产生新的执行 ID emit("waiting_for_input") + # 将暂停信号重新抛给 LangGraph，由图负责保存检查点和中断执行 raise except asyncio.CancelledError: # 协程被取消也要结束这次节点执行，继续向上传递取消信号 @@ -72,4 +89,5 @@ def node_progress(node: str, state): raise else: # 校验不通过等业务结果仍属于节点正常执行完成 + # completed 仅表示本次节点调用未抛异常，不代表整个任务成功或前端已执行 emit("completed")