bot_ym/iAOP
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(#64): 配置项 CRUD 存储引擎（模型超参/RAG/布局三类，文件系统版本化 JSON+校验）

Browse Source
This commit is contained in:
bot_dev1
2026-08-05 05:29:53 +08:00
parent 94c2782e75
commit b0ae378150
2 changed files with 482 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/template-console/config_store.py
+291
View File
@@ -0,0 +1,291 @@
# -*- coding: utf-8 -*-
"""⑤.7 配置项 CRUD 存储引擎 —— issue #64 / PRD ⑤.7。
配置台要管理三类业务配置：**模型超参 / RAG / 布局**。这些配置是模板交付物的
"活"部分——实施工程师按现场调参，每次改动都要**可解释、可校验、可版本化**
（为 #66 发布/回滚提供快照源）。本模块提供基于文件系统的版本化 JSON 存储：
- 三类配置各对应一个 JSON 文件（``model_params.json`` / ``rag_configs.json`` /
``layout.json``），存放在一个 store 根目录下；
- 每条配置项是一个 ``ConfigItem``（key + value + 含义 + 校验规则）；
- 提供 ``list / get / upsert / delete`` CRUD，所有写操作都先**校验**再落盘，
并记录 ``updated_by`` / ``reason``（对齐 PRD「可解释可溯源」）；
- 校验规则按类别内置（模型超参的范围/类型、RAG 的来源数、布局的 widget 类型），
非法值在 upsert 阶段即被拒绝，避免坏数据进入版本快照。
存储格式（每类一个 JSON，内容为 ``{items: [ConfigItem, ...], schema_version}``）
刻意简单、人可读，便于实施工程师直接查看/备份。
零运行时依赖：仅用 json / dataclass / Enum / 标准库。
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
# ---------------------------------------------------------------------------
# 配置类别
# ---------------------------------------------------------------------------
class ConfigKind(str, Enum):
"""三类业务配置（对齐 #64 需求）。"""
MODEL_PARAM = "model_param" # 模型超参（学习率/迭代数/特征开关…）
RAG_CONFIG = "rag_config" # RAG 知识库配置（top_k/相似度阈值/来源…）
LAYOUT = "layout" # 驾驶舱布局（widget 列表）
#: 各类别对应的存储文件名
KIND_FILENAME: Dict[ConfigKind, str] = {
ConfigKind.MODEL_PARAM: "model_params.json",
ConfigKind.RAG_CONFIG: "rag_configs.json",
ConfigKind.LAYOUT: "layout.json",
}
#: 存储结构版本（schema 演进时升级，发布快照会带上）
STORE_SCHEMA_VERSION = 1
#: 驾驶舱布局允许的 widget 类型（对齐 iAOP-cockpit-layout-v1 / resin cockpit）
ALLOWED_WIDGET_TYPES = {"process_view", "trend", "kpi_card", "alarm_panel", "nl_query"}
# ---------------------------------------------------------------------------
# 配置项数据模型
# ---------------------------------------------------------------------------
@dataclass
class ConfigItem:
"""一条配置项（可解释：带含义、更新人、原因）。"""
key: str # 配置键（类别内唯一，如 learning_rate）
value: Any # 配置值（标量或结构化）
kind: ConfigKind # 所属类别
meaning: str = "" # 业务含义（供配置台展示与审计）
updated_by: str = "system" # 最后修改人（对接 RBAC 用户名）
reason: str = "" # 本次修改原因（可解释可溯源）
updated_at: str = "" # ISO8601 时间戳
def to_dict(self) -> dict:
d = asdict(self)
d["kind"] = self.kind.value # 枚举序列化为字符串
return d
@classmethod
def from_dict(cls, raw: dict) -> "ConfigItem":
return cls(
key=raw["key"],
value=raw.get("value"),
kind=ConfigKind(raw.get("kind")),
meaning=raw.get("meaning", ""),
updated_by=raw.get("updated_by", "system"),
reason=raw.get("reason", ""),
updated_at=raw.get("updated_at", ""),
)
def _now_iso() -> str:
"""当前 UTC 时间 ISO8601（无时区歧义）。"""
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# ---------------------------------------------------------------------------
# 校验（按类别内置规则）
# ---------------------------------------------------------------------------
@dataclass
class ValidationResult:
"""配置项校验结果。"""
ok: bool
errors: List[str] = field(default_factory=list)
def __bool__(self) -> bool:
return self.ok
def validate_item(kind: ConfigKind, key: str, value: Any) -> ValidationResult:
"""按类别校验配置项的 key/value 合法性。
校验规则（配置台 upsert 前置门禁，防止坏数据进快照）：
- 通用：key 非空、匹配 ``[a-z0-9_.-]+``；
- model_param：value 为标量（int/float/bool/str）或标量列表；
- rag_config：top_k 为 1~50 的正整数、similarity_threshold 为 0~1 浮点、
sources 为非空字符串列表；
- layout：value 为 widget 列表，每个 widget 有合法 type 与 x/y/w/h。
"""
errors: List[str] = []
if not key or not isinstance(key, str):
errors.append("key 不能为空")
elif not re.match(r"^[a-z0-9_.\-]+$", key):
errors.append(f"key '{key}' 仅允许小写字母/数字/._-")
if kind == ConfigKind.MODEL_PARAM:
if not isinstance(value, (int, float, bool, str, list)):
errors.append("model_param 的 value 必须为标量或标量列表")
elif isinstance(value, list) and any(
not isinstance(v, (int, float, bool, str)) for v in value):
errors.append("model_param 列表 value 仅允许标量元素")
# 常见超参范围提示（软约束，仅对已知键）
if key == "learning_rate" and isinstance(value, (int, float)):
if not (0 < value < 1):
errors.append("learning_rate 应在 (0, 1) 区间")
if key == "iterations" and isinstance(value, int):
if value <= 0:
errors.append("iterations 必须为正整数")
elif kind == ConfigKind.RAG_CONFIG:
if key == "top_k":
if not (isinstance(value, int) and 1 <= value <= 50):
errors.append("top_k 必须为 1~50 的整数")
elif key == "similarity_threshold":
if not (isinstance(value, (int, float)) and 0 <= value <= 1):
errors.append("similarity_threshold 必须为 0~1 的数")
elif key == "sources":
if not (isinstance(value, list) and value
and all(isinstance(s, str) and s for s in value)):
errors.append("sources 必须为非空字符串列表")
elif kind == ConfigKind.LAYOUT:
if not isinstance(value, list):
errors.append("layout 的 value 必须为 widget 列表")
else:
for i, w in enumerate(value):
if not isinstance(w, dict):
errors.append(f"widget[{i}] 必须为对象")
continue
wt = w.get("type")
if wt not in ALLOWED_WIDGET_TYPES:
errors.append(
f"widget[{i}] 非法 type '{wt}'（合法：{sorted(ALLOWED_WIDGET_TYPES)})")
for coord in ("x", "y", "w", "h"):
if not isinstance(w.get(coord), int) or w.get(coord) < 0:
errors.append(f"widget[{i}] {coord} 必须为非负整数")
return ValidationResult(ok=not errors, errors=errors)
# ---------------------------------------------------------------------------
# 存储引擎
# ---------------------------------------------------------------------------
class ConfigStore:
"""基于文件系统的版本化配置存储（三类配置各一 JSON）。
用法：
store = ConfigStore("/path/to/store")
store.upsert(ConfigKind.MODEL_PARAM, "learning_rate", 0.001,
meaning="学习率", updated_by="li", reason="首次标定")
items = store.list(ConfigKind.MODEL_PARAM)
"""
def __init__(self, root: str) -> None:
self.root = root
os.makedirs(root, exist_ok=True)
# -- 路径 --
def _path(self, kind: ConfigKind) -> str:
return os.path.join(self.root, KIND_FILENAME[kind])
def _read(self, kind: ConfigKind) -> List[ConfigItem]:
path = self._path(kind)
if not os.path.isfile(path):
return []
with open(path, "r", encoding="utf-8") as fh:
blob = json.load(fh)
return [ConfigItem.from_dict(r) for r in blob.get("items", [])]
def _write(self, kind: ConfigKind, items: List[ConfigItem]) -> None:
blob = {
"schema_version": STORE_SCHEMA_VERSION,
"kind": kind.value,
"items": [it.to_dict() for it in items],
}
path = self._path(kind)
# 先写临时文件再替换，避免写一半被读到（原子写）
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(blob, fh, ensure_ascii=False, indent=2)
os.replace(tmp, path)
# -- 查询 --
def list(self, kind: ConfigKind) -> List[ConfigItem]:
"""列出某类全部配置项。"""
return self._read(kind)
def get(self, kind: ConfigKind, key: str) -> Optional[ConfigItem]:
"""取单条配置项（不存在返回 None）。"""
for it in self._read(kind):
if it.key == key:
return it
return None
# -- 写 --
def upsert(
self,
kind: ConfigKind,
key: str,
value: Any,
meaning: str = "",
updated_by: str = "system",
reason: str = "",
) -> ConfigItem:
"""新增或更新一条配置项（先校验，再落盘）。
Raises:
ValueError: 校验失败（带全部错误明细）。
"""
vr = validate_item(kind, key, value)
if not vr:
raise ValueError(f"配置项校验失败 [{kind.value}:{key}]：{'; '.join(vr.errors)}")
items = self._read(kind)
now = _now_iso()
existing_idx = next((i for i, it in enumerate(items) if it.key == key), None)
item = ConfigItem(
key=key, value=value, kind=kind, meaning=meaning,
updated_by=updated_by, reason=reason, updated_at=now,
)
if existing_idx is None:
items.append(item)
else:
items[existing_idx] = item
self._write(kind, items)
return item
def delete(self, kind: ConfigKind, key: str) -> bool:
"""删除一条配置项。返回是否实际删除。"""
items = self._read(kind)
new_items = [it for it in items if it.key != key]
if len(new_items) == len(items):
return False
self._write(kind, new_items)
return True
# -- 快照（供 #66 release 使用） --
def snapshot(self) -> Dict[str, Any]:
"""全量配置快照（三类聚合，供发布版本固化）。"""
return {
"schema_version": STORE_SCHEMA_VERSION,
"captured_at": _now_iso(),
"kinds": {
kind.value: [it.to_dict() for it in self._read(kind)]
for kind in ConfigKind
},
}
def restore(self, snapshot: Dict[str, Any]) -> None:
"""从快照恢复全部配置（#66 回滚入口）。"""
kinds = snapshot.get("kinds", {})
for kind in ConfigKind:
raw_items = kinds.get(kind.value, [])
items = [ConfigItem.from_dict(r) for r in raw_items]
self._write(kind, items)
def item_counts(self) -> Dict[str, int]:
"""各类配置项数量（配置台仪表盘用）。"""
return {kind.value: len(self._read(kind)) for kind in ConfigKind}
core/template-console/tests/test_config_store.py
+191
View File
@@ -0,0 +1,191 @@
# -*- coding: utf-8 -*-
"""配置项 CRUD 存储引擎测试（issue #64）。
覆盖：
1. 三类配置 CRUD（list/get/upsert/delete）；
2. 原子写 + 持久化（重开 store 仍在）；
3. 校验规则（model_param/rag_config/layout，非法值拒绝）；
4. 快照 snapshot/restore（为 #66 提供基础）；
5. 可解释字段（meaning/reason/updated_by/updated_at 落盘）。
"""
import json
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.config_store import ( # noqa: E402
ALLOWED_WIDGET_TYPES,
ConfigItem,
ConfigKind,
ConfigStore,
ValidationResult,
validate_item,
)
class _TmpStore:
def __init__(self):
self._tmp = tempfile.mkdtemp()
self.store = ConfigStore(self._tmp)
def cleanup(self):
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
class ValidationTest(unittest.TestCase):
"""校验规则。"""
def test_model_param_scalar_ok(self):
self.assertTrue(validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 0.001))
def test_model_param_learning_rate_range(self):
vr = validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 1.5)
self.assertFalse(vr)
self.assertTrue(any("learning_rate" in e for e in vr.errors))
def test_model_param_bad_key(self):
vr = validate_item(ConfigKind.MODEL_PARAM, "Bad Key!", 1)
self.assertFalse(vr)
def test_rag_top_k_bounds(self):
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 0))
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 51))
self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "top_k", 10))
def test_rag_similarity_threshold(self):
self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.5))
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 1.5))
def test_rag_sources_must_be_nonempty_list(self):
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", []))
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", ["", "x"]))
self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "sources", ["sop", "gb"]))
def test_layout_widget_type(self):
bad = [{"type": "unknown", "x": 0, "y": 0, "w": 1, "h": 1}]
self.assertFalse(validate_item(ConfigKind.LAYOUT, "dashboard", bad))
good = [{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}]
self.assertTrue(validate_item(ConfigKind.LAYOUT, "dashboard", good))
def test_layout_widget_coords_nonneg_int(self):
bad = [{"type": "trend", "x": -1, "y": 0, "w": 1, "h": 1}]
vr = validate_item(ConfigKind.LAYOUT, "dashboard", bad)
self.assertFalse(vr)
class CrudTest(unittest.TestCase):
"""CRUD + 持久化。"""
def setUp(self):
self.ctx = _TmpStore()
self.store = self.ctx.store
def tearDown(self):
self.ctx.cleanup()
def test_upsert_and_get(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "iterations", 100,
meaning="迭代数", updated_by="li", reason="标定")
it = self.store.get(ConfigKind.MODEL_PARAM, "iterations")
self.assertIsNotNone(it)
self.assertEqual(it.value, 100)
self.assertEqual(it.updated_by, "li")
self.assertEqual(it.reason, "标定")
self.assertTrue(it.updated_at) # 时间戳已写
def test_upsert_rejects_invalid(self):
with self.assertRaises(ValueError):
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 999)
def test_upsert_overwrites(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.1)
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01, reason="调小")
it = self.store.get(ConfigKind.MODEL_PARAM, "lr")
self.assertEqual(it.value, 0.01)
self.assertEqual(it.reason, "调小")
def test_list_and_delete(self):
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 5)
self.store.upsert(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.6)
self.assertEqual(len(self.store.list(ConfigKind.RAG_CONFIG)), 2)
self.assertTrue(self.store.delete(ConfigKind.RAG_CONFIG, "top_k"))
self.assertIsNone(self.store.get(ConfigKind.RAG_CONFIG, "top_k"))
self.assertFalse(self.store.delete(ConfigKind.RAG_CONFIG, "nope"))
def test_persistence_across_reopen(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
# 重开一个指向同一目录的 store
store2 = ConfigStore(self.ctx._tmp)
it = store2.get(ConfigKind.MODEL_PARAM, "lr")
self.assertIsNotNone(it)
self.assertEqual(it.value, 0.001)
def test_json_file_is_human_readable(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001, meaning="学习率")
path = os.path.join(self.ctx._tmp, "model_params.json")
with open(path, encoding="utf-8") as fh:
blob = json.load(fh)
self.assertEqual(blob["schema_version"], 1)
self.assertEqual(blob["kind"], "model_param")
self.assertEqual(blob["items"][0]["meaning"], "学习率")
class SnapshotTest(unittest.TestCase):
"""快照与恢复（#66 基础）。"""
def setUp(self):
self.ctx = _TmpStore()
self.store = self.ctx.store
def tearDown(self):
self.ctx.cleanup()
def test_snapshot_captures_all_kinds(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8)
snap = self.store.snapshot()
self.assertIn("captured_at", snap)
self.assertEqual(set(snap["kinds"].keys()),
{"model_param", "rag_config", "layout"})
self.assertEqual(len(snap["kinds"]["model_param"]), 1)
def test_restore_replicates_state(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8)
snap = self.store.snapshot()
# 清空再恢复
self.store.delete(ConfigKind.MODEL_PARAM, "lr")
self.store.delete(ConfigKind.RAG_CONFIG, "top_k")
self.store.restore(snap)
self.assertEqual(self.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.001)
self.assertEqual(self.store.get(ConfigKind.RAG_CONFIG, "top_k").value, 8)
def test_item_counts(self):
self.store.upsert(ConfigKind.LAYOUT, "dashboard",
[{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}])
counts = self.store.item_counts()
self.assertEqual(counts["layout"], 1)
self.assertEqual(counts["model_param"], 0)
class ConfigItemSerializationTest(unittest.TestCase):
"""ConfigItem 序列化往返。"""
def test_roundtrip(self):
it = ConfigItem(key="lr", value=0.1, kind=ConfigKind.MODEL_PARAM,
meaning="学习率", updated_by="li", reason="init",
updated_at="2026-01-01T00:00:00Z")
d = it.to_dict()
self.assertEqual(d["kind"], "model_param")
it2 = ConfigItem.from_dict(d)
self.assertEqual(it2.value, 0.1)
self.assertEqual(it2.kind, ConfigKind.MODEL_PARAM)
if __name__ == "__main__":
unittest.main()