458 lines
20 KiB
Python
458 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""知识草稿确认：把待审实体/关系写入作品正式知识面。
|
||
|
||
正文候选和知识卡共用 confirm skill 的主权边界，但知识表不是正文表：
|
||
实体写入 ``muse_knowledge_entity``，关系写入 ``muse_knowledge_relation``，
|
||
草稿状态、知识库绑定和向量 owner 在同一事务内完成。默认命令只确认点名
|
||
草稿；批量确认必须显式给出 ``--all-entities`` 或 ``--all-relations``。
|
||
"""
|
||
import argparse
|
||
import json
|
||
import pathlib
|
||
import sys
|
||
from copy import deepcopy
|
||
|
||
|
||
DB_SCRIPTS = pathlib.Path(__file__).resolve().parents[2] / "db" / "scripts"
|
||
sys.path.insert(0, str(DB_SCRIPTS))
|
||
from db import connect # noqa: E402
|
||
|
||
|
||
TENANT = 1
|
||
ACTOR = "1"
|
||
DEFAULT_KB_OWNER = 1
|
||
ENTITY_SOURCE_TYPES = frozenset({"chapter_extract", "parse_book", "upgrade_book"})
|
||
|
||
|
||
class KnowledgeConfirmationError(RuntimeError):
|
||
"""知识草稿不满足确认合同时失败关闭。"""
|
||
|
||
|
||
def _string(value, field, *, required=True):
|
||
if not isinstance(value, str) or (required and not value.strip()):
|
||
if required:
|
||
raise KnowledgeConfirmationError(f"{field} 必须是非空字符串")
|
||
return ""
|
||
return value.strip()
|
||
|
||
|
||
def _payload_value(payload, *keys, default=None):
|
||
for key in keys:
|
||
if key in payload:
|
||
return payload[key]
|
||
return default
|
||
|
||
|
||
def normalize_entity_payload(payload):
|
||
"""兼容作品卡英文键和拆书卡中文键，产出正式实体字段投影。"""
|
||
|
||
if not isinstance(payload, dict):
|
||
raise KnowledgeConfirmationError("实体 draft_payload 必须是 JSON 对象")
|
||
entity_type = _string(_payload_value(payload, "type", "型"), "entity.type")
|
||
name = _string(_payload_value(payload, "name", "名称"), "entity.name")
|
||
brief = _string(_payload_value(payload, "brief", "一句话摘要", default="?"), "entity.brief")
|
||
fields = _payload_value(payload, "fields", "字段", default={})
|
||
if not isinstance(fields, dict):
|
||
raise KnowledgeConfirmationError("entity.fields 必须是对象")
|
||
source = payload.get("source") or payload.get("来源") or {}
|
||
if not isinstance(source, dict):
|
||
source = {}
|
||
evidence = _payload_value(payload, "evidence", "证据", default="")
|
||
if evidence is not None and not isinstance(evidence, str):
|
||
raise KnowledgeConfirmationError("entity.evidence 必须是字符串")
|
||
source_ref = {
|
||
"workId": source.get("workId"),
|
||
"chapter": source.get("chapter"),
|
||
"chapterId": source.get("chapterId"),
|
||
"evidence": evidence or None,
|
||
}
|
||
source_ref = {key: value for key, value in source_ref.items() if value is not None}
|
||
return {
|
||
"entity_type": entity_type,
|
||
"name": name,
|
||
"normalized_name": name.casefold(),
|
||
"description": brief,
|
||
"attributes": deepcopy(fields),
|
||
"source_ref": source_ref,
|
||
"payload": payload,
|
||
}
|
||
|
||
|
||
def normalize_relation_payload(payload):
|
||
"""兼容关系草稿的 source/target 对象和字符串形状。"""
|
||
|
||
if not isinstance(payload, dict):
|
||
raise KnowledgeConfirmationError("关系 draft_payload 必须是 JSON 对象")
|
||
|
||
def endpoint(value, field):
|
||
if isinstance(value, dict):
|
||
value = _payload_value(value, "name", "名称")
|
||
return _string(value, field)
|
||
|
||
source = endpoint(_payload_value(payload, "source", "甲方"), "relation.source")
|
||
target = endpoint(_payload_value(payload, "target", "乙方"), "relation.target")
|
||
relation_type = _string(_payload_value(payload, "type", "关系类型"), "relation.type")
|
||
description = _string(
|
||
_payload_value(payload, "description", "描述", default="?"),
|
||
"relation.description",
|
||
)
|
||
attributes = _payload_value(payload, "attributes", "字段", default={})
|
||
if not isinstance(attributes, dict):
|
||
attributes = {"sourcePayload": attributes}
|
||
return {
|
||
"source_name": source,
|
||
"target_name": target,
|
||
"relation_type": relation_type,
|
||
"description": description,
|
||
"attributes": deepcopy(attributes),
|
||
"payload": payload,
|
||
}
|
||
|
||
|
||
def _json(value):
|
||
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
|
||
def _load_kb(conn, kb_id=None):
|
||
if kb_id is not None:
|
||
row = conn.execute(
|
||
"SELECT id,status,deleted FROM muse_knowledge_base "
|
||
"WHERE id=%s AND tenant_id=%s FOR SHARE",
|
||
(kb_id, TENANT),
|
||
).fetchone()
|
||
if not row or row[1] != "active" or row[2]:
|
||
raise KnowledgeConfirmationError(f"知识库不可用：kb_id={kb_id}")
|
||
return row[0]
|
||
row = conn.execute(
|
||
"SELECT id FROM muse_knowledge_base "
|
||
"WHERE tenant_id=%s AND owner_user_id=%s AND kb_type='user' "
|
||
"AND status='active' AND deleted=false ORDER BY id LIMIT 1 FOR SHARE",
|
||
(TENANT, DEFAULT_KB_OWNER),
|
||
).fetchone()
|
||
if not row:
|
||
raise KnowledgeConfirmationError("没有可用的用户私有知识库")
|
||
return row[0]
|
||
|
||
|
||
def _ensure_binding(conn, work_id, kb_id, *, dry_run=False):
|
||
row = conn.execute(
|
||
"SELECT id,binding_status FROM muse_knowledge_binding "
|
||
"WHERE tenant_id=%s AND work_id=%s AND kb_id=%s AND deleted=false FOR UPDATE",
|
||
(TENANT, work_id, kb_id),
|
||
).fetchone()
|
||
if row:
|
||
if row[1] != "active":
|
||
conn.execute(
|
||
"UPDATE muse_knowledge_binding SET binding_status='active', updater=%s, "
|
||
"revision=revision+1 WHERE id=%s",
|
||
(ACTOR, row[0]),
|
||
)
|
||
return row[0]
|
||
row = conn.execute(
|
||
"INSERT INTO muse_knowledge_binding(work_id,kb_id,binding_type,binding_scope,"
|
||
"binding_status,source_version,target_version,creator,updater,tenant_id) "
|
||
"VALUES (%s,%s,'read','work','active',1,1,%s,%s,%s) RETURNING id",
|
||
(work_id, kb_id, ACTOR, ACTOR, TENANT),
|
||
).fetchone()
|
||
return row[0]
|
||
|
||
|
||
def _lineage(current, source_ref, draft_id, source_type):
|
||
lineage = deepcopy(current) if isinstance(current, dict) else {}
|
||
refs = lineage.get("sourceRefs")
|
||
if not isinstance(refs, list):
|
||
refs = []
|
||
if source_ref and source_ref not in refs:
|
||
refs.append(source_ref)
|
||
lineage["sourceRefs"] = refs
|
||
lineage["lastDraftId"] = draft_id
|
||
lineage["lastSourceType"] = source_type
|
||
return lineage
|
||
|
||
|
||
def _find_or_insert_entity(conn, *, work_id, draft_id, draft, kb_id, decided_by):
|
||
card = normalize_entity_payload(draft["draft_payload"])
|
||
row = conn.execute(
|
||
"SELECT id,kb_id,description,attributes,lineage_payload,status,deleted,revision "
|
||
"FROM muse_knowledge_entity WHERE tenant_id=%s AND work_id=%s AND entity_type=%s "
|
||
"AND normalized_name=%s AND scope='local' FOR UPDATE",
|
||
(TENANT, work_id, card["entity_type"], card["normalized_name"]),
|
||
).fetchone()
|
||
if row and row[6]:
|
||
raise KnowledgeConfirmationError(
|
||
f"实体存在但已删除，禁止自动复活：work={work_id} name={card['name']}"
|
||
)
|
||
lineage = _lineage(row[4] if row else None, card["source_ref"], draft_id, draft["source_type"])
|
||
if row:
|
||
if row[5] != "active":
|
||
raise KnowledgeConfirmationError(
|
||
f"实体状态不可确认：entity={row[0]} status={row[5]}"
|
||
)
|
||
old_attributes = row[3] if isinstance(row[3], dict) else {}
|
||
attributes = {**old_attributes, **card["attributes"]}
|
||
entity_id = row[0]
|
||
conn.execute(
|
||
"UPDATE muse_knowledge_entity SET kb_id=%s,description=%s,attributes=%s::jsonb,"
|
||
"lineage_payload=%s::jsonb,updater=%s,revision=revision+1 "
|
||
"WHERE id=%s AND tenant_id=%s AND deleted=false",
|
||
(kb_id, card["description"], _json(attributes), _json(lineage), decided_by,
|
||
entity_id, TENANT),
|
||
)
|
||
return entity_id
|
||
row = conn.execute(
|
||
"INSERT INTO muse_knowledge_entity(work_id,kb_id,entity_type,normalized_name,scope,"
|
||
"description,attributes,source_status,source_action_policy,lineage_payload,status,"
|
||
"confidence,source_type,source_id,creator,updater,tenant_id) "
|
||
"VALUES (%s,%s,%s,%s,'local',%s,%s::jsonb,'active','allowed',%s::jsonb,'active',"
|
||
"%s,%s,%s,%s,%s,%s) RETURNING id",
|
||
(work_id, kb_id, card["entity_type"], card["normalized_name"], card["description"],
|
||
_json(card["attributes"]), _json(lineage), draft["confidence"] or 0.8,
|
||
draft["source_type"], draft["source_id"], decided_by, decided_by, TENANT),
|
||
).fetchone()
|
||
return row[0]
|
||
|
||
|
||
def _move_entity_embeddings(conn, draft_id, entity_id, *, updater):
|
||
rows = conn.execute(
|
||
"SELECT id,content_hash,model,entity_id FROM example_knowledge_embedding "
|
||
"WHERE tenant_id=%s AND draft_id=%s AND deleted=false FOR UPDATE",
|
||
(TENANT, draft_id),
|
||
).fetchall()
|
||
if len(rows) > 1:
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 存在多条活向量，禁止确认")
|
||
for embedding_id, content_hash, model, current_entity_id in rows:
|
||
if current_entity_id is not None and current_entity_id != entity_id:
|
||
raise KnowledgeConfirmationError(
|
||
f"向量已归属其他实体：embedding={embedding_id} entity={current_entity_id}"
|
||
)
|
||
conflict = conn.execute(
|
||
"SELECT id,entity_id,draft_id FROM example_knowledge_embedding "
|
||
"WHERE tenant_id=%s AND content_hash=%s AND model=%s AND deleted=false "
|
||
"AND id<>%s FOR UPDATE",
|
||
(TENANT, content_hash, model, embedding_id),
|
||
).fetchone()
|
||
if conflict and (conflict[1] is not None or conflict[2] != draft_id):
|
||
raise KnowledgeConfirmationError(
|
||
f"向量唯一 owner 冲突：embedding={embedding_id} other={conflict}"
|
||
)
|
||
conn.execute(
|
||
"UPDATE example_knowledge_embedding SET draft_id=NULL,entity_id=%s,updater=%s "
|
||
"WHERE id=%s AND tenant_id=%s AND deleted=false",
|
||
(entity_id, updater, embedding_id, TENANT),
|
||
)
|
||
return len(rows)
|
||
|
||
|
||
def _confirm_entity_locked(conn, row, *, kb_id, decided_by):
|
||
(draft_id, work_id, draft_type, payload, status, source_type, source_id, confidence,
|
||
source_status, source_action_policy, revision, entity_id, _target_object_id) = row
|
||
if draft_type != "entity":
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 不是 entity 草稿")
|
||
if status != "pending":
|
||
if status == "confirmed":
|
||
return {"draft_id": draft_id, "status": "already_confirmed", "entity_id": entity_id}
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 状态不可确认：{status}")
|
||
if source_type not in ENTITY_SOURCE_TYPES:
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 来源类型不允许：{source_type}")
|
||
if source_status not in ("active", "authorized") or source_action_policy != "allowed":
|
||
raise KnowledgeConfirmationError(
|
||
f"draft={draft_id} 来源资格不允许：status={source_status} policy={source_action_policy}"
|
||
)
|
||
draft = {
|
||
"draft_payload": payload,
|
||
"source_type": source_type,
|
||
"source_id": source_id,
|
||
"confidence": confidence,
|
||
}
|
||
entity_id = _find_or_insert_entity(
|
||
conn, work_id=work_id, draft_id=draft_id, draft=draft, kb_id=kb_id,
|
||
decided_by=decided_by,
|
||
)
|
||
vectors = _move_entity_embeddings(conn, draft_id, entity_id, updater=decided_by)
|
||
updated = conn.execute(
|
||
"UPDATE muse_knowledge_draft SET status='confirmed',entity_id=%s,revision=revision+1,"
|
||
"updater=%s WHERE id=%s AND tenant_id=%s AND deleted=false AND status='pending' "
|
||
"AND revision=%s RETURNING revision",
|
||
(entity_id, decided_by, draft_id, TENANT, revision),
|
||
).fetchone()
|
||
if not updated:
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 确认时 revision 冲突")
|
||
return {"draft_id": draft_id, "status": "confirmed", "entity_id": entity_id,
|
||
"embedding_count": vectors, "revision": updated[0]}
|
||
|
||
|
||
def _find_entity(conn, work_id, name):
|
||
return conn.execute(
|
||
"SELECT id FROM muse_knowledge_entity WHERE tenant_id=%s AND work_id=%s "
|
||
"AND normalized_name=%s AND status='active' AND deleted=false FOR SHARE",
|
||
(TENANT, work_id, name.casefold()),
|
||
).fetchone()
|
||
|
||
|
||
def _confirm_relation_locked(conn, row, *, kb_id, decided_by):
|
||
(draft_id, work_id, draft_type, payload, status, source_type, source_id, confidence,
|
||
source_status, source_action_policy, revision, entity_id, target_object_id) = row
|
||
if draft_type != "relation":
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 不是 relation 草稿")
|
||
if status != "pending":
|
||
if status == "confirmed":
|
||
return {"draft_id": draft_id, "status": "already_confirmed",
|
||
"relation_id": target_object_id}
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 状态不可确认：{status}")
|
||
if source_type not in ENTITY_SOURCE_TYPES:
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 来源类型不允许：{source_type}")
|
||
if source_status not in ("active", "authorized") or source_action_policy != "allowed":
|
||
raise KnowledgeConfirmationError(
|
||
f"draft={draft_id} 来源资格不允许：status={source_status} policy={source_action_policy}"
|
||
)
|
||
relation = normalize_relation_payload(payload)
|
||
source_row = _find_entity(conn, work_id, relation["source_name"])
|
||
target_row = _find_entity(conn, work_id, relation["target_name"])
|
||
if not source_row or not target_row:
|
||
missing = []
|
||
if not source_row:
|
||
missing.append(relation["source_name"])
|
||
if not target_row:
|
||
missing.append(relation["target_name"])
|
||
raise KnowledgeConfirmationError(
|
||
f"关系两端尚未确认：draft={draft_id} missing={','.join(missing)}"
|
||
)
|
||
command_id = f"confirm-knowledge-relation-{draft_id}"
|
||
row = conn.execute(
|
||
"INSERT INTO muse_knowledge_relation(work_id,source_entity_id,target_entity_id,"
|
||
"relation_type,description,attributes,source_status,source_action_policy,command_id,"
|
||
"creator,updater,tenant_id) VALUES (%s,%s,%s,%s,%s,%s::jsonb,'active','allowed',%s,%s,%s,%s) "
|
||
"ON CONFLICT (tenant_id,command_id) WHERE command_id IS NOT NULL DO NOTHING RETURNING id",
|
||
(work_id, source_row[0], target_row[0], relation["relation_type"], relation["description"],
|
||
_json(relation["attributes"]), command_id, decided_by, decided_by, TENANT),
|
||
).fetchone()
|
||
relation_id = row[0] if row else conn.execute(
|
||
"SELECT id FROM muse_knowledge_relation WHERE tenant_id=%s AND command_id=%s",
|
||
(TENANT, command_id),
|
||
).fetchone()[0]
|
||
conn.execute(
|
||
"UPDATE example_knowledge_embedding SET deleted=true,updater=%s "
|
||
"WHERE tenant_id=%s AND draft_id=%s AND entity_id IS NULL AND deleted=false",
|
||
(decided_by, TENANT, draft_id),
|
||
)
|
||
updated = conn.execute(
|
||
"UPDATE muse_knowledge_draft SET status='confirmed',target_object_id=%s,"
|
||
"revision=revision+1,updater=%s WHERE id=%s AND tenant_id=%s AND deleted=false "
|
||
"AND status='pending' AND revision=%s RETURNING revision",
|
||
(relation_id, decided_by, draft_id, TENANT, revision),
|
||
).fetchone()
|
||
if not updated:
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 确认时 revision 冲突")
|
||
return {"draft_id": draft_id, "status": "confirmed", "relation_id": relation_id,
|
||
"revision": updated[0]}
|
||
|
||
|
||
def _load_draft(conn, draft_id):
|
||
row = conn.execute(
|
||
"SELECT id,work_id,draft_type,draft_payload,status,source_type,source_id,confidence,"
|
||
"source_status,source_action_policy,revision,entity_id,target_object_id "
|
||
"FROM muse_knowledge_draft WHERE id=%s AND tenant_id=%s AND deleted=false FOR UPDATE",
|
||
(draft_id, TENANT),
|
||
).fetchone()
|
||
if not row:
|
||
raise KnowledgeConfirmationError(f"draft 不存在或已删除：{draft_id}")
|
||
return row
|
||
|
||
|
||
def confirm_draft(draft_id, *, kb_id=None, decided_by=ACTOR, dry_run=False):
|
||
"""确认一张实体或关系草稿；dry_run 走完整事务后回滚。"""
|
||
|
||
with connect() as conn:
|
||
try:
|
||
row = _load_draft(conn, draft_id)
|
||
selected_kb = _load_kb(conn, kb_id)
|
||
_ensure_binding(conn, row[1], selected_kb)
|
||
if row[2] == "entity":
|
||
result = _confirm_entity_locked(conn, row, kb_id=selected_kb, decided_by=decided_by)
|
||
elif row[2] == "relation":
|
||
result = _confirm_relation_locked(conn, row, kb_id=selected_kb, decided_by=decided_by)
|
||
else:
|
||
raise KnowledgeConfirmationError(f"draft={draft_id} 类型不支持：{row[2]}")
|
||
result["kb_id"] = selected_kb
|
||
if dry_run:
|
||
conn.rollback()
|
||
result["status"] = "dry_run_ok"
|
||
result["note"] = "试跑已回滚，未落库"
|
||
return result
|
||
conn.commit()
|
||
return result
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
|
||
|
||
def confirm_work(work_id, *, draft_type, kb_id=None, decided_by=ACTOR, dry_run=False):
|
||
"""按类型显式确认作品下全部 pending 草稿；任一张失败则整批回滚。"""
|
||
|
||
if draft_type not in {"entity", "relation"}:
|
||
raise ValueError("draft_type 只能是 entity 或 relation")
|
||
with connect() as conn:
|
||
try:
|
||
selected_kb = _load_kb(conn, kb_id)
|
||
_ensure_binding(conn, work_id, selected_kb)
|
||
rows = conn.execute(
|
||
"SELECT id,work_id,draft_type,draft_payload,status,source_type,source_id,confidence,"
|
||
"source_status,source_action_policy,revision,entity_id,target_object_id "
|
||
"FROM muse_knowledge_draft WHERE tenant_id=%s AND work_id=%s AND draft_type=%s "
|
||
"AND status='pending' AND deleted=false ORDER BY id FOR UPDATE",
|
||
(TENANT, work_id, draft_type),
|
||
).fetchall()
|
||
results = []
|
||
for row in rows:
|
||
if draft_type == "entity":
|
||
result = _confirm_entity_locked(
|
||
conn, row, kb_id=selected_kb, decided_by=decided_by
|
||
)
|
||
else:
|
||
result = _confirm_relation_locked(
|
||
conn, row, kb_id=selected_kb, decided_by=decided_by
|
||
)
|
||
results.append(result)
|
||
if dry_run:
|
||
conn.rollback()
|
||
return {"status": "dry_run_ok", "work_id": work_id, "draft_type": draft_type,
|
||
"count": len(results), "results": results, "note": "试跑已回滚，未落库"}
|
||
conn.commit()
|
||
return {"status": "confirmed", "work_id": work_id, "draft_type": draft_type,
|
||
"count": len(results), "results": results}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="知识草稿确认（Shadow -> Canonical）")
|
||
group = parser.add_mutually_exclusive_group(required=True)
|
||
group.add_argument("--draft-id", type=int, help="点名确认一张草稿")
|
||
group.add_argument("--all-entities", type=int, metavar="WORK_ID", help="显式确认作品下全部实体草稿")
|
||
group.add_argument("--all-relations", type=int, metavar="WORK_ID", help="显式确认作品下全部关系草稿")
|
||
parser.add_argument("--kb-id", type=int, default=None)
|
||
parser.add_argument("--decided-by", default=ACTOR)
|
||
parser.add_argument("--dry-run", action="store_true")
|
||
args = parser.parse_args()
|
||
try:
|
||
if args.draft_id is not None:
|
||
result = confirm_draft(
|
||
args.draft_id, kb_id=args.kb_id, decided_by=args.decided_by, dry_run=args.dry_run
|
||
)
|
||
else:
|
||
draft_type = "entity" if args.all_entities is not None else "relation"
|
||
work_id = args.all_entities if args.all_entities is not None else args.all_relations
|
||
result = confirm_work(
|
||
work_id, draft_type=draft_type, kb_id=args.kb_id,
|
||
decided_by=args.decided_by, dry_run=args.dry_run,
|
||
)
|
||
except (KnowledgeConfirmationError, ValueError) as error:
|
||
print(f"[拒绝] {error}", file=sys.stderr)
|
||
raise SystemExit(1) from error
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|