feat：优化节点错误处理

app/processor/base.py

@@ -54,6 +54,11 @@ class BaseNode(ABC):
     def process(self, state: T) -> T:
         """子类实现节点核心逻辑，返回状态更新"""
 
+    def fail_on_invalid_output(self, state: T, error: ValueError, reason: str) -> T:
+        """模型输出未通过校验时收尾为失败结果，详细原因只写日志"""
+        self.logger.warning(f"{reason}: {error}")
+        return {**state, "status": "failed", "failure_reason": reason}
+
     def log_step(self, step_name: str, message: str = ""):
         """记录步骤日志。"""
         log_msg = f"[{step_name}]"

app/processor/graph.py

@@ -28,9 +28,11 @@ def route_after_intent(state: QueryState) -> str:
     return state["intent"]
 
 
-def route_after_requirement(state: QueryState) -> bool:
-    # True 进入澄清，False 进入修改范围判断
-    return state["requirement"].needs_clarification
+def route_after_requirement(state: QueryState) -> str:
+    # 需求输出校验失败直接收尾；需要澄清进入澄清，否则进入修改范围判断
+    if state["status"] == "failed":
+        return "fallback_node"
+    return "clarify_node" if state["requirement"].needs_clarification else "scope_node"
 
 
 def route_after_wait(state: QueryState) -> str:
@@ -38,7 +40,15 @@ def route_after_wait(state: QueryState) -> str:
     return "input_node" if state["status"] == "running" else END
 
 
+# TODO 目前“资源”只有物料目录，数据源和 RAG 尚未准备好
+# 接入后把 needs_planning 与 resource_needs 拆成两个独立开关：
+#   规划 -> 按需查资源 -> planner；无需规划但有 resource_needs -> 查资源 -> action
+# 并让 component_resource_node 按 resource_needs 决定查什么，而不是固定全量目录
+# 在此之前，无需规划且带 resource_needs 的请求会由 action_block_reason 判为失败
 def route_after_scope(state: QueryState) -> str:
+    # 范围输出校验失败或已判定无法执行，直接失败收尾
+    if state["status"] == "failed":
+        return "fallback_node"
     # 需要规划时先获取组件能力，供规划节点选择可用组件
     if state["change_scope"].needs_planning:
         return "component_resource_node"
@@ -132,7 +142,11 @@ def build_graph(checkpointer=None):
     workflow.add_conditional_edges(
         source="requirement_node",
         path=route_after_requirement,
-        path_map={True: "clarify_node", False: "scope_node"},
+        path_map={
+            "clarify_node": "clarify_node",
+            "scope_node": "scope_node",
+            "fallback_node": "fallback_node",
+        },
     )
 
     # 先保存澄清问题和等待状态，再进入调用 interrupt 的节点暂停执行
@@ -160,7 +174,11 @@ def build_graph(checkpointer=None):
     )
 
     # 规划完成后交给动作节点转换，动作节点还会检查规划是否可执行
-    workflow.add_edge("planner_node", "action_node")
+    workflow.add_conditional_edges(
+        "planner_node",
+        lambda state: "fallback_node" if state["status"] == "failed" else "action_node",
+        {"action_node": "action_node", "fallback_node": "fallback_node"},
+    )
 
     # 动作生成后按结果进入校验、修正分析或失败收尾
     workflow.add_conditional_edges(

app/processor/nodes/planner_node.py

@@ -10,6 +10,12 @@ from app.processor.state import QueryState
 from app.prompt.planner import PLANNER_SYSTEM_PROMPT
 from app.schemas.screen_plan import ScreenPlan
 
+# 规划输出未通过校验时，带着错误让模型重新规划的最大次数
+# TODO 上游节点目前只有这里做了节点内重试，scope_node 和 requirement_node 尚未跟进
+# TODO 将来可改为图级回路：规划校验失败 -> 规划版 reflection -> planner，
+#      与动作阶段的重试统一由 route_failure 控制，届时可去掉这里的节点内重试
+PLAN_MAX_REPAIRS = 1
+
 
 class PlannerNode(BaseNode):
     name = "planner_node"
@@ -26,28 +32,64 @@ class PlannerNode(BaseNode):
         # 图保证组件目录已就绪后才进入规划
         resources = state["component_resources"]
 
+        messages = [
+            # 系统消息提供稳定的规划规则和输出边界
+            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"),
+                        # 仅提供规划需要的组件摘要，省略完整默认配置
+                        "component_resources": resources.to_planner_context(),
+                    },
+                    ensure_ascii=False,
+                )
+            ),
+        ]
         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"),
