"""需求落地策略的单一来源:模型只负责理解需求,本轮是执行、澄清还是直接回复由这里决定 决策依据是当前大屏里真实存在的组件,而不是模型对"要不要澄清"的判断: - 新增组件或目标可唯一确定:执行,缺少的信息采用默认值 - 目标有多个候选:澄清,选项直接由候选组件生成,不由模型编造 - 目标没有任何候选:直接回复,说明原因并可建议替代做法,不执行也不追问 - 没有可执行的变更:交给澄清节点询问 """ from dataclasses import dataclass from typing import Literal from app.schemas.clarification import Clarification, ClarificationOption from app.schemas.requirement import Requirement, RequirementChange from app.schemas.screen import Screen, ScreenNode Outcome = Literal["execute", "clarify", "inform"] # 默认值补齐的说明,模型提出了非阻塞问题但仍执行时写入 assumptions DEFAULT_ASSUMPTION = "未明确的位置、尺寸、样式和数据采用默认值,数据使用物料示例数据" # 候选超过选项上限时不提供选项,只让用户描述 MAX_OPTIONS = 3 @dataclass(frozen=True) class OutcomeDecision: outcome: Outcome requirement: Requirement # 已由候选组件生成的澄清问题;为 None 且 outcome 为 clarify 时由澄清节点生成 clarification: Clarification | None = None # 直接回复时交给回答节点的事实说明 notice: str | None = None def decide_outcome(requirement: Requirement, screen: Screen) -> OutcomeDecision: nodes = {node.id: node for node in screen.nodes} unresolved = [c for c in requirement.changes if c.target_scope == "unresolved"] for change in unresolved: if not _candidates(change, nodes): return OutcomeDecision( "inform", requirement, notice=_missing_target_notice(change, screen) ) ambiguous = [c for c in unresolved if len(_candidates(c, nodes)) > 1] if ambiguous: change = ambiguous[0] return OutcomeDecision( "clarify", requirement, clarification=_candidate_question( change, [nodes[i] for i in _candidates(change, nodes)], requirement ), ) if not requirement.changes: # 没有可执行的变更时才需要模型组织问题 outcome: Outcome = "clarify" if requirement.needs_clarification else "execute" return OutcomeDecision(outcome, requirement) # 所有目标均已确定:唯一候选改写为明确目标,模型提出的非阻塞问题改用默认值 changes = [ _resolve(change, _candidates(change, nodes)[0]) if change.target_scope == "unresolved" else change for change in requirement.changes ] assumptions = list(requirement.assumptions) if requirement.needs_clarification and DEFAULT_ASSUMPTION not in assumptions: assumptions.append(DEFAULT_ASSUMPTION) resolved = Requirement.model_validate( { **requirement.model_dump(), "changes": [change.model_dump() for change in changes], "needs_clarification": False, "clarification_question": None, "assumptions": assumptions, } ) return OutcomeDecision("execute", resolved) def _candidates(change: RequirementChange, nodes: dict[str, ScreenNode]) -> list[str]: # 只保留真实存在的组件,模型虚构的 ID 视为没有候选 return [ node_id for node_id in dict.fromkeys(change.candidate_ids) if node_id in nodes ] def _resolve(change: RequirementChange, node_id: str) -> RequirementChange: return RequirementChange.model_validate( { **change.model_dump(), "target_scope": "nodes", "target_ids": [node_id], "candidate_ids": [], } ) def _missing_target_notice(change: RequirementChange, screen: Screen) -> str: target = change.target_description or "目标组件" if not screen.nodes: state = "当前画布上还没有任何组件" else: names = "、".join(_label(node) for node in screen.nodes[:5]) state = f"当前画布上的组件有:{names}" return f"本次请求无法执行:没有找到“{target}”对应的组件,{state}" def _candidate_question( change: RequirementChange, candidates: list[ScreenNode], requirement: Requirement ) -> Clarification: target = change.target_description or "组件" question = requirement.clarification_question or f"你指的是哪个{target}?" if len(candidates) > MAX_OPTIONS: return Clarification(question=question) # 选项 ID 使用组件 ID,用户点选后需求理解可直接定位目标 return Clarification( question=question, options=[ ClarificationOption( id=node.id, label=_label(node), description=f"{node.type},位于 ({node.layout.x:g}, {node.layout.y:g})", ) for node in candidates ], ) def _label(node: ScreenNode) -> str: return node.name or node.type