数据大屏两阶段分离架构 —— 设计时 vs 运行时解耦 · devcfg

🎯 本章核心问题

数据大屏（Dashboard）的核心矛盾是什么？

维度	传统方案的问题	我们的解决方案
性能	每次刷新都调 LLM，3-5 秒延迟	运行时只执行 SQL，200ms 内完成
成本	刷新 100 次 = 调用 100 次 LLM API	只在设计时调 1 次 LLM，后续零成本
可维护性	硬编码 UI 布局，改需求要改代码	配置驱动，修改 JSON 即可生效
灵活性	固定模板，无法自定义布局	Grid 网格系统 + 拖拽交互

---

📐 架构总览：两阶段分离模型

数据大屏两阶段分离架构

核心思想

设计时调 LLM 写配置，运行时仅执行 SQL

为什么这种设计是革命性的？

类比理解：

就像预制菜 vs 现炒：

• 设计时 = 厨师精心烹饪一道菜（调用 LLM 生成配置）
• 运行时 = 微波炉加热即可食用（直接读取配置 + 执行 SQL）

传统方案的致命缺陷：

# ❌ 传统方案：每次都调 LLM
async def get_dashboard_data(dashboard_id: int):
    dashboard = await db.get(dashboard_id)

    # 每个组件都要重新生成 SQL！
    for widget in dashboard.widgets:
        sql = await llm.generate_sql(widget.description)  # 💸 每次消耗 Token
        data = await db.execute(sql)                       # ⏳ 等待 LLM 响应

    return data

我们的方案：

# ✅ 我们的方案：运行时不调 LLM
async def get_dashboard_data(dashboard_id: int):
    # Step 1: 直接从数据库加载预生成的配置
    config = await load_dashboard_config(dashboard_id)  # ~10ms

    # Step 2: 并发执行所有 widget 的 SQL（无 LLM 调用）
    tasks = [
        execute_query(widget.sql_query)
        for widget in config.widgets
    ]
    results = await asyncio.gather(*tasks)  # 并发执行，~200ms

    return dict(zip([w.id for w in config.widgets], results))

---

🎨 一、阶段一：设计时 —— LLM 驱动的配置生成

1.1 用户输入 → 结构化配置

app/services/dashboard_service.py

async def create_from_natural_language(
    db: AsyncSession,
    datasource_id: int,
    description: str,
    name: Optional[str] = None,
) -> Dashboard:
    """
    根据自然语言描述创建 Dashboard

    Args:
        datasource_id: 数据源 ID
        description: 用户描述，如 "帮我创建一个销售看板..."
        name: 大屏名称（可选，LLM 可自动生成）
    """
    # Step 1: 获取语义模型（注入到 Prompt 中）
    semantic_model = await get_semantic_model(db, datasource_id)

    # Step 2: 构建 Prompt
    prompt = f"""你是一个专业的数据可视化专家。根据用户的需求描述和数据库结构，
生成一个完整的 Dashboard 配置对象。

## 数据库结构
{json.dumps(semantic_model, ensure_ascii=False, indent=2)}

## 用户需求
{description}

## 输出要求
请返回一个符合以下 JSON Schema 的配置对象（不要包含其他文字）：
{DASHBOARD_CONFIG_SCHEMA}"""

    # Step 3: 调用 LLM 生成配置
    gateway = LLMGateway()
    raw_response = await gateway.generate(prompt)

    # Step 4: 清洗并解析 JSON
    config_json = extract_json(raw_response)
    config = DashboardConfig(**config_json)

    # Step 5: 持久化到数据库
    dashboard = Dashboard(
        datasource_id=datasource_id,
        name=name or config.title,
        config_json=json.dumps(config, ensure_ascii=False),
    )
    db.add(dashboard)
    await db.flush()
    await db.refresh(dashboard)

    # Step 6: 创建 Widget 记录
    for widget_config in config.widgets:
        widget = DashboardWidget(
            dashboard_id=dashboard.id,
            type=widget_config.type,
            x=widget_config.x,
            y=widget_config.y,
            width=widget_config.width,
            height=widget_config.height,
            sql_query=widget_config.sql,  # ← 关键：SQL 已预编译好！
            chart_config=json.dumps(
                widget_config.chart_options or {},
                ensure_ascii=False
            ),
            title=widget_config.title or "",
            format_type=widget_config.format or "auto",
        )
        db.add(widget)

    return dashboard

