框架: A3收口——23型schema入库(23 schema/23 version/294 field/23 policy),字段级aiContext落visibility_policy.policy_snapshot;seed_schemas.py幂等种子挂db skill;两条主仓粒度缺口记设计发现(字段说明列缺失/字段级aiContext无显式列)

This commit is contained in:
zizi 2026-07-13 10:29:22 +08:00
parent ce1ab01b6d
commit 7d6cf34442
5 changed files with 140 additions and 2 deletions

View File

@ -24,6 +24,9 @@ description: muse-example 实验库的唯一数据库通道——查询/DML/DDL/
# 表清单+活行数deleted=FALSE 计数,无 deleted 列的表计全行)
.venv/bin/python .claude/skills/db/scripts/db.py tables
# A3 种子23 型 YAML → meta 表行(幂等可重跑;字段改动=改 YAML 后重跑)
.venv/bin/python .claude/skills/db/scripts/seed_schemas.py
```
## 红线

View File

@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""A323 型结构本体 YAML → meta 表行W1 种子演练,幂等可重跑)。
映射约定详见 db/表映射.md 与本次收口记录
- muse_meta_schema 一型一行schema_key=target_type状态 启用active / 待启用inactive
- muse_meta_schema_version 每型 v1active_flag=TRUEfield_contract_snapshot=完整 YAML判据/说明/设计发现全在
- muse_meta_field 规范化投影行基础字段 sort_order 19特有字段 11 段位即基础/特有约定
- muse_meta_visibility_policy 版本级一行ai_context=是否存在 AI 可见字段
字段级细则进 policy_snapshot.fieldAiContext主仓无字段级列设计发现待回填
- 幂等schema/version/policy upsertfield 行重跑=删旧插新meta 种子行豁免软删约定见表映射.md
"""
import json
import pathlib
import sys
import psycopg
import yaml
from psycopg.types.json import Jsonb
DSN = "postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
SCHEMA_DIR = pathlib.Path(__file__).resolve().parents[4] / "meta" / "schemas"
TENANT, ACTOR = 1, "1" # 实验写入约定:系统主账号
# 基础字段所有型共有README §使用规则aiContext 为实验设计判断,可在 B4 优化环调整):
# 名称/别名/摘要/标签全用途可见;来源(出处)生成不可见防抄袭腔;状态是系统轨道值不进上下文(授权过滤在查询层)
BASE_FIELDS = [
("名称", "text", True, None, True),
("别名", "array", False, None, True),
("一句话摘要", "text", True, None, True),
("标签", "array", False, None, True),
("来源", "text", False, None, ["planning", "detection", "extraction"]),
("状态", "enum", True, ["草稿", "已确认"], False),
]
def seed_one(conn, doc: dict, src_name: str):
"""单型入库schema → version(v1) → fields → visibility_policy返回审查摘要。"""
key = doc["target_type"]
status = "active" if doc.get("状态", "启用") == "启用" else "inactive"
# 1) schema 行 upsertuk: tenant_id+domain+scope+target_type+schema_key
row = conn.execute(
"""INSERT INTO muse_meta_schema (schema_key, domain, scope, target_type, display_name, status,
creator, updater, tenant_id)
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
ON CONFLICT (tenant_id, domain, scope, target_type, schema_key)
DO UPDATE SET display_name=EXCLUDED.display_name, status=EXCLUDED.status, updater=EXCLUDED.updater
RETURNING id""",
(key, doc["domain"], doc["scope"], key, doc.get("中文名", key), status,
ACTOR, ACTOR, TENANT)).fetchone()
schema_id = row[0]
# 2) version v1 upsertuk: tenant_id+schema_key+version_no快照=完整 YAML
row = conn.execute(
"""INSERT INTO muse_meta_schema_version (schema_id, schema_key, version_no, status, active_flag,
field_contract_snapshot, change_note, published_by, activated_by, creator, updater, tenant_id)
VALUES (%s,%s,1,'published',TRUE,%s,%s,%s,%s,%s,%s,%s)
ON CONFLICT (tenant_id, schema_key, version_no)
DO UPDATE SET field_contract_snapshot=EXCLUDED.field_contract_snapshot,
change_note=EXCLUDED.change_note, updater=EXCLUDED.updater
RETURNING id""",
(schema_id, key, Jsonb(doc), f"A3 种子入库(源:{src_name})", ACTOR, ACTOR, ACTOR, ACTOR, TENANT)).fetchone()
version_id = row[0]
conn.execute("UPDATE muse_meta_schema SET active_version_id=%s WHERE id=%s", (version_id, schema_id))
# 3) field 行:删旧插新(幂等重建)
conn.execute("DELETE FROM muse_meta_field WHERE tenant_id=%s AND schema_version_id=%s", (TENANT, version_id))
ai_map = {}
n_special = 0
for i, (fk, ftype, req, enum, ai) in enumerate(BASE_FIELDS, start=1):
conn.execute(
"""INSERT INTO muse_meta_field (schema_version_id, field_key, display_name, field_type,
is_required, enum_values, sort_order, creator, updater, tenant_id)
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
(version_id, fk, fk, ftype, req, Jsonb(enum) if enum else None, i, ACTOR, ACTOR, TENANT))
ai_map[fk] = ai
for i, f in enumerate(doc.get("特有字段") or [], start=11):
conn.execute(
"""INSERT INTO muse_meta_field (schema_version_id, field_key, display_name, field_type,
is_required, sort_order, creator, updater, tenant_id)
VALUES (%s,%s,%s,'text',FALSE,%s,%s,%s,%s)""",
(version_id, f["key"], f["key"], i, ACTOR, ACTOR, TENANT))
ai_map[f["key"]] = f.get("aiContext", True)
n_special += 1
# 4) visibility_policy版本级一行 upsert无 uk先查后写字段级细则进 policy_snapshot
any_ai = any(v is not False for v in ai_map.values())
snap = Jsonb({"fieldAiContext": ai_map,
"用途枚举": ["planning", "generation", "detection", "extraction"],
"语义": "true=任何用途可入AI上下文;false=一律不入;[用途]=仅列出的用途可入"})
exist = conn.execute(
"SELECT id FROM muse_meta_visibility_policy WHERE tenant_id=%s AND schema_version_id=%s",
(TENANT, version_id)).fetchone()
if exist:
conn.execute("UPDATE muse_meta_visibility_policy SET ai_context=%s, policy_snapshot=%s, updater=%s WHERE id=%s",
(any_ai, snap, ACTOR, exist[0]))
else:
conn.execute(
"""INSERT INTO muse_meta_visibility_policy (schema_version_id, ui_visible, ai_context,
user_editable, user_searchable, exportable, policy_snapshot, creator, updater, tenant_id)
VALUES (%s,TRUE,%s,TRUE,FALSE,FALSE,%s,%s,%s,%s)""",
(version_id, any_ai, snap, ACTOR, ACTOR, TENANT))
full_open = sum(1 for v in ai_map.values() if v is True)
scoped = sum(1 for v in ai_map.values() if isinstance(v, list))
closed = sum(1 for v in ai_map.values() if v is False)
return (key, doc.get("中文名", ""), doc["domain"], doc["scope"], status,
len(BASE_FIELDS) + n_special, full_open, scoped, closed)
def main():
files = sorted(p for p in SCHEMA_DIR.glob("*.yaml"))
if not files:
sys.exit(f"未找到 schema YAML: {SCHEMA_DIR}")
rows = []
with psycopg.connect(DSN) as conn:
for p in files:
doc = yaml.safe_load(p.read_text())
rows.append(seed_one(conn, doc, p.name))
conn.commit()
# 审查面:逐型卡片行
print(f"{'':<20}{'中文名':<10}{'domain':<11}{'scope':<10}{'状态':<9}{'字段':<5}{'AI全开':<7}{'限用途':<7}{'关闭'}")
for r in rows:
print(f"{r[0]:<20}{r[1]:<10}{r[2]:<11}{r[3]:<10}{r[4]:<9}{r[5]:<5}{r[6]:<7}{r[7]:<7}{r[8]}")
print(f"\n{len(rows)} 型入库(基础字段 {len(BASE_FIELDS)} + 各型特有;快照含判据/说明/设计发现全文)")
if __name__ == "__main__":
main()

View File

@ -175,7 +175,7 @@ flowchart LR
|---|---|---|---|---|
| A1原K0 | 基座 | `muse-example` 建库+vector 插件嵌入通道已实测Qwen3-8B 默认 4096、`dimensions:1024` 生效) | psql 实测输出 | 未做,第一步 |
| A2原K1 | 库表映射 | 主仓 `sql/muse` 摘录**一致** DDLmeta / work·chapter / knowledge 三域)→应用;实验私货全进 `example_*` 前缀(嵌入边表 vector(1024)、tenant/creator 默认系统主账号=1交付 `db` 查询 skill | `\dt` + 表↔主仓迁移来源映射 | 未做 |
| A3原K2 | G1 元结构治理 | 23 型 YAML → meta 表行(=W1 种子演练);此后拆书/抽取一律读**库内** schema | schema/字段行卡片打印 | YAML 就绪,入库未做 |
| A3原K2 | G1 元结构治理 | 23 型 YAML → meta 表行(=W1 种子演练);此后拆书/抽取一律读**库内** schema | schema/字段行卡片打印 | ✅ 已收口2026-07-1323 型/294 字段行,字段级 aiContext 入 policy_snapshotseed_schemas.py 幂等 |
| A4 | G2 系统能力治理 | 四槽位默认智能体(身份段)+ 功能 skill ×9 + 功能链登记表(`meta/chains/`+ read-context/confirm/eval 保护流程 | 保护节点不可被装配替换 | 已就位(文件侧);装配入库随阶段二 |
**阶段B 全局知识生产(管理线 G3 主体;审查=可见知识库数据)**

View File

@ -52,5 +52,6 @@
- `tenant_id=1``creator='1'``owner_user_id=1`系统主账号——写入层统一给值列默认仍照主仓0/''),不改列。
- 公共范式的全局知识行:`work_id=0` + `scope='global'` + 挂 `kb_type='global'` 的库;参考书私有库 `kb_type='user'`
- 软删照主仓:`deleted=TRUE`,不物理删。
- 软删照主仓:`deleted=TRUE`,不物理删。**例外**meta 种子行(`muse_meta_field`)幂等重跑=删旧插新(种子演练场景,豁免软删)。
- meta 字段行 sort_order 段位约定19 基础字段11 起特有字段。
- 双轨对应B2 产出全落 `muse_knowledge_draft(status='pending')`B5 管理员确认(仅创始人触发)= draft 翻 `confirmed` + 落 `muse_knowledge_entity(status='active')`;丢弃= draft 翻 `ignored`

View File

@ -40,3 +40,9 @@
- 2026-07-09 拆书首轮——「例证出处」五型统一设计有效(隔离抽象范式与出处、支撑脱敏);建议可选增「反例出处(哪里用砸了)」强化检测。
- 范式卡模板可考虑可选「边界」字段(本卡为何不是邻型),缓解复用时五型互混;需权衡卡片负担。
- 2026-07-13 A3 种子入库——主仓 `muse_meta_field` 无「字段说明」列(仅 display_name字段语义合同只能靠 version 的 `field_contract_snapshot` 承载;建议主仓补 description 列或钉死 snapshot 为字段语义 SoT回填候选
- 2026-07-13 A3 种子入库——字段级+用途级 aiContext专题-06 §7 裁剪所需粒度)在主仓无显式列:`muse_meta_visibility_policy.ai_context` 是版本级布尔。实验约定细则落 `policy_snapshot.fieldAiContext`true/false/[用途]),建议主仓明确该 JSONB 的 schema 合同(回填候选)。
## A3 入库状态2026-07-13
23 型已全部入库 `muse_meta_schema`(+version/field/visibility_policy),此后拆书/抽取一律读**库内** schema经 db skill本目录 YAML 退为设计稿与种子来源;字段增删先改 YAML 再重跑 `seed_schemas.py`(幂等),保持两侧一致。