135 lines
5.1 KiB
Python
135 lines
5.1 KiB
Python
"""需求落地策略的单一来源：模型只负责理解需求，本轮是执行、澄清还是直接回复由这里决定
|
||
|
||
决策依据是当前大屏里真实存在的组件，而不是模型对"要不要澄清"的判断：
|
||
- 新增组件或目标可唯一确定：执行，缺少的信息采用默认值
|
||
- 目标有多个候选：澄清，选项直接由候选组件生成，不由模型编造
|
||
- 目标没有任何候选：直接回复，说明原因并可建议替代做法，不执行也不追问
|
||
- 没有可执行的变更：交给澄清节点询问
|
||
"""
|
||
|
||
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
|