1.2 DashboardConfig 数据结构定义

app/schemas/dashboard.py（schemas/dashboard.py）

from pydantic import BaseModel, Field
from typing import List, Optional, Any


class WidgetConfig(BaseModel):
    """单个 Widget 的配置"""
    type: str = Field(..., description="组件类型：kpi_card/line_chart/bar_chart/pie_chart/table")
    x: int = Field(0, description="Grid X 坐标（0-11）")
    y: int = Field(0, description="Grid Y 坐标")
    width: int = Field(6, description="Grid 宽度（1-12）")
    height: int = Field(2, description="Grid 高度")
    title: Optional[str] = Field(None, description="标题")
    sql: str = Field(..., description="预编译的 SQL 查询语句")
    format: Optional[str] = Field("auto", description="格式化方式：currency/percent/number/auto")
    chart_options: Optional[dict] = Field(None, description="ECharts 自定义选项")


class LayoutConfig(BaseModel):
    """网格布局配置"""
    cols: int = Field(12, description="列数")
    rows: int = Field(8, description="行数")
    gap: int = Field(16, description="间距（像素）")


class DashboardConfig(BaseModel):
    """完整的 Dashboard 配置"""
    id: Optional[str] = Field(None, description="唯一标识")
    title: str = Field("未命名看板", description="大屏标题")
    layout: LayoutConfig = Field(default_factory=LayoutConfig)
    widgets: List[WidgetConfig] = Field(..., description="组件列表")

    class Config:
        json_schema_extra = {
            "example": {
                "title": "销售数据分析看板",
                "layout": {"cols": 12, "rows": 8},
                "widgets": [
                    {
                        "type": "kpi_card",
                        "x": 0, "y": 0,
                        "width": 6, "height": 2,
                        "title": "今日销售额",
                        "sql": (
                            "SELECT SUM(total_amount) AS value "
                            "FROM orders "
                            "WHERE DATE(created_at) = CURDATE()"
                        ),
                        "format": "currency"
                    },
                    {
                        "type": "line_chart",
                        "x": 6, "y": 0,
                        "width": 6, "height": 2,
                        "title": "7日销售趋势",
                        "sql": (
                            "SELECT DATE(created_at) AS date, "
                            "SUM(total_amount) AS value "
                            "FROM orders "
                            "WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) "
                            "GROUP BY DATE(created_at) "
                            "ORDER BY date"
                        )
                    }
                ]
            }
        }

1.3 实际案例：从自然语言到完整配置

用户输入：

帮我创建一个销售数据看板，包含：
• 左上角：今日销售额 KPI 卡片
• 右上角：最近7天销售趋势折线图
• 下方：各品类销售额柱状图（TOP10）

LLM 生成的配置（存储在数据库中）：

每个 Widget 预存完整 SQL（已折行展示）

💡 关键观察

1. 每个 Widget 都有独立的、完整的 SQL — 这是运行时能脱离 LLM 的前提
2. 坐标使用 Grid 系统（x/y/w/h）— 支持拖拽调整位置和大小
3. chart_options 是可选的 — 允许高级用户微调图表样式

---

⚡ 二、阶段二：运行时 —— 高效的数据获取与渲染

2.1 加载 Dashboard 配置

app/api/dashboards.py（api/dashboards.py）