-                            # 仅提供规划需要的组件摘要，省略完整默认配置
-                            "component_resources": resources.to_planner_context(),
-                        },
-                        ensure_ascii=False,
-                    )
-                ),
-            ]
-        )
 
-        plan = result
+        for attempt in range(PLAN_MAX_REPAIRS + 1):
+            plan = None
+            try:
+                plan = await llm_client.ainvoke(messages)
+                self._check(plan, state)
+                break
+            except ValueError as error:
+                # 解析失败、引用不存在的目标等可由模型自行纠正，超过次数才失败收尾
+                if attempt == PLAN_MAX_REPAIRS:
+                    return self.fail_on_invalid_output(
+                        state, error, "未能生成可用的大屏规划，请调整需求后重试"
+                    )
+                self.logger.warning(f"规划未通过校验，重新规划: {error}")
+                feedback = f"上一次输出未通过校验：{error}\n请修正后重新输出完整规划。"
+                if plan is not None:
+                    feedback = f"上一次输出：\n{plan.model_dump_json()}\n\n{feedback}"
+                messages = [*messages, HumanMessage(content=feedback)]
+
+        # 合并需求约束和规划约束，按首次出现顺序去重，保留需求约束在前
+        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
+
+    @staticmethod
+    def _check(plan: ScreenPlan, state: QueryState) -> None:
+        """校验规划的组件类型、目标引用和修改范围，不通过时抛出 ValueError"""
+        requirement = state["requirement"]
+        scope = state["change_scope"]
+        context = state["input_context"]
+        resources = state["component_resources"]
 
         # 校验规划中的物料类型是否真实受支持
         supported_types = {component.type for component in resources.components}
@@ -97,20 +139,6 @@ class PlannerNode(BaseNode):
             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
-
 
 if __name__ == "__main__":
     import asyncio

app/processor/nodes/requirement_node.py

@@ -15,6 +15,15 @@ class RequirementNode(BaseNode):
     name = "requirement_node"
 
     async def process(self, state: QueryState) -> QueryState:
+        try:
+            return await self._generate(state)
+        except ValueError as error:
+            # 结构化输出解析失败或引用校验不通过（含 ValidationError、OutputParserException）
+            return self.fail_on_invalid_output(
+                state, error, "未能可靠理解你的需求，请换个说法再试"
+            )
+
+    async def _generate(self, state: QueryState) -> QueryState:
         # 浅拷贝状态
         updated_state: QueryState = state.copy()
         # 获取输入节点已校验的上下文，包含用户需求、当前大屏和澄清记录

app/processor/nodes/scope_node.py

@@ -16,6 +16,15 @@ class ScopeNode(BaseNode):
     name = "scope_node"
 
     async def process(self, state: QueryState) -> QueryState:
+        try:
+            return await self._generate(state)
+        except ValueError as error:
+            # 结构化输出解析失败或引用校验不通过（含 ValidationError、OutputParserException）
+            return self.fail_on_invalid_output(
+                state, error, "未能确定需要修改的范围，请描述得更具体一些"
+            )
+
+    async def _generate(self, state: QueryState) -> QueryState:
         # 浅拷贝状态
         updated_state: QueryState = state.copy()
         # 获取需求节点提取的结构化需求，作为范围判断的依据

app/processor/progress.py

@@ -1,7 +1,6 @@
 """通过 LangGraph custom 流发送进度，不额外保存任务状态"""
 
 import asyncio
-from collections.abc import Mapping
 from contextlib import contextmanager
 from time import perf_counter
 from uuid import uuid4
@@ -39,10 +38,8 @@ def node_progress(node: str, state):
 
     # 获取当前图的 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 = config.get("configurable", {}).get("thread_id") or state.get(
         "thread_id"
     )
     # 为本次节点调用生成唯一编号，区分同一节点的首次执行、重试和恢复执行
@@ -54,7 +51,7 @@ def node_progress(node: str, state):
         # 将任务、用户消息和本次执行编号放入同一事件，便于消费方关联进度
         event = NodeProgress(
             thread_id=thread_id,
-            message_id=values.get("message_id"),
+            message_id=state.get("message_id"),
             execution_id=execution_id,
             node=node,
             # 未配置用户说明时返回 None，序列化时省略该字段