diff --git a/app/actions/plan.py b/app/actions/plan.py index dce7fc8..fcb2c40 100644 --- a/app/actions/plan.py +++ b/app/actions/plan.py @@ -22,9 +22,9 @@ from app.schemas.screen_plan import PlanComponent def plan_block_reason(state: QueryState) -> str | None: """检查规划是否在当前支持范围内,不支持时返回原因""" - # 获取当前规划和可选的组件目录,后续分别检查操作范围与新增物料能力 + # 获取当前规划和组件目录,后续分别检查操作范围与新增物料能力 plan = state["screen_plan"] - resources = state.get("component_resources") + resources = state["component_resources"] # 如果需求尚未明确,不能根据规划直接生成动作 if state["requirement"].needs_clarification: @@ -47,12 +47,8 @@ def plan_block_reason(state: QueryState) -> str | None: 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() + supported = {item.type for item in resources.components} if any(item.component_type not in supported for item in additions): return "规划中的新增类型未在组件目录中找到" @@ -179,8 +175,7 @@ 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 {} + materials = {item.key: item for item in state["component_resources"].components} # 复制整批动作,补全配置时不改动调用方提供的动作对象 result = batch.model_copy(deep=True) @@ -328,8 +323,7 @@ def validate_plan_actions( } # 物料按 key 索引,用于核对新增组件的类型和示例数据标记 - resources = state.get("component_resources") - materials = {item.key: item for item in resources.components} if resources else {} + materials = {item.key: item for item in state["component_resources"].components} seen: set[str] = set() # 已关联的新增规划编号 modified: set[str] = set() # 已覆盖的修改目标 @@ -439,6 +433,6 @@ def validate_plan_result( code="scope_error", message=f"执行后的画布属性与规划不一致: {key}" ) ) - + # 返回布局及画布结果错误,空列表表示这些最终值符合规划 return errors diff --git a/app/clients/decision.py b/app/clients/decision.py index 95c5aa8..d9256f1 100644 --- a/app/clients/decision.py +++ b/app/clients/decision.py @@ -1,8 +1,6 @@ """可复用的 Jev 决策客户端;重试和 HTTP 连接管理交给官方 SDK""" from functools import cache -from math import isclose -from typing import Self from pydantic import ValidationError from typesafe_sdk import ( @@ -29,24 +27,11 @@ from app.schemas.decision import ( class DecisionEngine: - def __init__( - self, - config: Settings | None = None, - *, - client: AsyncTypeSafeClient | None = None, - ): + def __init__(self, config: Settings | None = None): self._config = config if config is not None else settings - self._client = client - self._owns_client = client is None - self._closed = False - - @property - def closed(self) -> bool: - return self._closed + self._client: AsyncTypeSafeClient | None = None def _get_client(self) -> AsyncTypeSafeClient: - if self._closed: - raise DecisionError("决策客户端已关闭", kind="closed") if self._client is None: config = self._config if not config.typesafe_api_key.strip(): @@ -69,8 +54,6 @@ class DecisionEngine: return self._client async def run(self, request: DecisionRequest) -> DecisionResponse: - # 即便模型实例在创建后被修改,也必须重新经过请求校验 - request = DecisionRequest.model_validate(request.model_dump(mode="json")) client = self._get_client() questions = { name: question.model_dump(mode="json", exclude_unset=True) @@ -136,27 +119,18 @@ class DecisionEngine: answer = result.results[name] if answer.type != question.type: raise ValueError("返回的决策类型与请求不一致") - if isinstance(question, ChoiceDecision) and isinstance( - answer, ChoiceResult + if ( + isinstance(question, ChoiceDecision) + and isinstance(answer, ChoiceResult) + and answer.choice not in question.criteria ): - if answer.choice not in question.criteria: - raise ValueError("返回了候选集合之外的选项") - if answer.probabilities.keys() != question.criteria.keys(): - raise ValueError("选项概率不完整") - if isinstance(question, ScoreDecision) and isinstance( - answer, ScoreResult + raise ValueError("返回了候选集合之外的选项") + if ( + isinstance(question, ScoreDecision) + and isinstance(answer, ScoreResult) + and not 0 <= answer.score <= len(question.criteria) - 1 ): - levels = {str(index) for index in range(len(question.criteria))} - if ( - answer.probabilities.keys() != levels - or answer.legend.keys() != levels - ): - raise ValueError("评分等级不完整") - if not 0 <= answer.score <= len(question.criteria) - 1: - raise ValueError("评分超出等级范围") - if isinstance(answer, (ChoiceResult, ScoreResult)): - if not isclose(sum(answer.probabilities.values()), 1, abs_tol=1e-4): - raise ValueError("概率之和不为 1") + raise ValueError("评分超出等级范围") return result except (ValidationError, ValueError) as error: raise DecisionError( @@ -166,17 +140,9 @@ class DecisionEngine: ) from error async def aclose(self) -> None: - if not self._closed: - if self._owns_client and self._client is not None: - await self._client.aclose() - self._closed = True - - async def __aenter__(self) -> Self: - self._get_client() - return self - - async def __aexit__(self, exc_type, exc_value, traceback) -> None: - await self.aclose() + if self._client is not None: + await self._client.aclose() + self._client = None @cache @@ -186,11 +152,7 @@ def _cached_engine() -> DecisionEngine: def get_decision_engine() -> DecisionEngine: """与 get_llm 类似的共享入口;适用于单个服务事件循环""" - engine = _cached_engine() - if engine.closed: - _cached_engine.cache_clear() - engine = _cached_engine() - return engine + return _cached_engine() async def close_decision_engine() -> None: diff --git a/app/clients/llm.py b/app/clients/llm.py index 7b1a32a..df147b1 100644 --- a/app/clients/llm.py +++ b/app/clients/llm.py @@ -1,10 +1,8 @@ """统一获取普通模型或绑定结构化输出的模型""" from functools import cache -from typing import Literal, TypeVar, cast, overload +from typing import Any, TypeVar, overload -from langchain_core.language_models import LanguageModelInput -from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.runnables import Runnable from langchain_openai import ChatOpenAI from pydantic import BaseModel @@ -12,7 +10,6 @@ from pydantic import BaseModel from app.config import settings OutputModel = TypeVar("OutputModel", bound=BaseModel) -StructuredMethod = Literal["function_calling", "json_schema", "json_mode"] @cache @@ -29,64 +26,22 @@ def _get_base_llm() -> ChatOpenAI: ) +# 下面两个 @overload 只给类型检查器和编辑器看,运行时不会执行 +# 不传 schema:返回普通聊天模型,输出是自由文本 @overload -def get_llm( - schema: type[OutputModel], - *, - method: StructuredMethod = "function_calling", - strict: bool | None = False, - client: BaseChatModel | None = None, -) -> Runnable[LanguageModelInput, OutputModel]: ... +def get_llm() -> ChatOpenAI: ... +# 传入 Pydantic 类:返回结构化调用器,ainvoke 的结果就是该类的实例 +# OutputModel 让返回类型跟着传入的类走,例如 get_llm(Requirement) 得到 Runnable[..., Requirement] @overload -def get_llm( - schema: None = None, - *, - method: StructuredMethod = "function_calling", - strict: bool | None = False, - client: None = None, -) -> ChatOpenAI: ... +def get_llm(schema: type[OutputModel]) -> Runnable[Any, OutputModel]: ... -@overload -def get_llm( - schema: None = None, - *, - method: StructuredMethod = "function_calling", - strict: bool | None = False, - client: BaseChatModel, -) -> BaseChatModel: ... - - -def get_llm( - schema: type[OutputModel] | None = None, - *, - method: StructuredMethod = "function_calling", - strict: bool | None = False, - client: BaseChatModel | None = None, -) -> BaseChatModel | Runnable[LanguageModelInput, OutputModel]: - """不传 schema 返回普通模型,传入 Pydantic 类则返回结构化调用器 - - client 用于注入模型,省略时使用缓存客户端 - strict 控制服务端 schema 约束,本地仍由 Pydantic 校验 - """ - if schema is not None and ( - not isinstance(schema, type) or not issubclass(schema, BaseModel) - ): - raise TypeError("schema 必须是 Pydantic BaseModel 类") - if method not in ("function_calling", "json_schema", "json_mode"): - raise ValueError(f"不支持的结构化输出方式: {method}") - if schema is None and (method != "function_calling" or strict not in (False, None)): - raise ValueError("设置结构化输出参数时必须提供 schema") - if method == "json_mode" and strict is True: - raise ValueError("json_mode 不支持 strict=True") - - llm = client if client is not None else _get_base_llm() +# 真正的实现:schema 用 type[BaseModel] | None 覆盖上面两种调用 +# 返回类型由 overload 决定,这里不再重复标注 +def get_llm(schema: type[BaseModel] | None = None) -> Any: + llm = _get_base_llm() if schema is None: return llm - # json_mode 不接受 strict 参数,提示词还需要明确要求 JSON 及其结构 - structured = llm.with_structured_output( - schema, method=method, strict=None if method == "json_mode" else strict - ) - return cast(Runnable[LanguageModelInput, OutputModel], structured) + return llm.with_structured_output(schema, method="function_calling", strict=False) diff --git a/app/core/exceptions.py b/app/core/exceptions.py index e7e7f48..de41f55 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -10,7 +10,7 @@ from openai import APIError, APITimeoutError from starlette.exceptions import HTTPException from app.common.response import error_content -from app.exceptions import AppError, DecisionError, ImportProcessError +from app.exceptions import AppError, DecisionError class BizException(AppError): @@ -58,22 +58,14 @@ async def handle_unexpected_exception( exc._logged_by_api = True - # 节点包装异常时沿原始原因查找,区分上游服务故障与内部错误 - cause = exc - while isinstance(cause, ImportProcessError): - nested = cause.cause or cause.__cause__ - if nested is None: - break - cause = nested - if isinstance(cause, APITimeoutError) or ( - isinstance(cause, DecisionError) and cause.kind == "timeout" + if isinstance(exc, APITimeoutError) or ( + isinstance(exc, DecisionError) and exc.kind == "timeout" ): return JSONResponse( status_code=504, content=error_content(504, "模型服务请求超时") ) - if isinstance(cause, APIError) or ( - isinstance(cause, DecisionError) - and cause.kind not in {"configuration", "closed"} + if isinstance(exc, APIError) or ( + isinstance(exc, DecisionError) and exc.kind != "configuration" ): return JSONResponse( status_code=502, content=error_content(502, "模型服务调用失败") diff --git a/app/exceptions.py b/app/exceptions.py index 589de4d..871a7a4 100644 --- a/app/exceptions.py +++ b/app/exceptions.py @@ -17,23 +17,3 @@ class DecisionError(AppError): self.status = status self.request_id = request_id super().__init__(message) - - -class ImportProcessError(AppError): - """导入流程异常,保留节点名称和原始异常信息""" - - def __init__( - self, message: str, node_name: str = "", cause: Exception | None = None - ): - self.node_name = node_name - self.cause = cause - super().__init__(message) - - def __str__(self) -> str: - parts = [] - if self.node_name: - parts.append(f"[{self.node_name}]") - parts.append(super().__str__()) - if self.cause is not None: - parts.append(f"(原因: {self.cause})") - return " ".join(parts) diff --git a/app/processor/base.py b/app/processor/base.py index 9cde8d3..6fb9e54 100644 --- a/app/processor/base.py +++ b/app/processor/base.py @@ -1,15 +1,13 @@ import inspect import time from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, TypeVar +from contextlib import contextmanager +from typing import TypeVar from langgraph.errors import GraphInterrupt from loguru import logger -from pydantic import ValidationError -if TYPE_CHECKING: - from app.config import Settings -from app.exceptions import AppError, ImportProcessError +from app.config import Settings, settings from app.processor.progress import node_progress T = TypeVar("T") @@ -18,67 +16,44 @@ T = TypeVar("T") class BaseNode(ABC): name: str = "base_node" - def __init__(self, config: "Settings | None" = None): - """保存显式配置;全局配置在节点实际使用时才加载。""" - self._config = config + def __init__(self, config: Settings | None = None): + self.config = config if config is not None else settings self.logger = logger.bind(node=self.name) - @property - def config(self) -> "Settings": - if self._config is None: - from app.config import settings - - self._config = settings - return self._config - - def __call__(self, state: T) -> T: - """执行节点,统一记录日志、耗时并保留业务异常类型""" - if inspect.iscoroutinefunction(self.process): - return self._acall(state) + @contextmanager + def _instrument(self, state): + """统一记录日志、耗时和进度事件,异常保持原类型向上抛出""" start_time = time.perf_counter() self.logger.info("节点开始") try: with node_progress(self.name, state): - result = self.process(state) + yield except GraphInterrupt: raise - except (AppError, ValidationError): + except Exception: self.logger.error("节点执行失败") raise - except Exception as e: - self.logger.error("节点执行失败") - raise ImportProcessError(str(e), node_name=self.name) from e else: self.logger.info("节点完成") - return result finally: self.logger.info(f"节点耗时 {time.perf_counter() - start_time:.3f} 秒") + def __call__(self, state: T) -> T: + # 异步节点直接调用时返回协程,日志与进度在 _acall 内等待完成后记录 + if inspect.iscoroutinefunction(self.process): + return self._acall(state) + with self._instrument(state): + return self.process(state) + + async def _acall(self, state: T) -> T: + """异步节点入口,由图根据 process 是否为协程选择""" + with self._instrument(state): + return await self.process(state) + @abstractmethod def process(self, state: T) -> T: """子类实现节点核心逻辑,返回状态更新""" - async def _acall(self, state: T) -> T: - """异步节点等待执行完成后再记录结果和耗时""" - start_time = time.perf_counter() - self.logger.info("节点开始") - try: - with node_progress(self.name, state): - result = await self.process(state) - except GraphInterrupt: - raise - except (AppError, ValidationError): - self.logger.error("节点执行失败") - raise - except Exception as error: - self.logger.error("节点执行失败") - raise ImportProcessError(str(error), node_name=self.name) from error - else: - self.logger.info("节点完成") - return result - finally: - self.logger.info(f"节点耗时 {time.perf_counter() - start_time:.3f} 秒") - 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 117d04e..bf1772f 100644 --- a/app/processor/graph.py +++ b/app/processor/graph.py @@ -6,6 +6,7 @@ from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from langgraph.graph import END, StateGraph +from app.config import settings from app.processor.nodes.action_node import ActionNode, can_generate_actions from app.processor.nodes.clarify_node import ClarifyNode from app.processor.nodes.component_resource_node import ComponentResourceNode @@ -46,8 +47,6 @@ def route_after_scope(state: QueryState) -> str: def build_graph(checkpointer=None): - from app.config import settings - # 最大重试次数 max_retries = settings.max_action_retries @@ -331,8 +330,6 @@ if __name__ == "__main__": print( f"{index}. {option['label']}{marker}:{option['description']}" ) - if "error" in pending: - print(pending["error"]) answer = input("输入编号或自由回答(/cancel 取消):") reply = {"cancel": True} if answer == "/cancel" else {"answer": answer} if answer.isdigit() and 1 <= int(answer) <= len(options): diff --git a/app/processor/nodes/action_node.py b/app/processor/nodes/action_node.py index 6276a44..cedc374 100644 --- a/app/processor/nodes/action_node.py +++ b/app/processor/nodes/action_node.py @@ -4,7 +4,7 @@ import json from langchain_core.exceptions import OutputParserException from langchain_core.messages import HumanMessage, SystemMessage -from pydantic import BaseModel, ValidationError +from pydantic import ValidationError from app.actions.plan import assemble_plan_actions, plan_block_reason, plan_materials from app.clients.llm import get_llm @@ -84,12 +84,6 @@ class ActionNode(BaseNode): ) 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"), @@ -143,11 +137,13 @@ class ActionNode(BaseNode): ] ) - # 将模型对象转成字典,非模型返回值保持原样,统一交给后面的协议校验 - raw = result.model_dump() if isinstance(result, BaseModel) else result + # 响应中没有工具调用时解析器会返回 None + if result is None: + raise OutputParserException("模型未返回结构化结果") + raw = result.model_dump() # 规划生成的新增节点只保留显式提供的字段,避免模型默认值覆盖物料默认配置 - if isinstance(result, ActionBatch) and state.get("screen_plan") is not None: + if state.get("screen_plan") is not None: # 将原动作对象与序列化结果逐项对应,只调整新增动作的节点数据 for source, target in zip(result.actions, raw["actions"]): if isinstance(source, AddAction): @@ -277,6 +273,7 @@ if __name__ == "__main__": ) # 加 --modify 测试规划到已有组件调整,无需新增物料 if "--modify" in sys.argv: + state = ComponentResourceNode().process(state) state["change_scope"].needs_planning = True state["screen_plan"] = ScreenPlan( goal="向右移动 50px", diff --git a/app/processor/nodes/planner_node.py b/app/processor/nodes/planner_node.py index c89faaa..e8ca0ff 100644 --- a/app/processor/nodes/planner_node.py +++ b/app/processor/nodes/planner_node.py @@ -21,26 +21,10 @@ class PlannerNode(BaseNode): requirement = state["requirement"] scope = state["change_scope"] - # 如果需求仍需澄清、不是新建或修改大屏,或无需规划,则拒绝进入本节点 - if ( - requirement.needs_clarification - or requirement.intent - not in ( - "create_screen", - "modify_screen", - ) - or not scope.needs_planning - ): - raise ValueError("PlannerNode 只处理信息完整且需要规划的新建或修改需求") - # 获取当前大屏和用户输入,供模型结合已有内容安排布局与组件 context = state["input_context"] - # 获取可选的组件查询结果 - resources = state.get("component_resources") - - # 如果已有查询结果但目录未就绪,则停止规划,避免忽略资源查询失败 - if resources is not None and resources.status != "resolved": - raise ValueError("组件目录尚未就绪,不能继续规划") + # 图保证组件目录已就绪后才进入规划 + resources = state["component_resources"] llm_client = get_llm(ScreenPlan) result = await llm_client.ainvoke( @@ -54,10 +38,8 @@ class PlannerNode(BaseNode): "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, + # 仅提供规划需要的组件摘要,省略完整默认配置 + "component_resources": resources.to_planner_context(), }, ensure_ascii=False, ) @@ -65,26 +47,23 @@ class PlannerNode(BaseNode): ] ) - # 将模型结果转成字典后重新校验,确保规划字段及组件关系符合输出协议 - plan = ScreenPlan.model_validate(result.model_dump()) + plan = result - # 如果提供了组件目录,则额外校验规划中的物料类型是否真实受支持 - 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 - ): - if component.component_type not in supported_types: - raise ValueError( - f"规划使用了目录中不存在的组件类型: {component.component_type}" - ) + # 校验规划中的物料类型是否真实受支持 + 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 + and component.component_type not in supported_types + ): + raise ValueError( + f"规划使用了目录中不存在的组件类型: {component.component_type}" + ) # 收集当前大屏中真实存在的节点 ID,用于检查规划引用的已有目标 existing_ids = {node.id for node in context.screen.nodes} diff --git a/app/processor/nodes/reflection_node.py b/app/processor/nodes/reflection_node.py index 4d8baf1..d500dea 100644 --- a/app/processor/nodes/reflection_node.py +++ b/app/processor/nodes/reflection_node.py @@ -4,7 +4,7 @@ import json from langchain_core.exceptions import OutputParserException from langchain_core.messages import HumanMessage, SystemMessage -from pydantic import BaseModel, ValidationError +from pydantic import ValidationError from app.actions.plan import plan_materials from app.clients.llm import get_llm @@ -48,7 +48,7 @@ class ReflectionNode(BaseNode): # 只捕获输出结构和解析错误,配置、网络等异常继续向上抛出 try: # 异步生成修正建议,本节点不直接修改动作 - result = await llm_client.ainvoke( + plan = await llm_client.ainvoke( [ # 系统消息提供稳定的失败分析和修正规则 SystemMessage(content=REFLECTION_SYSTEM_PROMPT), @@ -56,10 +56,9 @@ class ReflectionNode(BaseNode): HumanMessage(content=json.dumps(payload, ensure_ascii=False)), ] ) - # 将模型对象转成字典后统一校验,非模型返回值直接交给协议校验 - plan = ActionRepairPlan.model_validate( - result.model_dump() if isinstance(result, BaseModel) else result - ) + # 响应中没有工具调用时解析器会返回 None + if plan is None: + raise OutputParserException("模型未返回结构化结果") except (ValidationError, OutputParserException): # 如果未能得到合法建议,清空旧修正计划,避免再次使用已失效的建议 updated_state["repair_plan"] = None diff --git a/app/processor/nodes/scope_node.py b/app/processor/nodes/scope_node.py index a1a34bc..a99a59c 100644 --- a/app/processor/nodes/scope_node.py +++ b/app/processor/nodes/scope_node.py @@ -21,13 +21,6 @@ class ScopeNode(BaseNode): # 获取需求节点提取的结构化需求,作为范围判断的依据 requirement = state["requirement"] - # 如果需求仍需澄清,或意图不是新建、修改大屏,则不能进入范围判断 - if requirement.needs_clarification or requirement.intent not in ( - "create_screen", - "modify_screen", - ): - raise ValueError("ScopeNode 只处理信息完整的新建或修改需求") - # 获取当前大屏和用户输入,帮助模型结合已有组件判断影响范围 context = state["input_context"] diff --git a/app/processor/nodes/validator_node.py b/app/processor/nodes/validator_node.py index 1227347..572282a 100644 --- a/app/processor/nodes/validator_node.py +++ b/app/processor/nodes/validator_node.py @@ -1,6 +1,6 @@ """校验动作范围,再按顺序修改大屏 JSON 副本,检查整批动作能否执行""" -from pydantic import BaseModel, ValidationError +from pydantic import ValidationError from app.actions.plan import validate_plan_actions, validate_plan_result from app.actions.registry import apply_action @@ -22,17 +22,9 @@ class ValidatorNode(BaseNode): actions = state["actions"] try: - # 重新校验模型内容,避免赋值后绕过字段校验 + # 转回字典重新校验,避免赋值后绕过字段校验 batch = ActionBatch.model_validate( - { - "actions": [ - # 将动作模型转回字典,确保重新触发字段校验,字典输入则保持原样 - action.model_dump() if isinstance(action, BaseModel) else action - for action in actions - ] - if isinstance(actions, list) - else actions - } + {"actions": [action.model_dump() for action in actions]} ) except ValidationError as error: # 将结构异常转换成统一错误列表,供后续修正流程定位问题 @@ -43,8 +35,7 @@ class ValidatorNode(BaseNode): updated_state["simulated_screen"] = None return updated_state - # 获取可选的结构化需求,没有需求和规划时仅做结构校验与独立模拟执行 - requirement = state.get("requirement") + requirement = state["requirement"] # 如果已有规划,检查动作是否覆盖规划目标,以及是否修改了规划外的内容 if state.get("screen_plan") is not None: @@ -55,7 +46,7 @@ class ValidatorNode(BaseNode): updated_state["validation_errors"] = errors updated_state["simulated_screen"] = None return updated_state - elif requirement is not None: + else: # 此分支没有规划,只校验简单修改,新增动作需要走上面的规划分支 if any(isinstance(action, AddAction) for action in batch.actions): # 为每个新增动作记录下标和范围错误,便于修正时逐项定位 @@ -64,7 +55,7 @@ class ValidatorNode(BaseNode): ActionValidationError( action_index=index, code="scope_error", - message="新增动作尚未接入需求范围校验,当前仅支持独立模拟执行", + message="新增动作需要先经过规划,不能作为简单修改直接校验", ) for index, action in enumerate(batch.actions) if isinstance(action, AddAction) @@ -175,6 +166,7 @@ if __name__ == "__main__": from app.processor.nodes.input_node import InputNode from app.schemas.actions import MoveAction, ResizeAction, UpdateAction + from app.schemas.requirement import Requirement # 1. 创建 state,不需要调用模型 state = InputNode().process( @@ -195,6 +187,15 @@ if __name__ == "__main__": } } ) + state["requirement"] = Requirement( + intent="modify_screen", + goal="移动、调整尺寸并修改标题", + needs_clarification=False, + changes=[ + {"target_scope": "nodes", "target_ids": ["sales-chart"], "operation": op} + for op in ("move", "resize", "update") + ], + ) state["actions"] = [ MoveAction(target_id="sales-chart", delta_x=50), ResizeAction(target_id="sales-chart", width=500), @@ -217,6 +218,15 @@ if __name__ == "__main__": # 3. 在第二个动作制造错误,验证整批失败且不返回部分结果 failed_state = { **final_state, + "requirement": Requirement( + intent="modify_screen", + goal="移动两个组件", + needs_clarification=False, + changes=[ + {"target_scope": "nodes", "target_ids": [target], "operation": "move"} + for target in ("sales-chart", "missing") + ], + ), "actions": [ state["actions"][0], MoveAction(target_id="missing", delta_x=50), diff --git a/app/processor/nodes/wait_for_user_node.py b/app/processor/nodes/wait_for_user_node.py index f711109..b4afe72 100644 --- a/app/processor/nodes/wait_for_user_node.py +++ b/app/processor/nodes/wait_for_user_node.py @@ -1,7 +1,6 @@ """暂停图并接收用户回答,不在恢复时重复调用模型""" from langgraph.types import interrupt -from pydantic import ValidationError from app.processor.base import BaseNode from app.processor.state import QueryState @@ -13,25 +12,10 @@ class WaitForUserNode(BaseNode): def process(self, state: QueryState) -> QueryState: updated_state: QueryState = state.copy() - payload = state["clarification"].model_dump(mode="json") - while True: - # interrupt 的控制异常必须交给 LangGraph 处理 - value = interrupt(payload) - try: - reply = ClarificationReply.model_validate( - {"answer": value} if isinstance(value, str) else value - ) - if not reply.cancel and reply.option_id is not None: - if reply.option_id not in { - option.id for option in state["clarification"].options - }: - raise ValueError("无效选项") - break - except (ValidationError, ValueError): - payload = { - **payload, - "error": "请选择本题有效选项或填写回答,并提供有效的大屏数据;也可设置 cancel=true 取消", - } + # 回复的格式与选项有效性已在恢复前由服务层校验,此处只还原为模型 + reply = ClarificationReply.model_validate( + interrupt(state["clarification"].model_dump(mode="json")) + ) updated_state["clarification_reply"] = reply updated_state["status"] = "cancelled" if reply.cancel else "running"