"""输入解析、意图分流和需求理解流程""" import inspect from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from langgraph.graph import END, StateGraph from app.config import settings from app.processor.nodes.action_node import ActionNode, can_generate_actions from app.processor.nodes.clarify_node import ClarifyNode from app.processor.nodes.component_resource_node import ComponentResourceNode from app.processor.nodes.fallback_node import FallbackNode from app.processor.nodes.input_node import InputNode from app.processor.nodes.intent_node import IntentNode from app.processor.nodes.planner_node import PlannerNode from app.processor.nodes.qa_node import QANode from app.processor.nodes.reflection_node import ReflectionNode from app.processor.nodes.requirement_node import RequirementNode from app.processor.nodes.scope_node import ScopeNode from app.processor.nodes.validator_node import ValidatorNode from app.processor.nodes.wait_for_user_node import WaitForUserNode from app.processor.state import QueryState def build_checkpoint_serde() -> JsonPlusSerializer: # 允许恢复状态中使用的业务模型类型，使检查点读取后仍可按模型对象访问字段 # 内存与持久化检查点共用，否则持久化读回的状态无法还原为模型对象 return JsonPlusSerializer( allowed_msgpack_modules=[ ("app.schemas.input_context", "InputContext"), ("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.add", "AddAction"), ("app.schemas.actions.delete", "DeleteAction"), ("app.schemas.actions.move", "MoveAction"), ("app.schemas.actions.resize", "ResizeAction"), ("app.schemas.actions.update", "UpdateAction"), ("app.schemas.actions.update_canvas", "UpdateCanvasAction"), ("app.schemas.action_validation", "ActionValidationError"), ("app.schemas.action_repair", "ActionRepairPlan"), ("app.schemas.screen", "Screen"), ("app.schemas.clarification", "Clarification"), ("app.schemas.clarification_reply", "ClarificationReply"), ] ) def route_after_intent(state: QueryState) -> str: # 按最终意图分流：问答进入 QA，意图不明进入澄清，新建或修改进入需求理解 return state["intent"] def route_after_requirement(state: QueryState) -> str: # 需求输出校验失败直接收尾；需要澄清进入澄清，否则进入修改范围判断 if state["status"] == "failed": return "fallback_node" return "clarify_node" if state["requirement"].needs_clarification else "scope_node" def route_after_wait(state: QueryState) -> str: # 用户补充后回到输入节点合并上下文，取消则结束流程 return "input_node" if state["status"] == "running" else END # TODO 目前“资源”只有物料目录，数据源和 RAG 尚未准备好 # 接入后把 needs_planning 与 resource_needs 拆成两个独立开关： # 规划 -> 按需查资源 -> planner；无需规划但有 resource_needs -> 查资源 -> action # 并让 component_resource_node 按 resource_needs 决定查什么，而不是固定全量目录 # 在此之前，无需规划且带 resource_needs 的请求会由 action_block_reason 判为失败 def route_after_scope(state: QueryState) -> str: # 范围输出校验失败或已判定无法执行，直接失败收尾 if state["status"] == "failed": return "fallback_node" # 需要规划时先获取组件能力，供规划节点选择可用组件 if state["change_scope"].needs_planning: return "component_resource_node" # 已支持的简单修改直接生成动作，尚不支持的需求进入失败收尾 return "action_node" if can_generate_actions(state) else "fallback_node" def build_graph(checkpointer=None): # 最大重试次数 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: # 如果动作节点已明确标记失败，直接收尾，不再进入内容修正 if state.get("status") == "failed": return "fallback_node" # validation_passed 为 False：动作生成时解析或结构校验失败，未达重试上限则进入修正分析，已达上限则失败收尾 # validation_passed 为 None：动作已生成但尚未模拟执行，进入 validator_node 检查动作范围和执行结果 # 必须用 is False 区分失败与尚未校验，不能将 None 也当作失败 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" # 使用 QueryState 定义节点之间共享的状态结构 workflow = StateGraph(QueryState) # 为本次构建的图创建节点实例，字典键同时作为注册名称和路由目标 nodes = { "input_node": InputNode(), "intent_node": IntentNode(), "requirement_node": RequirementNode(), "clarify_node": ClarifyNode(), "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 的异步入口 for node_name, node in nodes.items(): # 根据 process 是否为异步函数选择调用入口，保留 BaseNode 的统一执行封装 action = node._acall if inspect.iscoroutinefunction(node.process) else node # 将节点名称与实际调用对象绑定，供后面的边引用 workflow.add_node(node_name, action) # 首次执行从输入节点开始，完成上下文校验后再判断意图 workflow.set_entry_point("input_node") # 输入处理完成后固定进入意图判断 workflow.add_edge("input_node", "intent_node") # 新建和修改进入需求理解，意图不明进入澄清 workflow.add_conditional_edges( source="intent_node", path=route_after_intent, path_map={ "create_screen": "requirement_node", "modify_screen": "requirement_node", "qa": "qa_node", "unclear": "clarify_node", }, ) # 需求信息不足时提问，信息完整时继续判断修改范围 workflow.add_conditional_edges( source="requirement_node", path=route_after_requirement, path_map={ "clarify_node": "clarify_node", "scope_node": "scope_node", "fallback_node": "fallback_node", }, ) # 先保存澄清问题和等待状态，再进入调用 interrupt 的节点暂停执行 workflow.add_edge("clarify_node", "wait_for_user_node") # 问答生成文本后直接结束，不进入规划和动作流程 workflow.add_edge("qa_node", END) # 先读取组件能力再规划，规划后的业务数据查询后续接入 workflow.add_conditional_edges( "scope_node", route_after_scope, { "component_resource_node": "component_resource_node", "action_node": "action_node", "fallback_node": "fallback_node", }, ) # 组件目录查询成功才允许规划，未找到或不支持时进入失败收尾 workflow.add_conditional_edges( "component_resource_node", lambda state: state["component_resources"].status == "resolved", {True: "planner_node", False: "fallback_node"}, ) # 规划完成后交给动作节点转换，动作节点还会检查规划是否可执行 workflow.add_conditional_edges( "planner_node", lambda state: "fallback_node" if state["status"] == "failed" else "action_node", {"action_node": "action_node", "fallback_node": "fallback_node"}, ) # 动作生成后按结果进入校验、修正分析或失败收尾 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, {"input_node": "input_node", END: END}, ) # 仅在调用方未提供检查点存储时创建默认实现，外部传入的持久化存储（如 MongoDB）直接复用 if checkpointer is None: # 将检查点保存在当前进程内存中，支持暂停恢复，但进程退出后不会保留 checkpointer = InMemorySaver(serde=build_checkpoint_serde()) # 编译节点和边，并绑定检查点存储，供执行时按 thread_id 保存和恢复状态 return workflow.compile(checkpointer=checkpointer).with_config( { # 随修正上限增加图执行步数预算，每轮预留分析、生成、校验三步，至少允许 25 步 "recursion_limit": max(25, 12 + 3 * max_retries), } ) # 模块加载时创建共享图实例，复用该实例的调用也共用其检查点存储 query_app = build_graph() if __name__ == "__main__": import argparse import asyncio import json from uuid import uuid4 from langgraph.types import Command from pydantic import BaseModel from app.clients.decision import close_decision_engine from app.observability.tracing import ainvoke_screen, flush_tracing # 选择场景时只切换用户需求，共用同一份画布，方便比较执行结果 examples = { "move": "把销售额统计向右移动 50px，其他内容保持不变", "layout": ( "重新排列现有两个图表，销售额统计放在左侧，销售占比放在右侧，" "统一尺寸为 600×400，画布改为 1600×900，不新增或删除组件" ), "add": "保留现有组件，在下方新增一个柱状图，使用物料示例数据，不需要真实数据绑定", "clarify": "这个图不太对", } parser = argparse.ArgumentParser(description="运行大屏 Agent 完整流程示例") parser.add_argument("--case", choices=examples, default="move", help="选择联调场景") args = parser.parse_args() async def main(): # 1. 创建请求，每次运行使用独立会话，避免混入之前的澄清历史 state: QueryState = { "request": { "prompt": examples[args.case], "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": "销售额统计"}}}, }, { "id": "sales-share", "type": "pie-chart", "name": "销售占比", "layout": { "x": 500, "y": 34.8, "width": 400, "height": 260, }, "props": {"option": {"title": {"text": "销售占比"}}}, }, ], "dataSource": [], }, "context": {"selected_node_ids": ["sales-chart"]}, } } # 2. 执行图，程序退出时关闭共享连接 thread_id = str(uuid4()) session_id = ( state["request"].get("context", {}).get("conversation_id") or thread_id ) print(f"场景：{args.case}\n需求：{examples[args.case]}") if args.case == "clarify": print("澄清回答示例：把销售额统计向右移动 50px，其他内容保持不变") try: final_state = await ainvoke_screen( query_app, state, thread_id=thread_id, session_id=session_id ) # 3. 展示真实澄清问题，用同一个 thread_id 恢复流程 while final_state.get("__interrupt__"): pending = final_state["__interrupt__"][0].value print(pending["question"]) options = pending.get("options", []) for index, option in enumerate(options, 1): marker = "（推荐）" if option["recommended"] else "" print( f"{index}. {option['label']}{marker}：{option['description']}" ) answer = input("输入编号或自由回答（/cancel 取消）：") reply = {"cancel": True} if answer == "/cancel" else {"answer": answer} if answer.isdigit() and 1 <= int(answer) <= len(options): reply = {"option_id": options[int(answer) - 1]["id"]} notes = input("补充说明（可留空）：").strip() if notes: reply["answer"] = notes final_state = await ainvoke_screen( query_app, Command(resume=reply), thread_id=thread_id, session_id=session_id, ) finally: await close_decision_engine() flush_tracing() # 4. 聚焦最终结果，校验失败时 simulated_screen 应为 None result = { "status": final_state.get("status"), "intent": final_state.get("intent"), # 保留需求与规划，方便定位模型输出和执行能力之间的不一致 "requirement": final_state.get("requirement"), "screen_plan": final_state.get("screen_plan"), "validation_passed": final_state.get("validation_passed"), "retry_count": final_state.get("retry_count", 0), "failure_reason": final_state.get("failure_reason"), "validation_errors": final_state.get("validation_errors", []), "actions": final_state.get("actions", []), "simulated_screen": final_state.get("simulated_screen"), "answer": final_state.get("answer"), } print( json.dumps( result, ensure_ascii=False, indent=2, default=lambda value: ( value.model_dump(mode="json") if isinstance(value, BaseModel) else str(value) ), ) ) # 原始坐标单独展示，确认后端试执行没有修改请求中的画布 original = state["request"]["screen"]["nodes"][0]["layout"] print(f"原始销售额坐标：{original['x']}，{original['y']}") asyncio.run(main())