176 lines
7.2 KiB
Python
176 lines
7.2 KiB
Python
"""意图识别节点"""
|
|
|
|
from typing import cast
|
|
|
|
from app.clients.decision import close_decision_engine, get_decision_engine
|
|
from app.processor.base import BaseNode
|
|
from app.processor.state import QueryState
|
|
from app.prompt.intent import build_intent_question
|
|
from app.schemas.base import JsonObject
|
|
from app.schemas.decision import DecisionRequest
|
|
from app.schemas.input_context import InputContext
|
|
from app.schemas.requirement import Intent
|
|
|
|
|
|
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,
|
|
"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
|
|
],
|
|
}
|
|
|
|
async def process(self, state: QueryState) -> QueryState:
|
|
# 浅拷贝状态
|
|
updated_state: QueryState = state.copy()
|
|
# 获取输入节点已校验的上下文，包含最新大屏和澄清历史
|
|
context = state["input_context"]
|
|
# 构建 Jev 决策请求，将上下文摘要与意图判断问题一起提交
|
|
request = DecisionRequest(
|
|
# 将 InputNode 的上下文整理为意图判断摘要
|
|
state=self.build_decision_state(context),
|
|
# 使用 intent 作为决策名称，后续通过同名键读取判断结果
|
|
decisions={"intent": build_intent_question()},
|
|
)
|
|
|
|
# 获取决策引擎，用于执行本次意图判断
|
|
decision_engine = get_decision_engine()
|
|
# 异步等待 Jev 返回决策结果，包含选项、置信度和调用信息
|
|
decision_res = await decision_engine.run(request)
|
|
|
|
# 从响应中取出 intent 对应的判断结果
|
|
decision = decision_res.results["intent"]
|
|
|
|
# 如果模型无法明确意图，或置信度低于配置阈值，则转入澄清流程
|
|
if (
|
|
decision.choice == "unclear"
|
|
or decision.confidence < self.config.intent_min_confidence
|
|
):
|
|
intent: Intent = "unclear"
|
|
else:
|
|
# 置信度达标时沿用模型选项，cast 仅作类型提示，不进行运行时转换或校验
|
|
intent = cast(Intent, decision.choice)
|
|
|
|
# 完整保留原始决策，便于查看模型选项及其置信度
|
|
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())
|