feat: jev 接入，判断创建、修改、问答、不明确四类意图
This commit is contained in:
+46
-17
@@ -6,42 +6,60 @@ from pydantic import BaseModel, ConfigDict, Field, JsonValue, model_validator
|
||||
|
||||
from .base import NonBlankString
|
||||
|
||||
# 文本或结构化 JSON，用于传递上下文和判断说明
|
||||
Content = str | dict[str, JsonValue] | list[JsonValue]
|
||||
# 概率和置信度都限制在 0～1
|
||||
Probability = Annotated[float, Field(ge=0, le=1)]
|
||||
|
||||
|
||||
class DecisionModel(BaseModel):
|
||||
"""公共校验规则：拒绝未知字段、隐式类型转换和非有限数值"""
|
||||
|
||||
model_config = ConfigDict(extra="forbid", strict=True, allow_inf_nan=False)
|
||||
|
||||
|
||||
class ChoiceDecision(DecisionModel):
|
||||
"""选择题：从候选项中选择一个"""
|
||||
|
||||
type: Literal["choice"] = "choice"
|
||||
instructions: Content
|
||||
instructions: Content # 告诉模型要判断什么
|
||||
criteria: dict[NonBlankString, Content | None] = Field(min_length=1, max_length=255)
|
||||
# criteria 的键是选项，值是说明，不需要说明时填 None
|
||||
|
||||
|
||||
class ScoreDecision(DecisionModel):
|
||||
"""评分题：按从低到高排列的等级进行评分"""
|
||||
|
||||
type: Literal["score"] = "score"
|
||||
instructions: Content
|
||||
criteria: list[Content] = Field(min_length=2, max_length=10)
|
||||
criteria: list[Content] = Field(min_length=2, max_length=10) # 等级从 0 开始编号
|
||||
|
||||
|
||||
class NoulDecision(DecisionModel):
|
||||
"""判断题：返回命题成立的概率，不直接返回布尔值"""
|
||||
|
||||
type: Literal["noul"] = "noul"
|
||||
instructions: Content
|
||||
criteria: dict[Literal["true", "false"], Content] | None = None
|
||||
criteria: dict[Literal["true", "false"], Content] | None = (
|
||||
None # 可选的成立与不成立说明
|
||||
)
|
||||
|
||||
|
||||
# 根据 type 选择对应模型，避免混用选择题和评分题的字段
|
||||
DecisionSchema = Annotated[
|
||||
ChoiceDecision | ScoreDecision | NoulDecision, Field(discriminator="type")
|
||||
]
|
||||
|
||||
|
||||
class DecisionRequest(DecisionModel):
|
||||
state: Content
|
||||
decisions: dict[NonBlankString, DecisionSchema] = Field(min_length=1)
|
||||
model: NonBlankString | None = None
|
||||
# 官方扩展字段入口，不允许覆盖已经校验的协议字段。
|
||||
"""一次调用的输入，可同时提交多个独立判断"""
|
||||
|
||||
state: Content # 所有判断共享的业务上下文
|
||||
decisions: dict[NonBlankString, DecisionSchema] = Field(
|
||||
min_length=1
|
||||
) # 键用于对应返回结果
|
||||
model: NonBlankString | None = None # 不传时使用客户端默认模型
|
||||
# 官方扩展字段入口，不允许覆盖已经校验的协议字段
|
||||
extra_body: dict[str, JsonValue] = Field(default_factory=dict)
|
||||
|
||||
@model_validator(mode="after")
|
||||
@@ -52,37 +70,48 @@ class DecisionRequest(DecisionModel):
|
||||
|
||||
|
||||
class ChoiceResult(DecisionModel):
|
||||
"""选择结果及所有候选项的概率分布"""
|
||||
|
||||
type: Literal["choice"] = "choice"
|
||||
choice: str
|
||||
confidence: Probability
|
||||
probabilities: dict[str, Probability]
|
||||
choice: str # 选中的候选项
|
||||
confidence: Probability # 分布的集中程度，不等于选中项的概率
|
||||
probabilities: dict[str, Probability] # 选项 → 概率
|
||||
|
||||
|
||||
class ScoreResult(DecisionModel):
|
||||
"""评分结果，保留等级说明和概率分布"""
|
||||
|
||||
type: Literal["score"] = "score"
|
||||
score: float = Field(ge=0)
|
||||
score: float = Field(ge=0) # 按等级概率加权的分数，可能是小数
|
||||
confidence: Probability
|
||||
legend: dict[str, Content]
|
||||
probabilities: dict[str, Probability]
|
||||
legend: dict[str, Content] # 等级编号字符串 → 等级说明
|
||||
probabilities: dict[str, Probability] # 等级编号字符串 → 概率
|
||||
|
||||
|
||||
class NoulResult(DecisionModel):
|
||||
"""成立概率：接近 1 表示是，接近 0 表示否，接近 0.5 表示不确定"""
|
||||
|
||||
type: Literal["noul"] = "noul"
|
||||
noul: Probability
|
||||
noul: Probability # 没有单独的 confidence，业务自行决定判断阈值
|
||||
|
||||
|
||||
# 返回结果也按 type 区分，读取前先确认结果类型
|
||||
DecisionResult = Annotated[
|
||||
ChoiceResult | ScoreResult | NoulResult, Field(discriminator="type")
|
||||
]
|
||||
|
||||
|
||||
class DecisionUsage(DecisionModel):
|
||||
"""本次调用的 token 用量，None 表示服务端未提供"""
|
||||
|
||||
input_tokens: int | None = Field(default=None, ge=0)
|
||||
output_tokens: int | None = Field(default=None, ge=0)
|
||||
|
||||
|
||||
class DecisionResponse(DecisionModel):
|
||||
results: dict[str, DecisionResult]
|
||||
model: str
|
||||
"""统一返回结构，包含判断结果和调用信息"""
|
||||
|
||||
results: dict[str, DecisionResult] # 键与请求中的 decisions 一一对应
|
||||
model: str # 实际执行判断的模型
|
||||
usage: DecisionUsage
|
||||
request_id: str | None = None
|
||||
request_id: str | None = None # 服务端请求标识，方便排查问题
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
"""需求理解的输出协议，描述用户诉求，不包含前端执行动作。"""
|
||||
|
||||
from typing import Literal, Self
|
||||
|
||||
from pydantic import Field, model_validator
|
||||
|
||||
from .base import ExtensibleModel, JsonObject, NonBlankString
|
||||
|
||||
Intent = Literal["create_screen", "modify_screen", "qa", "unclear"]
|
||||
TargetScope = Literal["nodes", "canvas", "screen", "unresolved"]
|
||||
|
||||
|
||||
class RequirementChange(ExtensibleModel):
|
||||
"""一项需求变更。目标范围显式声明，避免空 ID 被理解成修改整个大屏。"""
|
||||
|
||||
target_scope: TargetScope = Field(
|
||||
description="nodes=已有节点；canvas=画布属性；screen=整体或新增内容；unresolved=目标待确认"
|
||||
)
|
||||
target_ids: list[NonBlankString] = Field(
|
||||
default_factory=list, description="已有节点的 ID；新增组件不能虚构 ID"
|
||||
)
|
||||
target_description: NonBlankString | None = Field(
|
||||
default=None, description="目标的自然语言描述，例如右侧区域、用户所说的那个图"
|
||||
)
|
||||
operation: NonBlankString = Field(
|
||||
description="需求操作，例如 move、resize、add、delete、restyle；允许后续扩展"
|
||||
)
|
||||
parameters: JsonObject = Field(
|
||||
default_factory=dict,
|
||||
description="需求参数，例如 delta_x=50；保留用户要求的单位与相对变化语义",
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_target(self) -> Self:
|
||||
if len(self.target_ids) != len(set(self.target_ids)):
|
||||
raise ValueError("target_ids 不能包含重复 ID")
|
||||
if self.target_scope == "nodes":
|
||||
if not self.target_ids:
|
||||
raise ValueError("nodes 范围必须提供 target_ids")
|
||||
elif self.target_ids:
|
||||
raise ValueError("只有 nodes 范围可以提供 target_ids")
|
||||
if self.target_scope == "unresolved" and self.target_description is None:
|
||||
raise ValueError("目标待确认时必须提供 target_description")
|
||||
return self
|
||||
|
||||
|
||||
class Requirement(ExtensibleModel):
|
||||
intent: Intent
|
||||
goal: NonBlankString = Field(description="忠实概括用户目标，不能擅自扩大修改范围")
|
||||
changes: list[RequirementChange] = Field(default_factory=list)
|
||||
needs_clarification: bool
|
||||
clarification_question: NonBlankString | None = None
|
||||
needs_planning: bool = Field(description="是否需要布局或设计规划，与修改范围分别判断")
|
||||
needs_resources: bool = Field(description="是否需要补充查询组件、数据、模板或素材")
|
||||
assumptions: list[NonBlankString] = Field(
|
||||
default_factory=list, description="非阻塞信息采用的默认假设，不可用来猜测删除目标"
|
||||
)
|
||||
constraints: list[NonBlankString] = Field(
|
||||
default_factory=list, description="用户明确提出的约束，例如只修改右侧、保留其他组件"
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_consistency(self) -> Self:
|
||||
if self.needs_clarification != (self.clarification_question is not None):
|
||||
raise ValueError("需要澄清时必须提供问题，不需要澄清时问题应为 null")
|
||||
if self.intent == "unclear" and not self.needs_clarification:
|
||||
raise ValueError("意图不明确时必须澄清")
|
||||
if any(change.target_scope == "unresolved" for change in self.changes):
|
||||
if not self.needs_clarification:
|
||||
raise ValueError("存在待确认目标时必须澄清")
|
||||
if self.intent == "qa" and self.changes:
|
||||
raise ValueError("纯问答需求不应包含大屏变更")
|
||||
return self
|
||||
@@ -0,0 +1,48 @@
|
||||
"""大屏公共结构；组件属性、事件与数据源配置保持开放"""
|
||||
|
||||
from typing import Annotated, Self
|
||||
|
||||
from pydantic import Field, model_validator
|
||||
|
||||
from .base import ExtensibleModel, JsonObject, NonBlankString
|
||||
|
||||
Number = int | float
|
||||
PositiveNumber = Annotated[Number, Field(gt=0)]
|
||||
|
||||
|
||||
class Canvas(ExtensibleModel):
|
||||
width: PositiveNumber
|
||||
height: PositiveNumber
|
||||
|
||||
|
||||
class Layout(ExtensibleModel):
|
||||
x: Number
|
||||
y: Number
|
||||
width: PositiveNumber
|
||||
height: PositiveNumber
|
||||
|
||||
|
||||
class ScreenNode(ExtensibleModel):
|
||||
id: NonBlankString
|
||||
type: NonBlankString
|
||||
layout: Layout
|
||||
name: str | None = None
|
||||
props: JsonObject = Field(default_factory=dict)
|
||||
style: JsonObject | None = None
|
||||
events: list[JsonObject] | None = None
|
||||
|
||||
|
||||
class Screen(ExtensibleModel):
|
||||
canvas: Canvas
|
||||
nodes: list[ScreenNode]
|
||||
# 保持前端字段名；数据源的具体类型与绑定校验后续再引入
|
||||
dataSource: list[JsonObject] = Field(default_factory=list)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def unique_node_ids(self) -> Self:
|
||||
seen: set[str] = set()
|
||||
for node in self.nodes:
|
||||
if node.id in seen:
|
||||
raise ValueError(f"重复的节点 ID: {node.id}")
|
||||
seen.add(node.id)
|
||||
return self
|
||||
Reference in New Issue
Block a user