@router.get("/{dashboard_id}")
async def get_dashboard(dashboard_id: int, db: AsyncSession = Depends(get_db)):
    """
    加载 Dashboard 配置（不含数据）

    性能：~10ms（纯数据库读取）
    """
    result = await db.execute(
        select(Dashboard).where(Dashboard.id == dashboard_id)
    )
    dashboard = result.scalar_one_or_none()

    if not dashboard:
        raise HTTPException(status_code=404, detail="Dashboard not found")

    # 加载关联的 Widgets
    widgets_result = await db.execute(
        select(DashboardWidget)
        .where(DashboardWidget.dashboard_id == dashboard_id)
        .order_by(DashboardWidget.y, DashboardWidget.x)
    )
    widgets = list(widgets_result.scalars().all())

    return {
        "id": dashboard.id,
        "name": dashboard.name,
        "config": json.loads(dashboard.config_json),
        "widgets": [
            {
                "id": w.id,
                "type": w.type,
                "position": {"x": w.x, "y": w.y, "width": w.width, "height": w.height},
                "title": w.title,
                "format": w.format_type,
                "chartConfig": json.loads(w.chart_config) if w.chart_config else {},
            }
            for w in widgets
        ],
    }

2.2 并发查询所有 Widget 数据

app/api/dashboards.py（api/dashboards.py）

@router.post("/{dashboard_id}/data")
async def get_dashboard_data(
    dashboard_id: int,
    db: AsyncSession = Depends(get_db),
):
    """
    获取 Dashboard 所有 Widget 的实时数据

    性能优化点：
    1. 使用 asyncio.gather() 并发执行多个 SQL
    2. 不调用 LLM（SQL 已在设计时预编译）
    3. 结果缓存（可选，见第8篇）

    总耗时 ≈ max(单个SQL耗时)，而非 sum()
    """
    # 加载 Widgets
    widgets_result = await db.execute(
        select(DashboardWidget).where(DashboardWidget.dashboard_id == dashboard_id)
    )
    widgets = list(widgets_result.scalars().all())

    if not widgets:
        return {}

    async def fetch_widget_data(widget: DashboardWidget) -> tuple[int, list]:
        """查询单个 Widget 的数据"""
        try:
            data = await execute_query(widget.sql_query, datasource_id=None)
            return widget.id, data
        except Exception as e:
            logger.warning(f"Widget {widget.id} query failed: {e}")
            return widget.id, [{"error": str(e)}]

    # 🔥 核心：并发执行所有 Widget 的 SQL
    tasks = [fetch_widget_data(w) for w in widgets]
    results = await asyncio.gather(*tasks)

    # 组装为 {widget_id: data} 映射
    return {wid: data for wid, data in results}

🎯 为什么并发如此重要？

假设有 5 个 Widget，每个 SQL 执行需要 100ms：

方式	耗时计算	总时间
串行执行	100ms × 5 = 500ms	较慢
并发执行	max(100ms, 100ms, …) = ~100ms	快 5 倍

2.3 前端渲染引擎

views/dashboard/DashboardView.vue

<template>
  <div class="dashboard-container">
    <!-- 头部 -->
    <div class="dashboard-header">
      <h1>{{ dashboard.name }}</h1>
      <div class="actions">
        <button @click="refreshData">🔄 刷新</button>
        <span v-if="lastRefresh">最后更新: {{ lastRefresh }}</span>
      </div>
    </div>

    <!-- Grid 布局容器 -->
    <div
      class="grid-container"
      :style="{
        gridTemplateColumns: `repeat(${layout.cols}, 1fr)`,
        gridAutoRows: 'minmax(100px, auto)',
        gap: `${layout.gap}px`,
      }"
    >
      <!-- 动态渲染 Widget -->
      <div
        v-for="widget in widgets"
        :key="widget.id"
        class="widget-wrapper"
        :style="{
          gridColumnStart: widget.position.x + 1,
          gridRowStart: widget.position.y + 1,
          gridColumnEnd: `span ${widget.position.width}`,
          gridRowEnd: `span ${widget.position.height}`,
        }"
      >
        <!-- 组件映射表 -->
        <component
          :is="getComponentType(widget.type)"
          :title="widget.title"
          :data="widgetData[widget.id]"
          :chart-config="widget.chartConfig"
          :format="widget.format"
          :loading="loadingWidgets.has(widget.id)"
        />
      </div>
    </div>
  </div>
