908705107/ai-agent-admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restore AI workflow design nodes

Browse Source
This commit is contained in:
cls_宁波本机
2026-06-09 21:18:33 +08:00
parent fbd726ebdf
commit cb45c0a068
71 changed files with 22676 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-fastapi/ai_platform/nodes/builtin/dashboard_design_node.py
+483
View File
@@ -0,0 +1,483 @@
"""
仪表盘设计节点
提供可视化设计界面，让用户设计仪表盘布局
"""
import ast
import json
import logging
import random
from typing import Any, Dict, List, Optional
from uuid import uuid4
from ..base import BaseNode, NodeContext, NodeResult
from ..registry import NodeRegistry
logger = logging.getLogger(__name__)
@NodeRegistry.register
class DashboardDesignNode(BaseNode):
"""
仪表盘设计节点
提供可视化设计界面，让用户设计仪表盘布局和组件
"""
node_type = 'dashboard_design'
node_name = '仪表盘设计'
node_category = 'dashboard'
node_icon = 'layout-dashboard'
node_description = '可视化设计仪表盘布局和组件'
inputs = [
{
'name': 'dashboard_code',
'type': 'string',
'description': '仪表盘编码（来自基础信息节点）',
'required': False,
},
{
'name': 'design_suggestion',
'type': 'string',
'description': 'LLM生成的设计建议（可选）',
'required': False,
},
{
'name': 'initial_config',
'type': 'object',
'description': '初始页面配置（可选）',
'required': False,
},
]
outputs = [
{
'name': 'page_config',
'type': 'object',
'description': '页面设计配置（dashboard-design的JSON配置）',
},
{
'name': 'design_title',
'type': 'string',
'description': '设计方案标题',
},
{
'name': 'confirmed',
'type': 'boolean',
'description': '是否已确认',
},
]
def execute(self, context: NodeContext) -> NodeResult:
"""执行节点"""
try:
# 检查是否有用户提交的设计配置
user_input = context.variables.get('__user_input__')
if user_input:
design_data = self._parse_user_design(user_input)
if design_data:
logger.info(f'DashboardDesignNode - 使用用户设计的配置')
context.variables.pop('__user_input__', None)
return self._build_result(design_data, confirmed=True)
# 获取输入参数
dashboard_code = context.resolve_template(self.config.get('dashboard_code', ''))
design_suggestion = context.resolve_template(self.config.get('design_suggestion', ''))
initial_config = self.config.get('initial_config', {})
# 如果initial_config是变量引用，解析它
if isinstance(initial_config, str) and initial_config.startswith('{{'):
initial_config = self._resolve_variable_value(context, initial_config) or {}
logger.info(f'DashboardDesignNode - dashboard_code={dashboard_code}')
# 检查是否需要确认（支持布尔值、字符串模式和变量引用）
require_confirmation = self.resolve_require_confirmation(context, default=True)
design_title = self.config.get('design_title', '仪表盘设计')
# 如果没有初始配置，根据 design_suggestion 生成默认模板
if not initial_config or not initial_config.get('widgets'):
initial_config = self._generate_dashboard_template(design_suggestion, dashboard_code)
# 构建设计数据
design_data = {
'dashboard_code': dashboard_code,
'design_title': design_title,
'design_suggestion': design_suggestion,
'page_config': initial_config,
}
if require_confirmation:
return NodeResult(
success=True,
output={
'page_config': design_data.get('page_config', {}),
'design_title': design_title,
'confirmed': False,
},
waiting_for_input=True,
waiting_config={
'type': 'design_preview',
'preview_type': 'dashboard_design',
'title': design_title,
'data': design_data,
'editable': True,
'fullscreen': True,
},
)
return self._build_result(design_data, confirmed=True)
except Exception as e:
logger.error(f'DashboardDesignNode - 执行失败: {str(e)}')
return NodeResult(
success=False,
error=f'仪表盘设计节点执行失败: {str(e)}',
)
def _parse_user_design(self, user_input: Any) -> Optional[Dict[str, Any]]:
"""解析用户提交的设计配置"""
try:
if isinstance(user_input, str):
data = json.loads(user_input)
else:
data = user_input
if isinstance(data, dict):
return data
return None
except Exception as e:
logger.error(f'解析用户设计配置失败: {e}')
return None
def _build_result(self, design_data: Dict[str, Any], confirmed: bool) -> NodeResult:
"""构建节点结果"""
page_config = design_data.get('page_config', {})
design_title = design_data.get('design_title', '仪表盘设计')
output_data = {
'page_config': page_config,
'design_title': design_title,
'confirmed': confirmed,
}
return NodeResult(
success=True,
output=output_data,
output_variables=output_data, # 添加 output_variables
)
def _resolve_variable_value(self, context: NodeContext, value: Any) -> Any:
"""解析变量引用"""
if isinstance(value, str):
resolved = context.resolve_template(value)
if resolved != value:
try:
return json.loads(resolved) if isinstance(resolved, str) else resolved
except (json.JSONDecodeError, TypeError):
return resolved
return value
def _generate_dashboard_template(self, design_suggestion: str, dashboard_code: str) -> Dict[str, Any]:
"""根据 LLM 建议生成仪表盘模板"""
try:
# 解析 design_suggestion（可能是字符串形式的字典）
modules = self._parse_design_suggestion(design_suggestion)
# 生成基础模板
template = {
"id": str(uuid4()),
"name": "我的仪表盘",
"columns": 12,
"rowHeight": 50,
"margin": [12, 12],
"widgets": []
}
# 添加欢迎卡片
welcome_id = str(uuid4())
template["widgets"].append({
"id": welcome_id,
"i": welcome_id,
"type": "welcome-card",
"x": 0,
"y": 0,
"w": 8,
"h": 2,
"minW": 4,
"minH": 2,
"title": "欢迎卡片",
"props": {
"title": "欢迎回来",
"subtitle": "今天是个好日子",
"showTime": True,
"showWeather": False
}
})
# 添加天气组件
weather_id = str(uuid4())
template["widgets"].append({
"id": weather_id,
"i": weather_id,
"type": "weather",
"x": 8,
"y": 0,
"w": 4,
"h": 2,
"minW": 2,
"minH": 2,
"title": "天气",
"props": {
"title": "今日天气",
"city": "北京",
"temperature": 25,
"weather": "晴",
"humidity": 45,
"wind": "东北风 3级",
"icon": "sunny"
}
})
# 添加日历
calendar_id = str(uuid4())
template["widgets"].append({
"id": calendar_id,
"i": calendar_id,
"type": "calendar",
"x": 8,
"y": 2,
"w": 4,
"h": 5,
"minW": 3,
"minH": 4,
"title": "日历",
"props": {
"title": "日历",
"showLunar": False
}
})
# 根据模块生成图表
y_offset = 2
if modules and len(modules) > 0:
# 生成折线图（基于第一个模块）
module1 = modules[0] if len(modules) > 0 else {"module_name": "数据", "moduld_fields": []}
line_chart = self._generate_line_chart(module1, 0, y_offset)
template["widgets"].append(line_chart)
y_offset += 5
# 生成柱状图（基于第二个模块）
module2 = modules[1] if len(modules) > 1 else module1
bar_chart = self._generate_bar_chart(module2, 0, y_offset)
template["widgets"].append(bar_chart)
# 生成饼图（基于第三个模块）
module3 = modules[2] if len(modules) > 2 else module1
pie_chart = self._generate_pie_chart(module3, 8, y_offset)
template["widgets"].append(pie_chart)
else:
# 没有模块信息，使用默认图表
template["widgets"].extend([
self._generate_line_chart({"module_name": "访问趋势", "moduld_fields": []}, 0, y_offset),
self._generate_bar_chart({"module_name": "销售统计", "moduld_fields": []}, 0, y_offset + 5),
self._generate_pie_chart({"module_name": "流量来源", "moduld_fields": []}, 8, y_offset + 5)
])
return template
except Exception as e:
logger.error(f'生成仪表盘模板失败: {e}')
# 返回空模板
return {
"id": str(uuid4()),
"name": "我的仪表盘",
"columns": 12,
"rowHeight": 50,
"margin": [12, 12],
"widgets": []
}
def _parse_design_suggestion(self, design_suggestion: str) -> List[Dict[str, Any]]:
"""解析 LLM 生成的设计建议"""
try:
if not design_suggestion:
return []
# 尝试解析为 JSON
if isinstance(design_suggestion, str):
try:
data = json.loads(design_suggestion)
except json.JSONDecodeError:
# 尝试使用 ast.literal_eval
try:
data = ast.literal_eval(design_suggestion)
except (ValueError, SyntaxError):
return []
else:
data = design_suggestion
# 提取模块信息
if isinstance(data, dict):
modules = data.get('app_modules', [])
if isinstance(modules, list):
return modules
return []
except Exception as e:
logger.error(f'解析设计建议失败: {e}')
return []
def _generate_line_chart(self, module: Dict[str, Any], x: int, y: int) -> Dict[str, Any]:
"""生成折线图"""
module_name = module.get('module_name', '访问趋势')
fields = module.get('moduld_fields', [])
# 生成模拟数据
months = ["1月", "2月", "3月", "4月", "5月", "6月", "7月", "8月", "9月", "10月", "11月", "12月"]
series_data = []
# 根据字段生成系列（最多2个）
for i, field in enumerate(fields[:2]):
data = [random.randint(500, 1500) for _ in range(12)]
series_data.append({
"name": field if isinstance(field, str) else f"系列{i+1}",
"data": data
})
# 如果没有字段，使用默认系列
if not series_data:
series_data = [
{"name": "访问量", "data": [820, 932, 901, 934, 1290, 1330, 1320, 1450, 1200, 1100, 1350, 1500]},
{"name": "下载量", "data": [620, 732, 701, 734, 1090, 1130, 1120, 1250, 1000, 900, 1150, 1300]}
]
line_id = str(uuid4())
return {
"id": line_id,
"i": line_id,
"type": "chart-line",
"x": x,
"y": y,
"w": 8,
"h": 5,
"minW": 4,
"minH": 3,
"title": "折线图",
"props": {
"title": module_name,
"smooth": True,
"showArea": False,
"showSymbol": True,
"symbolSize": 6,
"lineWidth": 2,
"showLegend": True,
"legendPosition": "top",
"colors": ["#5470c6", "#91cc75", "#fac858", "#ee6666", "#73c0de"],
"xAxisData": months,
"seriesData": series_data
}
}
def _generate_bar_chart(self, module: Dict[str, Any], x: int, y: int) -> Dict[str, Any]:
"""生成柱状图"""
module_name = module.get('module_name', '销售统计')
fields = module.get('moduld_fields', [])
# 生成模拟数据
regions = ["华东", "华南", "华北", "华中", "西南", "西北", "东北"]
series_data = []
# 根据字段生成系列（最多2个）
for i, field in enumerate(fields[:2]):
data = [random.randint(200, 500) for _ in range(7)]
series_data.append({
"name": field if isinstance(field, str) else f"{2023+i}",
"data": data
})
# 如果没有字段，使用默认系列
if not series_data:
series_data = [
{"name": "2023", "data": [320, 302, 301, 334, 390, 330, 320]},
{"name": "2024", "data": [420, 382, 391, 434, 490, 430, 420]}
]
bar_id = str(uuid4())
return {
"id": bar_id,
"i": bar_id,
"type": "chart-bar",
"x": x,
"y": y,
"w": 8,
"h": 6,
"minW": 4,
"minH": 3,
"title": "柱状图",
"props": {
"title": module_name,
"horizontal": False,
"stack": False,
"barWidth": "auto",
"barRadius": 4,
"showBackground": False,
"showLegend": True,
"legendPosition": "top",
"colors": ["#5470c6", "#91cc75", "#fac858", "#ee6666", "#73c0de"],
"xAxisData": regions,
"seriesData": series_data
}
}
def _generate_pie_chart(self, module: Dict[str, Any], x: int, y: int) -> Dict[str, Any]:
"""生成饼图"""
module_name = module.get('module_name', '流量来源')
fields = module.get('moduld_fields', [])
# 根据字段生成数据
series_data = []
if fields and len(fields) > 0:
# 使用字段名作为分类（最多5个）
for field in fields[:5]:
series_data.append({
"name": field if isinstance(field, str) else f"分类{len(series_data)+1}",
"value": random.randint(300, 1000)
})
# 如果没有字段，使用默认数据
if not series_data:
series_data = [
{"name": "搜索引擎", "value": 1048},
{"name": "直接访问", "value": 735},
{"name": "邮件营销", "value": 580},
{"name": "联盟广告", "value": 484},
{"name": "视频广告", "value": 300}
]
pie_id = str(uuid4())
return {
"id": pie_id,
"i": pie_id,
"type": "chart-pie",
"x": x,
"y": y,
"w": 4,
"h": 6,
"minW": 3,
"minH": 3,
"title": "饼图",
"props": {
"title": module_name,
"pieType": "rose",
"radius": ["0%", "70%"],
"showLabel": True,
"labelPosition": "outside",
"showLegend": True,
"legendPosition": "bottom",
"colors": ["#5470c6", "#91cc75", "#fac858", "#ee6666", "#73c0de"],
"seriesData": series_data
}
}