JimuChatBI v1.2.0：用语义模型把自然语言问数接入业务分析

自然语言问题
    ↓
识别指标、维度、筛选条件和时间范围
    ↓
匹配语义模型中的业务定义
    ↓
生成受约束的 SQL
    ↓
执行权限检查和查询
    ↓
返回表格、图表与 SQL

import sqlite3
from dataclasses import dataclass


@dataclass(frozen=True)
class Metric:
    name: str
    expression: str
    required_status: str


# 语义层：业务人员说“销售额”时，固定使用实付金额，且只统计已支付订单。
METRICS = {
    "销售额": Metric(
        name="销售额",
        expression="SUM(paid_amount)",
        required_status="paid",
    ),
    "订单量": Metric(
        name="订单量",
        expression="COUNT(*)",
        required_status="paid",
    ),
}


def init_database() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    conn.executescript(
        """
        CREATE TABLE orders (
            id INTEGER PRIMARY KEY,
            region TEXT NOT NULL,
            paid_amount REAL NOT NULL,
            status TEXT NOT NULL,
            paid_at TEXT NOT NULL
        );

        INSERT INTO orders(region, paid_amount, status, paid_at) VALUES
            ('华东', 1200.00, 'paid',      '2025-01-05'),
            ('华东',  800.00, 'paid',      '2025-01-18'),
            ('华南', 1500.00, 'paid',      '2025-01-20'),
            ('华南',  500.00, 'cancelled', '2025-01-22');
        """
    )
    return conn


def build_query(question: str) -> tuple[str, dict]:
    metric = next(
        (definition for keyword, definition in METRICS.items() if keyword in question),
        None,
    )
    if metric is None:
        raise ValueError("只支持查询：销售额、订单量")

    if "地区" not in question:
        raise ValueError("当前示例只支持按地区汇总")

    # 表名、字段名和聚合表达式只能来自语义层，不接受用户直接输入。
    sql = f"""
        SELECT
            region AS 地区,
            {metric.expression} AS {metric.name}
        FROM orders
        WHERE status = :status
          AND paid_at >= :start_date
          AND paid_at < :end_date
        GROUP BY region
        ORDER BY {metric.name} DESC
    """

    params = {
        "status": metric.required_status,
        "start_date": "2025-01-01",
        "end_date": "2025-02-01",
    }
    return sql.strip(), params


def ask(question: str) -> None:
    conn = init_database()
    sql, params = build_query(question)
    rows = conn.execute(sql, params).fetchall()

    print(f"问题：{question}")
    print("\n结果：")
    for row in rows:
        print(dict(row))

    print("\n生成的 SQL：")
    print(sql)
    print("\n参数：", params)


if __name__ == "__main__":
    ask("查询 2025 年 1 月各地区销售额")

python chat2bi_demo.py

{
  "metric": "销售额",
  "dimensions": ["地区"],
  "time_range": {
    "start": "2025-01-01",
    "end": "2025-02-01"
  }
}