</template>

<script setup lang="ts">
import { ref, onMounted, computed } from 'vue'
import { useRoute } from 'vue-router'
import KpiCard from '@/components/dashboard/KpiCard.vue'
import LineChart from '@/components/dashboard/LineChart.vue'
import BarChart from '@/components/dashboard/BarChart.vue'
import PieChart from '@/components/dashboard/PieChart.vue'
import DataTable from '@/components/dashboard/DataTable.vue'

const route = useRoute()
const dashboardId = computed(() => Number(route.params.id))

// 组件类型映射表
const componentMap = {
  kpi_card: KpiCard,
  line_chart: LineChart,
  bar_chart: BarChart,
  pie_chart: PieChart,
  table: DataTable,
}

function getComponentType(type: string) {
  return componentMap[type] || DataTable
}

// 状态管理
const dashboard = ref<any>(null)
const widgets = ref<any[]>([])
const widgetData = ref<Record<number, any>>({})
const loadingWidgets = ref(new Set<number>())
const lastRefresh = ref<string>('')

// 加载配置
onMounted(async () => {
  const res = await fetch(`/api/dashboards/${dashboardId.value}`)
  const data = await res.json()
  dashboard.value = data
  widgets.value = data.widgets

  // 自动加载数据
  await refreshData()

  // 设置定时刷新（默认 60 秒）
  setInterval(refreshData, 60000)
})

// 刷新数据（只重新获取数据，不重新加载配置）
async function refreshData() {
  loadingWidgets.value = new Set(widgets.value.map((w: any) => w.id))

  try {
    const res = await fetch(`/api/dashboards/${dashboardId.value}/data`, {
      method: 'POST',
    })
    const data = await res.json()
    widgetData.value = data
    lastRefresh.value = new Date().toLocaleTimeString()
  } finally {
    loadingWidgets.value.clear()
  }
}
</script>

<style scoped>
.grid-container {
  display: grid;
  padding: 16px;
}

.widget-wrapper {
  background: white;
  border-radius: 8px;
  box-shadow: 0 2px 8px rgba(0, 0, 0, 0.08);
  overflow: hidden;
}
</style>

2.4 自动刷新机制

// 定时刷新策略
interface RefreshConfig {
  enabled: boolean       // 是否启用
  interval: number       // 间隔（毫秒），默认 60000（60秒）
  pauseOnHidden: boolean // 页面不可见时暂停（节省资源）
}

// 实现
let timer: number | null = null

function startAutoRefresh(config: RefreshConfig) {
  if (!config.enabled) return

  const tick = () => {
    // 页面不可见时跳过
    if (config.pauseOnHidden && document.hidden) return
    refreshData()
  }

  timer = window.setInterval(tick, config.interval)

  // 监听页面可见性变化
  document.addEventListener('visibilitychange', () => {
    if (document.hidden && timer) {
      clearInterval(timer)
      timer = null
    } else if (!document.hidden && !timer) {
      timer = window.setInterval(tick, config.interval)
    }
  })
}

---

🗄️ 三、数据模型设计

3.1 ER 图

1:N 关系：一个大屏包含多个 Widget

3.2 SQLAlchemy 模型

app/models/dashboard.py（models/dashboard.py）

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime


class Dashboard(Base):
    __tablename__ = "dashboards"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(200), nullable=False, default="未命名看板")
    datasource_id = Column(Integer, ForeignKey("datasources.id"), nullable=False)
    config_json = Column(Text, nullable=True)  # 完整的 DashboardConfig JSON
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # 关联关系
    widgets = relationship("DashboardWidget", back_populates="dashboard", cascade="all, delete-orphan")


