143 lines
6.2 KiB
Python
143 lines
6.2 KiB
Python
"""输入解析与校验，可直接运行本文件修改示例进行调试"""
|
|
|
|
from app.processor.base import BaseNode
|
|
from app.processor.state import QueryState
|
|
from app.schemas.input_context import InputContext
|
|
|
|
|
|
class InputNode(BaseNode):
|
|
"""纯输入节点，不依赖模型配置；校验失败保留 ValidationError"""
|
|
|
|
name = "input_node"
|
|
|
|
def process(self, state: QueryState) -> QueryState:
|
|
# 浅拷贝状态
|
|
updated_state: QueryState = state.copy()
|
|
# 获取待合并的用户澄清回复
|
|
reply = state.get("clarification_reply")
|
|
|
|
# 如果有澄清回复，将补充信息合并到已有上下文后重新处理
|
|
if reply is not None:
|
|
# 将已有上下文转成独立的字典，避免直接修改原上下文中的嵌套数据
|
|
data = state["input_context"].model_dump(mode="json")
|
|
# 记录本轮澄清问题和用户回答，供后续节点理解补充需求
|
|
turn = {
|
|
"question": state["clarification"].question,
|
|
"answer": reply.answer,
|
|
}
|
|
|
|
# 如果用户选择了候选项，按选项 ID 查找并保存完整选项内容
|
|
if reply.option_id is not None:
|
|
option = next(
|
|
option
|
|
for option in state["clarification"].options
|
|
if option.id == reply.option_id
|
|
)
|
|
turn["selected_option"] = option.model_dump(mode="json")
|
|
|
|
# 将本轮问答追加到历史中，保留多轮澄清信息
|
|
data["clarification_history"].append(turn)
|
|
|
|
# 如果回复携带最新大屏快照，用它替换旧快照
|
|
if reply.screen is not None:
|
|
data["screen"] = reply.screen.model_dump(mode="json")
|
|
# 如果没有同时提供交互上下文，清空可能已失效的选中节点和版本号
|
|
if reply.context is None:
|
|
data["context"]["selected_node_ids"] = []
|
|
data["context"]["screen_revision"] = None
|
|
|
|
# 如果回复携带交互上下文，使用本次提交的选中节点和版本等信息
|
|
if reply.context is not None:
|
|
data["context"] = reply.context.model_dump(mode="json")
|
|
# 重新校验合并后的数据并构建上下文，校验失败时向上抛出 ValidationError
|
|
context = InputContext.model_validate(data)
|
|
|
|
# 保存新上下文并清空已消费的回复，避免后续重复合并
|
|
updated_state["input_context"] = context
|
|
updated_state["clarification_reply"] = None
|
|
# 清空旧意图、需求和规划，让后续节点根据补充信息重新推导
|
|
# LangGraph 的状态合并不会删除省略字段，因此需要显式赋空值
|
|
updated_state["intent_decision"] = None
|
|
updated_state["intent"] = None
|
|
updated_state["requirement"] = None
|
|
updated_state["requirement_outcome"] = None
|
|
updated_state["notice"] = None
|
|
updated_state["offer"] = None
|
|
updated_state["accepted_offer"] = None
|
|
updated_state["change_scope"] = None
|
|
updated_state["screen_plan"] = None
|
|
updated_state["component_resources"] = None
|
|
# 清空旧动作及其校验、模拟结果，避免复用补充需求前的处理结果
|
|
updated_state["actions"] = None
|
|
updated_state["validation_passed"] = None
|
|
updated_state["validation_errors"] = []
|
|
updated_state["simulated_screen"] = None
|
|
# 重置自动修正次数和失败信息，让本轮处理重新计算重试次数
|
|
updated_state["retry_count"] = 0
|
|
updated_state["repair_plan"] = None
|
|
updated_state["failure_reason"] = None
|
|
# 清空上一轮澄清问题和回答，恢复任务处理状态
|
|
updated_state["clarification"] = None
|
|
updated_state["answer"] = None
|
|
updated_state["status"] = "running"
|
|
|
|
# 返回澄清恢复后的状态，不再执行下方首次请求的初始化逻辑
|
|
return updated_state
|
|
|
|
# 没有澄清回复时，按首次请求校验原始输入，并深拷贝隔离提交时的快照
|
|
context = InputContext.model_validate(state["request"]).model_copy(deep=True)
|
|
|
|
# 保存校验后的上下文，供后续节点统一读取
|
|
updated_state["input_context"] = context
|
|
# 初始化自动修正次数，并清空修正建议和失败原因
|
|
updated_state["retry_count"] = 0
|
|
updated_state["repair_plan"] = None
|
|
updated_state["failure_reason"] = None
|
|
# 首次请求进入处理状态，恢复澄清时也使用同一状态
|
|
updated_state["status"] = "running"
|
|
# 返回初始化后的状态，由图接续执行后续节点
|
|
return updated_state
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import json
|
|
|
|
# 1. 创建 state：可将 screen 替换成前端的完整大屏 JSON
|
|
state: QueryState = {
|
|
"request": {
|
|
"prompt": "把销售额统计向右移动 50px",
|
|
"screen": {
|
|
"canvas": {
|
|
"width": 1920,
|
|
"height": 1080,
|
|
"backgroundColor": "#ffffff10",
|
|
},
|
|
"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"]},
|
|
"attachments": [],
|
|
}
|
|
}
|
|
|
|
# 2. 执行节点
|
|
node = InputNode()
|
|
final_state = node.process(state)
|
|
|
|
# 3. 将上下文模型转成可序列化的字典，再打印完整状态
|
|
printable_state = {
|
|
**final_state,
|
|
"input_context": final_state["input_context"].model_dump(
|
|
mode="json", exclude_unset=True
|
|
),
|
|
}
|
|
print(json.dumps(printable_state, ensure_ascii=False, indent=2))
|