"""意图识别节点""" from app.clients.decision import close_decision_engine, get_decision_engine from app.processor.base import BaseNode from app.processor.intent_policy import resolve_intent, resolve_offer_response from app.processor.state import QueryState from app.prompt.intent import build_intent_question, build_offer_question from app.schemas.base import JsonObject from app.schemas.decision import DecisionRequest from app.schemas.input_context import InputContext class IntentNode(BaseNode): name = "intent_node" @staticmethod def build_decision_state(context: InputContext) -> JsonObject: """只整理判断意图需要的摘要，完整 Screen 仍保留在输入上下文""" # 收集每个组件的识别信息和布局，帮助判断用户指的是哪个组件 nodes: list[JsonObject] = [] for node in context.screen.nodes: # 获取图表配置，组件没有 option 时返回 None option = node.props.get("option") # 只有字典配置才能读取 title，其他类型按没有标题处理 title = option.get("title") if isinstance(option, dict) else None # 将单个标题统一包装成列表，兼容图表配置中的多个标题 titles = title if isinstance(title, list) else [title] # 保存组件摘要，不传递完整 props，减少意图判断时的无关信息 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, "history": context.history, "clarification_history": context.clarification_history, "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 ], } @staticmethod def pending_offer(context: InputContext) -> str | None: """只有紧接着的上一轮留下的提议才等待确认；澄清回答针对的是澄清问题，不是提议""" if context.clarification_history or not context.history: return None offer = context.history[-1].get("offer") return offer if isinstance(offer, str) and offer.strip() else None async def process(self, state: QueryState) -> QueryState: # 浅拷贝状态 updated_state: QueryState = state.copy() # 获取输入节点已校验的上下文，包含最新大屏和澄清历史 context = state["input_context"] # 上一轮助手提出了待确认的修改，且本轮不是澄清回答时，额外判断用户是否同意 offer = self.pending_offer(context) decision_state = self.build_decision_state(context) # 使用 intent 作为决策名称，后续通过同名键读取判断结果 decisions = {"intent": build_intent_question()} if offer is not None: decision_state["pending_offer"] = offer decisions["offer_response"] = build_offer_question() # 构建 Jev 决策请求，同一次调用提交全部判断 request = DecisionRequest(state=decision_state, decisions=decisions) # 获取决策引擎，用于执行本次意图判断 decision_engine = get_decision_engine() # 异步等待 Jev 返回决策结果，包含选项、置信度和调用信息 decision_res = await decision_engine.run(request) # 从响应中取出 intent 对应的判断结果 decision = decision_res.results["intent"] # 用户明确回应了上一轮提议时，直接确定意图，不再经过常规门控 intent = None if offer is not None: intent = resolve_offer_response( decision_res.results["offer_response"], accept_min_probability=self.config.intent_min_probability, decline_min_probability=self.config.intent_qa_min_probability, ) # 同意时把提议交给需求理解，作为本次要执行的修改 updated_state["accepted_offer"] = offer if intent == "modify_screen" else None if intent is None: # 用户刚点选了澄清选项时，选择本身已表达明确意图 history = context.clarification_history # 按选中项概率、风险和澄清轮数确定最终意图，规则集中在 intent_policy intent = resolve_intent( decision, action_min_probability=self.config.intent_min_probability, qa_min_probability=self.config.intent_qa_min_probability, clarified_rounds=len(history), max_clarify_rounds=self.config.intent_max_clarify_rounds, option_selected=bool(history and history[-1].get("selected_option")), ) # 完整保留原始决策，便于查看模型选项及其置信度 updated_state["intent_decision"] = decision_res # 保存经过置信度判断的最终意图，供后续路由选择处理分支 updated_state["intent"] = intent # 返回更新后的状态，由图继续执行对应的处理节点 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，退出时关闭 Jev 连接 state = InputNode().process(state) node = IntentNode() try: final_state = await node.process(state) finally: # 无论调用成功还是抛出异常，都关闭示例使用的 Jev 连接 await close_decision_engine() # 3. 打印完整 state，包含选项、概率、置信度和调用信息 # 将状态中的 Pydantic 模型转成字典，其他值保持原样，以便输出 JSON 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())