diff --git a/templates/ti-cl4/llm-scenarios/config/nl_query.template.yaml b/templates/ti-cl4/llm-scenarios/config/nl_query.template.yaml new file mode 100644 index 0000000..8b743c1 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/config/nl_query.template.yaml @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +# 模板「自然语言查询（NL→SQL/API）」配置资产：ti-cl4（Template-Ti 一期）。 +# +# 说明（issue #76 / 父 Issue #11）： +# - metrics：自然语言指标名 → 点位（point_id），NL 查询翻译的依据； +# - intents：意图关键词（trend/latest/kpi/alarm），长词优先； +# - table：TDengine 超级表（对齐 data-bus tdengine_schema 命名）； +# - default_time_range：未识别时间范围时的默认窗口； +# - 换行业/换点位只改本文件，翻译器内核零改动。 +template: ti-cl4 +version: 1.0.0 + +nl_query: + table: tpl_ti_cl4.points + default_time_range: 1h + + # 自然语言指标名 → point_id（演示字典，接客户数据时按点表补充） + metrics: + 氯气流量: CLF-01.FLOW + 炉温: CLF-01.TEMP + 炉压: CLF-01.PRES + 进料量: CLF-01.FEED + 钛纯度: LAB-01.TI_PURITY + 交联度: LAB-01.CROSSLINK + + # 意图关键词（长词优先；未命中 → unsupported，转 LLM 问答） + intents: + trend: [趋势, 走势, 曲线, 变化] + alarm: [报警, 告警, 异常] + kpi: [平均, 统计, 均值, 最大值, 最小值] + latest: [最新, 现在, 当前, 多少, 数值] diff --git a/templates/ti-cl4/llm-scenarios/nl_query.py b/templates/ti-cl4/llm-scenarios/nl_query.py new file mode 100644 index 0000000..4b7de95 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/nl_query.py @@ -0,0 +1,161 @@ +# -*- coding: utf-8 -*- +"""Template-Ti 一期 · 自然语言查询接口（NL→SQL/API）—— issue #76。 + +父 Issue #11「④ LLM 报警解释 / 交接班 / NL 查询」子任务： +把驾驶舱/对话中的自然语言问题翻译为**结构化查询**： + +- 意图识别（intent）：trend（趋势）/ latest（最新值）/ kpi（统计指标）/ alarm（告警）； +- 指标映射（metric）：自然语言指标名 → 点位（point_id），配置驱动 + （`config/nl_query.template.yaml` 指标字典）； +- 时间范围（time_range）：从问句抽取（"最近 1 小时" → 1h）； +- 产出：TDengine SQL（超级表查询）+ 驾驶舱 API 调用参数（to_api_params）。 + +纯本地规则实现（无 LLM 依赖、可离线测试）；未识别意图/指标时给出 +结构化降级（intent=unsupported），由上层转 LLM 问答（query_cockpit）。 +""" +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from typing import Dict, List, Optional + +#: 默认配置资产路径（相对本模块） +DEFAULT_CONFIG_PATH = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "config", + "nl_query.template.yaml") + +#: 默认 TDengine 超级表（对齐 data-bus tdengine_schema 命名） +DEFAULT_TABLE = "tpl_ti_cl4.points" + +#: 时间范围抽取正则：最近 N 小时/分钟/天 +_TIME_RANGE_RE = re.compile(r"最近\s*(\d+)\s*(小时|分钟|天|h|min|d)") +_TIME_UNIT = {"小时": "h", "分钟": "m", "天": "d", "h": "h", "min": "m", "d": "d"} + + +@dataclass +class NLQuery: + """一次 NL 查询的结构化结果。""" + + question: str + intent: str # trend | latest | kpi | alarm | unsupported + metric: str = "" + point_id: str = "" + device: str = "" + time_range: str = "" # 如 "1h"；空 = 默认窗口 + sql: str = "" # TDengine SQL（intent=unsupported 时为空） + meta: dict = field(default_factory=dict) + + def to_api_params(self) -> dict: + """驾驶舱 API 调用参数（供前端查询接口使用）。""" + return { + "intent": self.intent, "metric": self.metric, + "point_id": self.point_id, "device": self.device, + "time_range": self.time_range or "1h", + } + + +class NLQueryTranslator: + """自然语言 → 结构化查询（NL→SQL/API，规则 + 配置驱动）。""" + + #: 意图关键词（长词优先） + _INTENT_KEYWORDS = [ + ("trend", ["趋势", "走势", "曲线", "变化"]), + ("alarm", ["报警", "告警", "异常"]), + ("kpi", ["平均", "统计", "均值", "最大值", "最小值"]), + ("latest", ["最新", "现在", "当前", "多少", "数值"]), + ] + + def __init__( + self, + metrics: Optional[Dict[str, str]] = None, + table: str = DEFAULT_TABLE, + default_range: str = "1h", + intent_keywords: Optional[Dict[str, List[str]]] = None, + ) -> None: + """Args: + metrics: 自然语言指标名 → point_id（如 {"氯气流量": "CLF-01.FLOW"}）； + table: TDengine 超级表名； + default_range: 未识别时间范围时的默认窗口； + intent_keywords: 意图关键词覆盖。 + """ + self.metrics: Dict[str, str] = dict(metrics or {}) + self.table = table + self.default_range = default_range + self._intent = intent_keywords or dict(self._INTENT_KEYWORDS) + + # ------------------------------------------------------------------ + @classmethod + def from_template_config(cls, path: str = DEFAULT_CONFIG_PATH) -> "NLQueryTranslator": + """从模板配置资产加载（config/nl_query.template.yaml）。""" + import yaml + + with open(path, "r", encoding="utf-8") as fh: + raw = yaml.safe_load(fh) or {} + cfg = raw.get("nl_query", {}) or {} + return cls( + metrics=cfg.get("metrics", {}), + table=cfg.get("table", DEFAULT_TABLE), + default_range=cfg.get("default_time_range", "1h"), + intent_keywords=cfg.get("intents"), + ) + + # ------------------------------------------------------------------ + def translate(self, question: str) -> NLQuery: + """把自然语言问题翻译为结构化查询。""" + intent = self._detect_intent(question) + if intent == "unsupported": + return NLQuery(question=question, intent="unsupported", + meta={"reason": "未识别查询意图"}) + metric = self._detect_metric(question) + time_range = self._detect_time_range(question) + point_id = self.metrics.get(metric, "") if metric else "" + query = NLQuery( + question=question, intent=intent, metric=metric, + point_id=point_id, time_range=time_range, + ) + query.sql = self._build_sql(query) + query.meta = {"table": self.table} + return query + + # ------------------------------------------------------------------ + def _detect_intent(self, question: str) -> str: + for intent, keywords in self._intent.items(): + for kw in keywords: + if kw in question: + return intent + return "unsupported" + + def _detect_metric(self, question: str) -> str: + """指标识别：配置字典中自然语言名作为子串匹配（长名优先）。""" + candidates = sorted(self.metrics, key=len, reverse=True) + for name in candidates: + if name in question: + return name + return "" + + @staticmethod + def _detect_time_range(question: str) -> str: + m = _TIME_RANGE_RE.search(question) + if not m: + return "" + return f"{int(m.group(1))}{_TIME_UNIT[m.group(2)]}" + + def _build_sql(self, query: NLQuery) -> str: + """生成 TDengine SQL（超级表，按 point_id 过滤）。""" + point_filter = f"point_id = '{query.point_id}'" if query.point_id else "1=1" + window = query.time_range or self.default_range + if query.intent == "latest": + return (f"SELECT last_row(value) AS value FROM {self.table} " + f"WHERE {point_filter} AND ts >= now - {window}") + if query.intent == "kpi": + return (f"SELECT avg(value) AS value_avg FROM {self.table} " + f"WHERE {point_filter} AND ts >= now - {window}") + if query.intent == "alarm": + return (f"SELECT count(*) AS alarms FROM {self.table} " + f"WHERE {point_filter} AND value > threshold " + f"AND ts >= now - {window}") + # trend + return (f"SELECT _wstart AS ts, avg(value) AS value_avg " + f"FROM {self.table} WHERE {point_filter} " + f"AND ts >= now - {window} INTERVAL(1m)") diff --git a/templates/ti-cl4/llm-scenarios/tests/test_nl_query.py b/templates/ti-cl4/llm-scenarios/tests/test_nl_query.py new file mode 100644 index 0000000..19b2200 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/tests/test_nl_query.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +"""NL→SQL/API 查询翻译器测试（issue #76）。 + +覆盖： +1. 配置资产加载（指标字典/意图关键词/表/默认窗口）； +2. 意图识别（trend/latest/kpi/alarm + 未识别降级 unsupported）； +3. 指标映射（自然语言名 → point_id，长名优先）； +4. 时间范围抽取（"最近 1 小时" → 1h）； +5. TDengine SQL 生成（latest/kpi/alarm/trend 模板）； +6. to_api_params（驾驶舱 API 调用参数）。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from ti_scenarios.nl_query import NLQueryTranslator # noqa: E402 + +CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "nl_query.template.yaml", +) + + +class TestConfigLoad(unittest.TestCase): + """配置资产加载。""" + + def setUp(self): + self.t = NLQueryTranslator.from_template_config(CONFIG) + + def test_metrics_and_table(self): + self.assertEqual(self.t.metrics["氯气流量"], "CLF-01.FLOW") + self.assertEqual(self.t.table, "tpl_ti_cl4.points") + self.assertEqual(self.t.default_range, "1h") + + def test_intent_keywords_loaded(self): + self.assertIn("趋势", self.t._intent["trend"]) + + +class TestTranslate(unittest.TestCase): + """翻译：意图/指标/时间范围 → SQL/API 参数。""" + + def setUp(self): + self.t = NLQueryTranslator.from_template_config(CONFIG) + + def test_trend_query(self): + q = self.t.translate("氯气流量最近1小时趋势") + self.assertEqual(q.intent, "trend") + self.assertEqual(q.metric, "氯气流量") + self.assertEqual(q.point_id, "CLF-01.FLOW") + self.assertEqual(q.time_range, "1h") + self.assertIn("INTERVAL(1m)", q.sql) + self.assertIn("avg(value)", q.sql) + self.assertIn("CLF-01.FLOW", q.sql) + + def test_latest_query(self): + q = self.t.translate("炉温最新数值是多少") + self.assertEqual(q.intent, "latest") + self.assertIn("last_row", q.sql) + self.assertEqual(q.point_id, "CLF-01.TEMP") + + def test_kpi_query_with_default_range(self): + q = self.t.translate("氯气流量平均") + self.assertEqual(q.intent, "kpi") + self.assertEqual(q.time_range, "") # 未识别 → 默认窗口 + self.assertIn("now - 1h", q.sql) # 默认 1h + + def test_alarm_query(self): + q = self.t.translate("炉温报警") + self.assertEqual(q.intent, "alarm") + self.assertIn("value > threshold", q.sql) + + def test_unsupported_intent(self): + q = self.t.translate("今天天气怎么样") + self.assertEqual(q.intent, "unsupported") + self.assertEqual(q.sql, "") + + def test_unknown_metric_keeps_intent(self): + # 未识别指标不阻断查询（SQL 用全表过滤，交由上层 LLM 兜底） + q = self.t.translate("进料泵转速趋势") + self.assertEqual(q.intent, "trend") + self.assertEqual(q.point_id, "") + + def test_api_params(self): + q = self.t.translate("氯气流量最近1小时趋势") + params = q.to_api_params() + self.assertEqual(params["intent"], "trend") + self.assertEqual(params["metric"], "氯气流量") + self.assertEqual(params["point_id"], "CLF-01.FLOW") + self.assertEqual(params["time_range"], "1h") + + +class TestTimeRange(unittest.TestCase): + """时间范围抽取。""" + + def test_hour_minute_day(self): + t = NLQueryTranslator(metrics={}) + self.assertEqual(t.translate("最近 2 小时趋势").time_range, "2h") + self.assertEqual(t.translate("最近30分钟走势").time_range, "30m") + self.assertEqual(t.translate("最近 3 天曲线").time_range, "3d") + + +if __name__ == "__main__": + unittest.main()