class DashboardWidget(Base):
    __tablename__ = "dashboard_widgets"

    id = Column(Integer, primary_key=True, autoincrement=True)
    dashboard_id = Column(Integer, ForeignKey("dashboards.id"), nullable=False)

    # 组件类型
    type = Column(String(50), nullable=False)  # kpi_card / line_chart / bar_chart / pie_chart / table

    # Grid 坐标系
    x = Column(Integer, default=0)
    y = Column(Integer, default=0)
    width = Column(Integer, default=6)
    height = Column(Integer, default=2)

    # 内容配置
    sql_query = Column(Text, nullable=False)  # ← 核心：预编译的 SQL
    chart_config = Column(Text, nullable=True)  # ECharts 选项 JSON
    title = Column(String(200), default="")
    format_type = Column(String(20), default="auto")  # currency / percent / number

    # 关联
    dashboard = relationship("Dashboard", back_populates="widgets")

---

🔄 四、两阶段对比总结

4.1 完整生命周期

设计时一次性生成 vs 运行时无限刷新

4.2 性能与成本对比矩阵

指标	传统方案（每次调 LLM）	我们的方案（两阶段分离）
首次加载	3000-5000ms	3000-5000ms（含 LLM）
刷新延迟	3000-5000ms	150-250ms（纯 SQL）
单日成本（100 次刷新）	~$0.50	$0.005（仅首次）
月度成本（30天）	~$15	$0.15
可扩展性	受限于 LLM QPS	仅受限于 MySQL 连接池
离线可用	❌ 需要 LLM 服务在线	✅ 配置已本地持久化

4.3 适用场景判断

场景	是否适合两阶段架构？	原因
✅ 实时监控大屏	强烈推荐	高频刷新，成本敏感
✅ 固定报表模板	推荐	SQL 稳定，无需频繁调整
⚠️ 探索式分析	可选	可能需要频繁修改配置
❌ 完全动态场景	不推荐	每次 SQL 都不同，无法预编译

---

🎯 五、最佳实践与设计模式

✅ 我们做到了什么

1. 真正的解耦：UI 渲染与业务逻辑完全分离
2. 声明式编程：通过 JSON 描述界面，而非命令式代码
3. 性能极致优化：运行时零 AI 成本，响应速度提升 15 倍+
4. 热更新能力：修改数据库配置即可实时生效
5. 前端组件化：基于 Vue 动态组件实现灵活的 Widget 系统

📚 最佳实践清单

• 使用 Pydantic 定义严格的 Config Schema（确保 LLM 输出格式正确）
• 每个 Widget 的 SQL 必须独立、完整（不能依赖其他 Widget 的结果）
• 使用 asyncio.gather() 并发查询（而非 for 循环串行）
• 前端使用 CSS Grid 或绝对定位实现自由布局
• 定时刷新时暂停页面隐藏状态（节省资源）
• 错误隔离：单个 Widget 查询失败不影响整体展示
• 配置版本控制：记录每次修改的时间戳和操作人

🚀 进阶优化方向

1. 增量更新：只刷新变化的 Widget（通过 SQL Hash 对比）
2. WebSocket 推送：替代轮询，实现真正的实时更新
3. SQL 缓存层：对于聚合类查询，缓存 TTL 内的结果
4. 多数据源支持：同一个大屏可以组合不同数据库的数据

---

相关代码文件：

• app/services/dashboard_service.py（dashboard_service.py） — 大屏服务核心逻辑
• app/models/dashboard.py（models/dashboard.py） — 数据模型定义
• app/schemas/dashboard.py（schemas/dashboard.py） — Pydantic 配置 Schema
• app/api/dashboards.py（api/dashboards.py） — RESTful API 接口
• DashboardView.vue — 前端大屏视图
• components/dashboard/*.vue — Widget 组件库

下一篇文章将深入 前端拖拽交互系统的实现细节 —— HTML5 Drag API、Grid 布局、mousedown 移动/缩放（碰撞检测为规划能力）！

敬请期待！🚀