jiawei/my-v0-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完成资源接入

Browse Source
This commit is contained in:
jiawei
2026-09-24 11:58:25 +08:00
parent 5f5d733a9b
commit 57ed92aeba
26 changed files with 3471 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/processor/graph.py
+88 -6
View File
@@ -15,6 +15,10 @@ from app.processor.nodes.qa_node import QANode
from app.processor.nodes.scope_node import ScopeNode
from app.processor.nodes.action_node import ActionNode, can_generate_actions
from app.processor.nodes.validator_node import ValidatorNode
from app.processor.nodes.reflection_node import ReflectionNode
from app.processor.nodes.fallback_node import FallbackNode
from app.processor.nodes.planner_node import PlannerNode
from app.processor.nodes.component_resource_node import ComponentResourceNode
from app.processor.state import QueryState
@@ -30,7 +34,39 @@ def route_after_wait(state: QueryState) -> str:
return "input_node" if state["status"] == "running" else END
def route_after_scope(state: QueryState) -> str:
if state["change_scope"].needs_planning:
return "component_resource_node"
return "action_node" if can_generate_actions(state) else END
def build_graph(checkpointer=None):
from app.config import settings
max_retries = settings.max_action_retries
def route_failure(state: QueryState) -> str:
# 上限检查放在 Reflection 前，避免最后一次无效模型调用
return (
"reflection_node"
if state.get("retry_count", 0) < max_retries
else "fallback_node"
)
def route_after_action(state: QueryState) -> str:
return (
route_failure(state)
if state["validation_passed"] is False
else "validator_node"
)
def route_after_validation(state: QueryState) -> str:
return END if state["validation_passed"] else route_failure(state)
def route_after_reflection(state: QueryState) -> str:
plan = state.get("repair_plan")
return "action_node" if plan is not None and plan.retryable else "fallback_node"
# 定义工作流
workflow = StateGraph(QueryState)
@@ -43,8 +79,12 @@ def build_graph(checkpointer=None):
"wait_for_user_node": WaitForUserNode(),
"qa_node": QANode(),
"scope_node": ScopeNode(),
"planner_node": PlannerNode(),
"component_resource_node": ComponentResourceNode(),
"action_node": ActionNode(),
"validator_node": ValidatorNode(),
"reflection_node": ReflectionNode(),
"fallback_node": FallbackNode(),
}
# 注册节点，异步节点使用 BaseNode 的异步入口
@@ -75,14 +115,49 @@ def build_graph(checkpointer=None):
)
workflow.add_edge("clarify_node", "wait_for_user_node")
workflow.add_edge("qa_node", END)
# 其余能力仍停在范围判断，后续接入 Planner 和 Resource
# 先读取组件能力再规划，规划后的业务数据查询后续接入
workflow.add_conditional_edges(
"scope_node",
can_generate_actions,
{True: "action_node", False: END},
route_after_scope,
{
"component_resource_node": "component_resource_node",
"action_node": "action_node",
END: END,
},
)
workflow.add_edge("action_node", "validator_node")
workflow.add_edge("validator_node", END)
workflow.add_conditional_edges(
"component_resource_node",
lambda state: state["component_resources"].status == "resolved",
{True: "planner_node", False: END},
)
workflow.add_edge("planner_node", END)
workflow.add_conditional_edges(
"action_node",
route_after_action,
{
"validator_node": "validator_node",
"reflection_node": "reflection_node",
"fallback_node": "fallback_node",
},
)
workflow.add_conditional_edges(
"validator_node",
route_after_validation,
{
END: END,
"reflection_node": "reflection_node",
"fallback_node": "fallback_node",
},
)
workflow.add_conditional_edges(
"reflection_node",
route_after_reflection,
{
"action_node": "action_node",
"fallback_node": "fallback_node",
},
)
workflow.add_edge("fallback_node", END)
workflow.add_conditional_edges(
"wait_for_user_node",
route_after_wait,
@@ -96,17 +171,24 @@ def build_graph(checkpointer=None):
("app.schemas.decision", "DecisionResponse"),
("app.schemas.requirement", "Requirement"),
("app.schemas.change_scope", "ChangeScope"),
("app.schemas.screen_plan", "ScreenPlan"),
("app.schemas.component_resource", "ComponentResult"),
("app.schemas.actions.move", "MoveAction"),
("app.schemas.actions.resize", "ResizeAction"),
("app.schemas.actions.update", "UpdateAction"),
("app.schemas.action_validation", "ActionValidationError"),
("app.schemas.action_repair", "ActionRepairPlan"),
("app.schemas.screen", "Screen"),
("app.schemas.clarification", "Clarification"),
("app.schemas.clarification_reply", "ClarificationReply"),
]
)
checkpointer = InMemorySaver(serde=serde)
return workflow.compile(checkpointer=checkpointer)
return workflow.compile(checkpointer=checkpointer).with_config(
{
"recursion_limit": max(25, 12 + 3 * max_retries),
}
)
# 创建全局图实例
app/processor/nodes/action_node.py
+54 -45
View File
@@ -3,12 +3,15 @@
import json
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.exceptions import OutputParserException
from pydantic import BaseModel, ValidationError
from app.clients.llm import get_llm
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.prompt.action import ACTION_SYSTEM_PROMPT
from app.schemas.actions import ActionBatch
from app.schemas.action_validation import ActionValidationError, schema_errors
def can_generate_actions(state: QueryState) -> bool:
@@ -38,57 +41,63 @@ class ActionNode(BaseNode):
name = "action_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
if not can_generate_actions(state):
raise ValueError(
"ActionNode 当前只支持无需规划和资源的 move、resize、update 需求"
)
batch = await get_llm(ActionBatch).ainvoke(
[
SystemMessage(content=ACTION_SYSTEM_PROMPT),
HumanMessage(
content=json.dumps(
{
"input_context": state["input_context"].model_dump(
mode="json"
),
"requirement": state["requirement"].model_dump(mode="json"),
"change_scope": state["change_scope"].model_dump(
mode="json"
),
},
ensure_ascii=False,
)
),
]
)
batch = ActionBatch.model_validate(batch.model_dump())
allowed = {}
for change in state["requirement"].changes:
for target_id in change.target_ids:
allowed.setdefault(target_id, set()).add(change.operation)
existing = {node.id for node in state["input_context"].screen.nodes}
for action in batch.actions:
if action.target_id not in existing:
raise ValueError(f"动作引用了不存在的组件: {action.target_id}")
if action.type not in allowed.get(action.target_id, set()):
raise ValueError(
f"动作超出需求范围: {action.target_id} / {action.type}"
)
expected = {
(target, operation) for target, ops in allowed.items() for operation in ops
payload = {
"input_context": state["input_context"].model_dump(mode="json"),
"requirement": state["requirement"].model_dump(mode="json"),
"change_scope": state["change_scope"].model_dump(mode="json"),
}
actual = {(action.target_id, action.type) for action in batch.actions}
if expected - actual:
raise ValueError(
f"动作遗漏了需求中的目标或操作: {sorted(expected - actual)}"
if state.get("repair_plan") is not None:
payload.update(
repair_plan=state["repair_plan"].model_dump(mode="json"),
previous_actions=[
action.model_dump(mode="json")
for action in state.get("actions") or []
],
validation_errors=[
error.model_dump() for error in state.get("validation_errors", [])
],
)
return {
**state,
"actions": batch.actions,
"validation_passed": None,
"validation_errors": [],
"simulated_screen": None,
}
# 配置和网络错误继续抛出，仅模型输出错误进入内容修正循环
llm_client = get_llm(ActionBatch)
updated_state["actions"] = None
updated_state["validation_passed"] = None
updated_state["validation_errors"] = []
updated_state["simulated_screen"] = None
updated_state["failure_reason"] = None
updated_state["status"] = "running"
try:
result = await llm_client.ainvoke(
[
SystemMessage(content=ACTION_SYSTEM_PROMPT),
HumanMessage(content=json.dumps(payload, ensure_ascii=False)),
]
)
batch = ActionBatch.model_validate(
result.model_dump() if isinstance(result, BaseModel) else result
)
except ValidationError as error:
updated_state["validation_passed"] = False
updated_state["validation_errors"] = schema_errors(error)
return updated_state
except OutputParserException:
updated_state["validation_passed"] = False
updated_state["validation_errors"] = [
ActionValidationError(
code="generation_error",
message="模型输出无法解析为 ActionBatch，请按协议重新生成",
)
]
return updated_state
updated_state["actions"] = batch.actions
return updated_state
if __name__ == "__main__":
app/processor/nodes/clarify_node.py
+9 -5
View File
@@ -15,24 +15,28 @@ class ClarifyNode(BaseNode):
name = "clarify_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
intent = state["intent"]
print(intent)
print("=================")
payload = {
"intent": intent,
"input_context": state["input_context"].model_dump(mode="json"),
}
if intent != "unclear":
# 以需求节点的问题为依据补齐选项，不重新判断意图
payload["requirement"] = state["requirement"].model_dump(mode="json")
llm = get_llm(Clarification)
clarification = await llm.ainvoke(
llm_client = get_llm(Clarification)
clarification = await llm_client.ainvoke(
[
SystemMessage(content=CLARIFY_SYSTEM_PROMPT),
HumanMessage(content=json.dumps(payload, ensure_ascii=False)),
]
)
return {**state, "clarification": clarification}
updated_state["clarification"] = clarification
return updated_state
if __name__ == "__main__":
app/processor/nodes/component_resource_node.py
+40
View File
@@ -0,0 +1,40 @@
"""为规划提供前端实际支持的物料目录"""
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.resources.component import ComponentProvider
from app.schemas.component_resource import ComponentQuery
class ComponentResourceNode(BaseNode):
name = "component_resource_node"
def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
# 当前目录较小，先提供全量，避免关键词筛选遗漏可用组件
result = ComponentProvider().query(ComponentQuery())
updated_state["component_resources"] = result
updated_state["screen_plan"] = None
if result.status != "resolved":
updated_state["status"] = "failed"
updated_state["failure_reason"] = result.message
else:
updated_state["status"] = "running"
updated_state["failure_reason"] = None
return updated_state
if __name__ == "__main__":
import json
state: QueryState = {"request": {}}
node = ComponentResourceNode()
final_state = node.process(state)
print(
json.dumps(
final_state,
ensure_ascii=False,
indent=2,
default=lambda value: value.model_dump(mode="json"),
)
)
app/processor/nodes/fallback_node.py
+37
View File
@@ -0,0 +1,37 @@
"""停止自动修正并保留诊断信息"""
from app.processor.base import BaseNode
from app.processor.state import QueryState
class FallbackNode(BaseNode):
name = "fallback_node"
def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
reason = state.get("failure_reason")
if not reason:
reason = f"动作校验未通过，已达到自动修正上限，共重试 {state.get('retry_count', 0)} 次"
# actions 仅保留作诊断，不交付部分结果
updated_state["status"] = "failed"
updated_state["failure_reason"] = reason
updated_state["validation_passed"] = False
updated_state["simulated_screen"] = None
return updated_state
if __name__ == "__main__":
from app.schemas.action_validation import ActionValidationError
state = {
"request": {},
"retry_count": 2,
"actions": None,
"validation_errors": [
ActionValidationError(code="generation_error", message="动作生成失败")
],
}
node = FallbackNode()
final_state = node.process(state)
print(final_state)
app/processor/nodes/input_node.py
+30 -13
View File
@@ -11,13 +11,16 @@ class InputNode(BaseNode):
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,
}
if reply.option_id is not None:
option = next(
option
@@ -26,6 +29,7 @@ class InputNode(BaseNode):
)
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")
# 新快照不沿用可能已经失效的选中节点
@@ -35,22 +39,35 @@ class InputNode(BaseNode):
if reply.context is not None:
data["context"] = reply.context.model_dump(mode="json")
context = InputContext.model_validate(data)
# 显式清空旧推导结果，LangGraph 的状态合并不会删除省略字段
return {
**state,
"input_context": context,
"clarification_reply": None,
"intent_decision": None,
"intent": None,
"requirement": None,
"change_scope": None,
"clarification": None,
"answer": None,
"status": "running",
}
updated_state["input_context"] = context
updated_state["clarification_reply"] = None
updated_state["intent_decision"] = None
updated_state["intent"] = None
updated_state["requirement"] = 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: QueryState = {**state, "input_context": context}
updated_state["input_context"] = context
updated_state["retry_count"] = 0
updated_state["repair_plan"] = None
updated_state["failure_reason"] = None
return updated_state
app/processor/nodes/intent_node.py
+4 -5
View File
@@ -36,6 +36,7 @@ class IntentNode(BaseNode):
"layout": node.layout.model_dump(mode="json", exclude_unset=True),
}
)
return {
"prompt": context.prompt,
"clarification_history": context.clarification_history,
@@ -66,6 +67,7 @@ class IntentNode(BaseNode):
}
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
context = state["input_context"]
# 构建jev请求
request = DecisionRequest(
@@ -89,11 +91,8 @@ class IntentNode(BaseNode):
else:
intent = cast(Intent, decision.choice)
updated_state: QueryState = {
**state,
"intent_decision": decision_res,
"intent": intent,
}
updated_state["intent_decision"] = decision_res
updated_state["intent"] = intent
return updated_state
app/processor/nodes/planner_node.py
+156
View File
@@ -0,0 +1,156 @@
"""将需要规划的需求整理为 ScreenPlan"""
import json
from langchain_core.messages import HumanMessage, SystemMessage
from app.clients.llm import get_llm
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.prompt.planner import PLANNER_SYSTEM_PROMPT
from app.schemas.screen_plan import ScreenPlan
class PlannerNode(BaseNode):
name = "planner_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
requirement = state["requirement"]
scope = state["change_scope"]
if (
requirement.needs_clarification
or requirement.intent
not in (
"create_screen",
"modify_screen",
)
or not scope.needs_planning
):
raise ValueError("PlannerNode 只处理信息完整且需要规划的新建或修改需求")
context = state["input_context"]
resources = state.get("component_resources")
if resources is not None and resources.status != "resolved":
raise ValueError("组件目录尚未就绪，不能继续规划")
result = await get_llm(ScreenPlan).ainvoke(
[
SystemMessage(content=PLANNER_SYSTEM_PROMPT),
HumanMessage(
content=json.dumps(
{
"input_context": context.model_dump(mode="json"),
"requirement": requirement.model_dump(mode="json"),
"change_scope": scope.model_dump(mode="json"),
"component_resources": resources.model_dump(mode="json")
if resources is not None
else None,
},
ensure_ascii=False,
)
),
]
)
plan = ScreenPlan.model_validate(result.model_dump())
if resources is not None:
supported_types = {component.type for component in resources.components}
for component in plan.components:
if component.disposition == "add" and component.component_type is None:
raise ValueError(f"新增规划组件未指定物料类型: {component.plan_id}")
if (
component.disposition in ("add", "modify")
and component.component_type is not None
):
if component.component_type not in supported_types:
raise ValueError(
f"规划使用了目录中不存在的组件类型: {component.component_type}"
)
existing_ids = {node.id for node in context.screen.nodes}
planned_ids = {
component.target_id
for component in plan.components
if component.target_id is not None
}
if unknown := planned_ids - existing_ids:
raise ValueError(f"规划引用了不存在的组件: {sorted(unknown)}")
required_ids = {
target for change in requirement.changes for target in change.target_ids
}
if missing := required_ids - planned_ids:
raise ValueError(f"规划遗漏了需求中的组件: {sorted(missing)}")
if scope.scope == "local":
changed_ids = {
component.target_id
for component in plan.components
if component.disposition in ("modify", "remove")
}
if outside := changed_ids - set(scope.target_ids):
raise ValueError(f"局部规划修改了范围外的组件: {sorted(outside)}")
# 原始约束和假设不能在规划阶段被无声丢弃
plan.constraints = list(
dict.fromkeys([*requirement.constraints, *plan.constraints])
)
plan.assumptions = list(
dict.fromkeys([*requirement.assumptions, *plan.assumptions])
)
updated_state["screen_plan"] = plan
return updated_state
if __name__ == "__main__":
import asyncio
from pydantic import BaseModel
from app.processor.nodes.component_resource_node import ComponentResourceNode
from app.processor.nodes.input_node import InputNode
from app.schemas.change_scope import ChangeScope
from app.schemas.requirement import Requirement
async def main():
# 1. 创建 state，直接提供需求和范围，单独测试规划
state = InputNode().process(
{
"request": {
"prompt": "创建销售大屏，上方展示销售总额，下方展示月度趋势",
"screen": {"canvas": {"width": 1920, "height": 1080}, "nodes": []},
}
}
)
state["requirement"] = Requirement(
intent="create_screen",
goal="创建销售总览大屏",
needs_clarification=False,
changes=[
{
"target_scope": "screen",
"operation": "add",
"parameters": {"metrics": ["销售总额", "月度销售额"]},
}
],
constraints=["总额在上，趋势在下"],
)
state["change_scope"] = ChangeScope(
scope="screen",
needs_planning=True,
resource_needs=["销售总额和月度销售数据，用于指标与趋势展示"],
reason="需要安排区域和组件组合",
)
# 2. 执行节点并打印完整 state，仅调用 LLM
node = PlannerNode()
state = ComponentResourceNode().process(state)
final_state = await node.process(state)
print(
json.dumps(
final_state,
ensure_ascii=False,
indent=2,
default=lambda value: (
value.model_dump(mode="json")
if isinstance(value, BaseModel)
else str(value)
),
)
)
asyncio.run(main())
app/processor/nodes/qa_node.py
+3 -1
View File
@@ -13,6 +13,7 @@ class QANode(BaseNode):
name = "qa_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
llm_client = get_llm()
llm_res = await llm_client.ainvoke(
[
@@ -26,7 +27,8 @@ class QANode(BaseNode):
answer = StrOutputParser().invoke(llm_res).strip()
if not answer:
raise ValueError("问答模型未返回文本回答")
return {**state, "answer": answer}
updated_state["answer"] = answer
return updated_state
if __name__ == "__main__":
app/processor/nodes/reflection_node.py
+128
View File
@@ -0,0 +1,128 @@
"""分析失败动作，生成下一次修正建议"""
import json
from langchain_core.exceptions import OutputParserException
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, ValidationError
from app.clients.llm import get_llm
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.prompt.reflection import REFLECTION_SYSTEM_PROMPT
from app.schemas.action_repair import ActionRepairPlan
class ReflectionNode(BaseNode):
name = "reflection_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
llm_client = get_llm(ActionRepairPlan)
payload = {
"input_context": state["input_context"].model_dump(mode="json"),
"requirement": state["requirement"].model_dump(mode="json"),
"actions": [
action.model_dump(mode="json") for action in state.get("actions") or []
],
"validation_errors": [
error.model_dump() for error in state["validation_errors"]
],
"previous_repair_plan": state["repair_plan"].model_dump()
if state.get("repair_plan")
else None,
}
try:
result = await llm_client.ainvoke(
[
SystemMessage(content=REFLECTION_SYSTEM_PROMPT),
HumanMessage(content=json.dumps(payload, ensure_ascii=False)),
]
)
plan = ActionRepairPlan.model_validate(
result.model_dump() if isinstance(result, BaseModel) else result
)
except (ValidationError, OutputParserException):
updated_state["repair_plan"] = None
updated_state["failure_reason"] = "未能生成有效的动作修正建议"
return updated_state
updated_state["repair_plan"] = plan
updated_state["retry_count"] = state.get("retry_count", 0) + int(plan.retryable)
updated_state["failure_reason"] = None if plan.retryable else plan.reason
return updated_state
if __name__ == "__main__":
import asyncio
from app.processor.nodes.input_node import InputNode
from app.schemas.action_validation import ActionValidationError
from app.schemas.actions import UpdateAction
from app.schemas.requirement import Requirement
async def main():
state = InputNode().process(
{
"request": {
"prompt": "把标题改成月度销售额",
"screen": {
"canvas": {"width": 1920, "height": 1080},
"nodes": [
{
"id": "chart",
"type": "chart",
"layout": {
"x": 100,
"y": 200,
"width": 400,
"height": 300,
},
"props": {"title": "销售额"},
}
],
},
}
}
)
state["requirement"] = Requirement(
intent="modify_screen",
goal="修改标题",
needs_clarification=False,
changes=[
{
"target_scope": "nodes",
"target_ids": ["chart"],
"operation": "update",
"parameters": {"title": "月度销售额"},
}
],
)
state["actions"] = [
UpdateAction(
target_id="chart",
section="props",
path=["option", "title"],
value="月度销售额",
)
]
state["validation_passed"] = False
state["validation_errors"] = [
ActionValidationError(
action_index=0,
code="execution_error",
message="中间路径不存在: props ['option']",
)
]
node = ReflectionNode()
final_state = await node.process(state)
print(
json.dumps(
final_state,
ensure_ascii=False,
indent=2,
default=lambda value: value.model_dump(mode="json"),
)
)
asyncio.run(main())
app/processor/nodes/requirement_node.py
+4 -1
View File
@@ -15,12 +15,14 @@ class RequirementNode(BaseNode):
name = "requirement_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
context = state["input_context"]
intent = state["intent"]
payload = {
"intent": intent,
"input_context": context.model_dump(mode="json", exclude_unset=True),
}
llm_client = get_llm(Requirement)
llm_res = await llm_client.ainvoke(
[
@@ -38,7 +40,8 @@ class RequirementNode(BaseNode):
if unknown_ids := set(change.target_ids) - node_ids:
raise ValueError(f"需求引用了不存在的节点: {sorted(unknown_ids)}")
return {**state, "requirement": requirement}
updated_state["requirement"] = requirement
return updated_state
if __name__ == "__main__":
app/processor/nodes/scope_node.py
+9 -1
View File
@@ -15,12 +15,15 @@ class ScopeNode(BaseNode):
name = "scope_node"
async def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
requirement = state["requirement"]
if requirement.needs_clarification or requirement.intent not in (
"create_screen",
"modify_screen",
):
raise ValueError("ScopeNode 只处理信息完整的新建或修改需求")
context = state["input_context"]
llm_client = get_llm(ChangeScope)
scope = await llm_client.ainvoke(
@@ -39,15 +42,20 @@ class ScopeNode(BaseNode):
),
]
)
existing_ids = {node.id for node in context.screen.nodes}
if unknown_ids := set(scope.target_ids) - existing_ids:
raise ValueError(f"范围引用了不存在的节点: {sorted(unknown_ids)}")
required_ids = {
node_id for change in requirement.changes for node_id in change.target_ids
}
if missing_ids := required_ids - set(scope.target_ids):
raise ValueError(f"范围遗漏了需求中的节点: {sorted(missing_ids)}")
return {**state, "change_scope": scope}
updated_state["change_scope"] = scope
return updated_state
if __name__ == "__main__":
app/processor/nodes/validator_node.py
+47 -36
View File
@@ -5,7 +5,7 @@ from pydantic import BaseModel, ValidationError
from app.actions.registry import apply_action
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.schemas.action_validation import ActionValidationError
from app.schemas.action_validation import ActionValidationError, schema_errors
from app.schemas.actions import ActionBatch
@@ -13,6 +13,7 @@ class ValidatorNode(BaseNode):
name = "validator_node"
def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
actions = state["actions"]
try:
# 重新校验模型内容，避免赋值后绕过字段校验
@@ -27,27 +28,42 @@ class ValidatorNode(BaseNode):
}
)
except ValidationError as error:
errors = []
for item in error.errors(include_url=False, include_input=False):
location = item["loc"]
index = (
location[1]
if len(location) > 1 and isinstance(location[1], int)
else None
errors = schema_errors(error)
updated_state["validation_passed"] = False
updated_state["validation_errors"] = errors
updated_state["simulated_screen"] = None
return updated_state
# 需求范围检查集中在校验节点，错误统一进入修正循环
requirement = state.get("requirement")
if requirement is not None:
allowed = {
(target, change.operation)
for change in requirement.changes
for target in change.target_ids
}
actual = {(action.target_id, action.type) for action in batch.actions}
errors = [
ActionValidationError(
action_index=index,
code="scope_error",
message=f"动作超出需求范围: {action.target_id} / {action.type}",
)
for index, action in enumerate(batch.actions)
if (action.target_id, action.type) not in allowed
]
if missing := allowed - actual:
errors.append(
ActionValidationError(
action_index=index,
code="schema_error",
message=item["msg"],
code="scope_error",
message=f"动作遗漏了需求中的目标或操作: {sorted(missing)}",
)
)
return {
**state,
"validation_passed": False,
"validation_errors": errors,
"simulated_screen": None,
}
if errors:
updated_state["validation_passed"] = False
updated_state["validation_errors"] = errors
updated_state["simulated_screen"] = None
return updated_state
screen = state["input_context"].screen.model_copy(deep=True)
for index, action in enumerate(batch.actions):
@@ -55,25 +71,20 @@ class ValidatorNode(BaseNode):
screen = apply_action(screen, action)
except ValueError as error:
# 首个失败即停止，后续动作可能依赖前序结果，不返回部分成功快照
return {
**state,
"validation_passed": False,
"validation_errors": [
ActionValidationError(
action_index=index,
code="execution_error",
message=str(error),
)
],
"simulated_screen": None,
}
return {
**state,
"actions": batch.actions,
"validation_passed": True,
"validation_errors": [],
"simulated_screen": screen,
}
updated_state["validation_passed"] = False
updated_state["validation_errors"] = [
ActionValidationError(
action_index=index, code="execution_error", message=str(error)
)
]
updated_state["simulated_screen"] = None
return updated_state
updated_state["actions"] = batch.actions
updated_state["validation_passed"] = True
updated_state["validation_errors"] = []
updated_state["simulated_screen"] = screen
return updated_state
if __name__ == "__main__":
app/processor/nodes/wait_for_user_node.py
+5 -5
View File
@@ -12,6 +12,7 @@ class WaitForUserNode(BaseNode):
name = "wait_for_user_node"
def process(self, state: QueryState) -> QueryState:
updated_state: QueryState = state.copy()
payload = state["clarification"].model_dump(mode="json")
while True:
# interrupt 的控制异常必须交给 LangGraph 处理
@@ -31,8 +32,7 @@ class WaitForUserNode(BaseNode):
**payload,
"error": "请选择本题有效选项或填写回答，并提供有效的大屏数据；也可设置 cancel=true 取消",
}
return {
**state,
"clarification_reply": reply,
"status": "cancelled" if reply.cancel else "running",
}
updated_state["clarification_reply"] = reply
updated_state["status"] = "cancelled" if reply.cancel else "running"
return updated_state
app/processor/state.py
+15 -1
View File
@@ -1,15 +1,18 @@
from typing import Literal, NotRequired, TypedDict
from app.schemas.action_repair import ActionRepairPlan
from app.schemas.action_validation import ActionValidationError
from app.schemas.actions import Action
from app.schemas.base import JsonObject
from app.schemas.change_scope import ChangeScope
from app.schemas.clarification import Clarification
from app.schemas.clarification_reply import ClarificationReply
from app.schemas.component_resource import ComponentResult
from app.schemas.decision import DecisionResponse
from app.schemas.input_context import InputContext
from app.schemas.requirement import Intent, Requirement
from app.schemas.screen import Screen
from app.schemas.screen_plan import ScreenPlan
class QueryState(TypedDict):
@@ -19,6 +22,12 @@ class QueryState(TypedDict):
intent: NotRequired[Intent | None] # 经过置信度门控的最终意图
requirement: NotRequired[Requirement | None] # 结构化需求及信息完整性判断
change_scope: NotRequired[ChangeScope | None] # 修改范围、规划和资源需求
screen_plan: NotRequired[
ScreenPlan | None
] # 大屏设计规划，尚未获取业务数据或生成动作
component_resources: NotRequired[
ComponentResult | None
] # 已查询的前端物料能力及目录版本
actions: NotRequired[
list[Action] | None
] # 生成的动作列表，是否可模拟执行由校验结果标记
@@ -32,8 +41,13 @@ class QueryState(TypedDict):
Screen | None
] # 整批通过后的模拟快照，不覆盖原始上下文
clarification: NotRequired[Clarification | None] # 当前澄清问题及选项
repair_plan: NotRequired[ActionRepairPlan | None] # 最近一次动作修正建议
retry_count: NotRequired[int] # 已启动的自动修正次数，不含首次生成
failure_reason: NotRequired[str | None] # 停止自动修正后的失败原因
clarification_reply: NotRequired[
ClarificationReply | None
] # 待合并的用户回答或取消指令
answer: NotRequired[str | None] # 问答节点生成的回答或建议
status: NotRequired[Literal["running", "cancelled"]] # 流程继续或用户取消
status: NotRequired[
Literal["running", "cancelled", "failed"]
] # 流程继续、用户取消或动作修正失败