diff --git a/app/processor/base.py b/app/processor/base.py index 33ca581..32dc489 100644 --- a/app/processor/base.py +++ b/app/processor/base.py @@ -1,4 +1,5 @@ import logging +import inspect import time from abc import ABC, abstractmethod from typing import TYPE_CHECKING, TypeVar @@ -30,6 +31,8 @@ class BaseNode(ABC): def __call__(self, state: T) -> T: """执行节点,统一记录日志、耗时并保留业务异常类型""" + if inspect.iscoroutinefunction(self.process): + return self._acall(state) start_time = time.perf_counter() self.logger.info("--- %s 开始 ---", self.name) try: @@ -52,6 +55,24 @@ class BaseNode(ABC): def process(self, state: T) -> T: """子类实现节点核心逻辑,返回状态更新""" + async def _acall(self, state: T) -> T: + """异步节点等待执行完成后再记录结果和耗时""" + start_time = time.perf_counter() + self.logger.info("--- %s 开始 ---", self.name) + try: + result = await self.process(state) + except (AppError, ValidationError): + self.logger.exception("%s 执行失败", self.name) + raise + except Exception as error: + self.logger.exception("%s 执行失败", self.name) + raise ImportProcessError(str(error), node_name=self.name) from error + else: + self.logger.info("--- %s 完成 ---", self.name) + return result + finally: + self.logger.info("%s 耗时 %.3f 秒", self.name, time.perf_counter() - start_time) + def log_step(self, step_name: str, message: str = ""): """记录步骤日志。""" log_msg = f"[{step_name}]" diff --git a/app/processor/nodes/requirement_node.py b/app/processor/nodes/requirement_node.py index e69de29..89b475f 100644 --- a/app/processor/nodes/requirement_node.py +++ b/app/processor/nodes/requirement_node.py @@ -0,0 +1,132 @@ +"""需求理解节点""" + +from app.clients.decision import DecisionEngine, get_decision_engine +from app.processor.base import BaseNode +from app.processor.state import QueryState +from app.prompt.requirement import build_intent_question +from app.schemas.base import JsonObject +from app.schemas.decision import DecisionRequest +from app.schemas.input_context import InputContext + + +class RequirementNode(BaseNode): + name = "requirement_node" + + def __init__(self, *, decision_engine: DecisionEngine | None = None): + super().__init__() + self._decision_engine = decision_engine + + @staticmethod + def build_decision_state(context: InputContext) -> JsonObject: + """只整理判断意图需要的摘要,完整 Screen 仍保留在输入上下文""" + nodes: list[JsonObject] = [] + for node in context.screen.nodes: + option = node.props.get("option") + title = option.get("title") if isinstance(option, dict) else None + titles = title if isinstance(title, list) else [title] + nodes.append( + { + "id": node.id, + "type": node.type, + "name": node.name, + "titles": [ + item["text"] + for item in titles + if isinstance(item, dict) and isinstance(item.get("text"), str) + ], + "layout": node.layout.model_dump(mode="json", exclude_unset=True), + } + ) + return { + "prompt": context.prompt, + "context": context.context.model_dump(mode="json", exclude_unset=True), + "screen_summary": { + "canvas": { + "width": context.screen.canvas.width, + "height": context.screen.canvas.height, + }, + "nodes": nodes, + "data_sources": [ + { + key: source[key] + for key in ("id", "name", "type") + if key in source + } + for source in context.screen.dataSource + ], + }, + "attachments": [ + { + key: attachment[key] + for key in ("id", "kind", "name") + if key in attachment + } + for attachment in context.attachments + ], + } + + async def process(self, state: QueryState) -> QueryState: + # 构建jev请求 + request = DecisionRequest( + # 获取 InputNode 整理好的完整上下文 + state=self.build_decision_state(state["input_context"]), + decisions={"intent": build_intent_question()}, + ) + + # 调用 jev + engine = self._decision_engine or get_decision_engine() + decision = await engine.run(request) + + updated_state: QueryState = {**state, "intent_decision": decision} + return updated_state + + +if __name__ == "__main__": + import asyncio + import json + + from pydantic import BaseModel + + from app.processor.nodes.input_node import InputNode + + async def main(): + # 1. 修改这里的请求,测试不同意图 + state: QueryState = { + "request": { + "prompt": "这个图太乱了,帮我重新排一下", + "screen": { + "canvas": {"width": 1920, "height": 1080}, + "nodes": [ + { + "id": "sales-chart", + "type": "bar-chart", + "name": "销售额统计", + "layout": {"x": 26, "y": 34.8, "width": 400, "height": 260}, + "props": {"option": {"title": {"text": "销售额统计"}}}, + } + ], + "dataSource": [], + }, + "context": {"selected_node_ids": ["sales-chart"]}, + } + } + # 2. 先校验输入,再真实调用 Jev,退出时关闭连接 + state = InputNode().process(state) + async with DecisionEngine() as engine: + node = RequirementNode(decision_engine=engine) + final_state = await node.process(state) + # 3. 打印完整 state,包含选项、概率、置信度和调用信息 + print( + json.dumps( + { + key: value.model_dump(mode="json", exclude_unset=True) + if isinstance(value, BaseModel) + else value + for key, value in final_state.items() + }, + ensure_ascii=False, + indent=2, + ) + ) + + asyncio.run(main()) diff --git a/app/processor/state.py b/app/processor/state.py index 1e755fb..3b680dd 100644 --- a/app/processor/state.py +++ b/app/processor/state.py @@ -2,8 +2,10 @@ from typing import NotRequired, TypedDict from app.schemas.base import JsonObject from app.schemas.input_context import InputContext +from app.schemas.decision import DecisionResponse class QueryState(TypedDict): request: JsonObject input_context: NotRequired[InputContext] + intent_decision: NotRequired[DecisionResponse] diff --git a/app/prompt/requirement.py b/app/prompt/requirement.py new file mode 100644 index 0000000..5b64de2 --- /dev/null +++ b/app/prompt/requirement.py @@ -0,0 +1,24 @@ +"""需求节点的意图判断定义""" + +from app.schemas.decision import ChoiceDecision + + +def build_intent_question() -> ChoiceDecision: + return ChoiceDecision( + # 判断主要意图;只有明确要求实际编辑时才判断为修改,不能仅根据选中节点判断 + instructions=( + "Classify the user's primary intent. " + "Choose modify_screen only when the user explicitly requests an actual edit. " + "Do not infer editing solely from selected nodes." + ), + criteria={ + # 创建新的大屏 + "create_screen": "Create a new screen", + # 明确要求修改、增加、删除或重新设计已有大屏 + "modify_screen": "Explicitly request modifying, adding, deleting, or redesigning an existing screen", + # 询问知识、原因、建议或如何修改,但没有要求实际执行 + "qa": "Ask for information, reasons, suggestions, or how to improve something without requesting an actual edit", + # 无法判断意图或不属于上述类型 + "unclear": "Intent is unclear or outside the above categories", + }, + ) diff --git a/app/schemas/decision.py b/app/schemas/decision.py index fa76de7..2de22db 100644 --- a/app/schemas/decision.py +++ b/app/schemas/decision.py @@ -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 # 服务端请求标识,方便排查问题 diff --git a/app/schemas/requirement.py b/app/schemas/requirement.py new file mode 100644 index 0000000..e01407d --- /dev/null +++ b/app/schemas/requirement.py @@ -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 diff --git a/app/schemas/screen.py b/app/schemas/screen.py new file mode 100644 index 0000000..c66349a --- /dev/null +++ b/app/schemas/screen.py @@ -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