diff --git a/app/core/exceptions.py b/app/core/exceptions.py index de41f55..46940c2 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -17,8 +17,6 @@ class BizException(AppError): """可安全返回给调用方的业务错误""" def __init__(self, code: int, msg: str = "请求失败") -> None: - if not 400 <= code <= 599: - raise ValueError("业务错误码必须是 400 到 599 的 HTTP 状态码") self.code = code self.msg = msg super().__init__(msg) diff --git a/app/processor/base.py b/app/processor/base.py index 6fb9e54..c1a2996 100644 --- a/app/processor/base.py +++ b/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}]" diff --git a/app/processor/graph.py b/app/processor/graph.py index bf1772f..cd27a03 100644 --- a/app/processor/graph.py +++ b/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( diff --git a/app/processor/nodes/planner_node.py b/app/processor/nodes/planner_node.py index e8ca0ff..3119701 100644 --- a/app/processor/nodes/planner_node.py +++ b/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 diff --git a/app/processor/nodes/requirement_node.py b/app/processor/nodes/requirement_node.py index 5d89585..2bda1ea 100644 --- a/app/processor/nodes/requirement_node.py +++ b/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() # 获取输入节点已校验的上下文，包含用户需求、当前大屏和澄清记录 diff --git a/app/processor/nodes/scope_node.py b/app/processor/nodes/scope_node.py index a99a59c..1ddc089 100644 --- a/app/processor/nodes/scope_node.py +++ b/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() # 获取需求节点提取的结构化需求，作为范围判断的依据 diff --git a/app/processor/progress.py b/app/processor/progress.py index f02cdc6..a0df689 100644 --- a/app/processor/progress.py +++ b/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，序列化时省略该字段 diff --git a/app/services/agent.py b/app/services/agent.py index 18bb9e6..1e0cb10 100644 --- a/app/services/agent.py +++ b/app/services/agent.py @@ -101,16 +101,15 @@ class AgentService: @staticmethod def _result(state, thread_id): waiting = bool(state.get("__interrupt__")) - # 只交付校验通过的完整动作批次，失败草稿留在后端 + deliver = state["status"] == "completed" and state.get("validation_passed") + return AgentResult( conversation_id=state["conversation_id"], thread_id=thread_id, message_id=state["message_id"], status="waiting_for_input" if waiting else state["status"], - actions=state.get("actions") or [] - if state.get("validation_passed") is True and state["status"] == "completed" - else [], + actions=state["actions"] if deliver else [], answer=state.get("answer"), clarification=state.get("clarification") if waiting else None, failure_reason=state.get("failure_reason"),