"
def _content_text(value):
"""原始 prompt/response 若是 JSON，格式化后再展示；普通文本原样保留。"""
if value is None:
return ""
text = str(value)
try:
return json.dumps(json.loads(text), ensure_ascii=False, indent=2)
except (TypeError, ValueError, json.JSONDecodeError):
return text
_RAW_KIND_LABELS = {
"prompt": "模型输入", "response": "模型输出", "stderr": "错误输出", "error": "错误信息",
}
def _raw_kind_label(value):
raw = "" if value is None else str(value)
return f"{html.escape(_RAW_KIND_LABELS.get(raw, '原始证据'))} {html.escape(raw)}"
def _reason_text(value):
if value is None or value == "":
return "—"
raw = str(value)
label = _REASON_LABELS.get(raw, "未登记原因")
return f"{html.escape(label)} {html.escape(raw)}"
def _row_dicts(cols, rows):
return [dict(zip(cols, row)) for row in rows]
def _query_href(path, current, **updates):
"""保留当前筛选条件，生成可复制的 GET 下钻地址。"""
values = {
key: vals[0]
for key, vals in (current or {}).items()
if vals and vals[0] not in (None, "")
}
for key, value in updates.items():
if value in (None, ""):
values.pop(key, None)
else:
values[key] = value
query = urlencode(values)
return f"{path}?{query}" if query else path
def _chapter_text(chapter_id, order_no, title, *, link_work=None):
"""章节主键只作为编码显示，用户看到章节序号和标题。"""
if order_no is None:
label = "未关联章节"
else:
label = f"第{order_no}章{f'《{title}》' if title else ''}"
if link_work is not None and order_no is not None:
label = f"{html.escape(label)}"
else:
label = html.escape(label)
if chapter_id is not None:
label += f" chapter_id={html.escape(str(chapter_id))}"
return label
def _mapping_html(mapping):
if not isinstance(mapping, dict) or not mapping:
return "暂无"
return ("
"
+ "".join(f"
{esc(key)}
{esc(_json_text(value))}
"
for key, value in mapping.items())
+ "
")
def display_trigger(value):
raw = "" if value is None else str(value)
if raw in _ENUM_LABELS["trigger_source"]:
return enum_text(raw, "trigger_source")
return annotated_text(value)
def _operator_text(value):
"""运行调用方可能是角色，也可能是技能；两者都先显示中文语义。"""
raw = "" if value is None else str(value)
if raw in _ENUM_LABELS.get("source_role", {}):
return enum_text(raw, "source_role")
return enum_text(raw, "skill_name")
def enum_chip(value, group="status"):
raw = "" if value is None else str(value)
visual = _ENUM_ROLES.get(group, {}).get(raw, raw)
label = _ENUM_LABELS.get(group, {}).get(raw)
if label is None:
label = raw if _is_human_text(raw) else "未登记"
return chip(visual, label, code=raw)
def chip(state, label=None, *, code=None):
"""状态标签：颜色按语义角色，图标+文字同在。未识别状态归 warn（待审）。"""
role = _STATE_MAP.get(str(state).lower(), "warn") if state else "gap"
if state in (None, "", "not_built"):
role = "gap"
r, icon = _STATUS[role]
raw = code if code is not None else state
if label is None:
label = _ENUM_LABELS.get("status", {}).get(str(raw), "未登记")
suffix = ""
if code is not None or (state and str(state).lower() not in _STATUS):
suffix = f" {html.escape(str(raw))}"
return (f'{icon} '
f'{html.escape(str(label))}{suffix}')
def esc(v):
if v is None:
return '∅'
return html.escape(str(v))
def fmt_num(v):
try:
return f"{int(v):,}"
except (TypeError, ValueError):
return esc(v)
def empty_gap(title, desc):
"""待落库空状态：中性虚线，与"失败"红严格区分。"""
return f"
"
f"{tiles}{ledger}{_quota_block()}")
return layout("总览", body, "/")
# ── 作品（01 作品领域：作品 → 章节 → 正文全文）────────────────────────────────
def view_works():
_, rows = q(
"SELECT w.id,w.title,w.status,w.genre,COUNT(DISTINCT c.id)::bigint AS chapter_count, "
"COALESCE(SUM(b.word_count),0)::bigint AS word_count "
"FROM muse_content_work w "
"LEFT JOIN muse_content_chapter c ON c.work_id=w.id AND c.deleted=false "
"LEFT JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=false "
"WHERE w.deleted=false GROUP BY w.id,w.title,w.status,w.genre ORDER BY w.id"
)
if not rows:
return layout("作品", "
")
return layout("作品", body, "/works")
def _knowledge_card_detail(wid, card):
payload = card.get("draft_payload") or {}
entity_id = card.get("entity_id")
source = payload.get("source") if isinstance(payload.get("source"), dict) else {}
target = payload.get("target") if isinstance(payload.get("target"), dict) else {}
name = (card.get("entity_name") or payload.get("name") or payload.get("名称") or
source.get("name") or target.get("name") or "未命名卡")
card_type = card.get("entity_type") or payload.get("type") or payload.get("型") or card.get("draft_type")
brief = (card.get("entity_description") or payload.get("brief") or payload.get("description") or
payload.get("摘要") or "")
fields = payload.get("fields") or payload.get("attributes") or {}
extract_run_id = payload.get("extractRunId") or payload.get("extract_run_id")
run_link = (f"{html.escape(str(extract_run_id))}"
if extract_run_id else "—")
chapter = _chapter_text(card.get("chapter_id"), card.get("chapter_order"), card.get("chapter_title"), link_work=wid)
entity_html = ""
if entity_id:
entity_html = (
"正式实体快照"
f"
实体编号
{esc(entity_id)}
"
f"
实体状态
{chip(card.get('entity_status'))}
"
f"
实体名称
{esc(card.get('entity_name'))}
"
f"
实体说明
{esc(card.get('entity_description'))}
"
f"{_data_block(card.get('entity_attributes'))}"
""
)
relation_html = ""
if source or target:
relation_html = ("
{esc(payload.get('evidence')) if payload.get('evidence') else '—'}
"
"
"
f"
字段
{_mapping_html(fields)}"
f"{entity_html}"
f"查看完整卡载荷{_data_block(payload)}"
+ (f"查看拟议变更{_data_block(card.get('proposed_changes'))}"
if card.get("proposed_changes") else "")
+ (f"查看当前正式快照{_data_block(card.get('current_canonical_snapshot'))}"
if card.get("current_canonical_snapshot") else "")
+ "
"
)
def _work_knowledge_panel(wid, params=None):
"""作品级知识链下钻：可筛选卡片、右侧详情、抽取运行与向量状态。"""
params = params or {}
if not table_exists("muse_knowledge_draft"):
return empty_gap("知识抽取 · 待落库", "知识草稿表尚未建立。")
_, summary = q(
"""SELECT
COUNT(*) FILTER (WHERE d.status='pending'),
COUNT(*) FILTER (WHERE d.status='confirmed'),
COUNT(*) FILTER (WHERE d.draft_type='entity' AND d.status='pending'),
COUNT(*) FILTER (WHERE d.draft_type='relation' AND d.status='pending'),
(SELECT COUNT(*) FROM muse_knowledge_entity e
WHERE e.work_id=%s AND e.deleted=false),
(SELECT COUNT(*) FROM muse_knowledge_relation r
WHERE r.work_id=%s AND r.deleted=false),
(SELECT COUNT(*) FROM muse_knowledge_binding b
WHERE b.work_id=%s AND b.deleted=false AND b.binding_status='active'),
(SELECT COUNT(*) FROM example_knowledge_embedding e
WHERE e.deleted=false AND (
EXISTS (SELECT 1 FROM muse_knowledge_draft d2
WHERE d2.id=e.draft_id AND d2.work_id=%s AND d2.deleted=false)
OR EXISTS (SELECT 1 FROM muse_knowledge_entity e2
WHERE e2.id=e.entity_id AND e2.work_id=%s AND e2.deleted=false)
))
FROM muse_knowledge_draft d
WHERE d.work_id=%s AND d.deleted=false""",
(wid, wid, wid, wid, wid, wid),
)
pending, confirmed_drafts, pending_entities, pending_relations, entities, relations, bindings, embeddings = summary[0]
search = (params.get("q") or [""])[0].strip()
type_filter = (params.get("type") or [""])[0].strip()
status_filter = (params.get("status") or [""])[0].strip()
chapter_filter = (params.get("chapter") or [""])[0].strip()
selected_card = (params.get("card") or [""])[0].strip()
try:
selected_card_id = int(selected_card) if selected_card else None
except ValueError:
selected_card_id = None
_, type_rows = q(
"""SELECT DISTINCT COALESCE(e.entity_type,d.draft_payload->>'type',d.draft_payload->>'型',d.draft_type)
FROM muse_knowledge_draft d LEFT JOIN muse_knowledge_entity e ON e.id=d.entity_id AND e.deleted=false
WHERE d.work_id=%s AND d.deleted=false ORDER BY 1""",
(wid,),
)
_, chapter_rows = q(
"SELECT id,order_no,title FROM muse_content_chapter WHERE work_id=%s AND deleted=false ORDER BY order_no",
(wid,),
)
type_options = [row[0] for row in type_rows if row[0]]
_, status_rows = q(
"SELECT DISTINCT status FROM muse_knowledge_draft WHERE work_id=%s AND deleted=false ORDER BY status",
(wid,),
)
status_options = [row[0] for row in status_rows if row[0]]
conditions = ["d.work_id=%s", "d.deleted=false"]
query_params = [wid]
if search:
like = f"%{search}%"
conditions.append("(d.draft_payload::text ILIKE %s OR COALESCE(e.normalized_name,'') ILIKE %s OR COALESCE(e.description,'') ILIKE %s)")
query_params.extend([like, like, like])
if type_filter:
conditions.append("COALESCE(e.entity_type,d.draft_payload->>'type',d.draft_payload->>'型',d.draft_type)=%s")
query_params.append(type_filter)
if status_filter:
conditions.append("d.status=%s")
query_params.append(status_filter)
if chapter_filter:
try:
chapter_order = int(chapter_filter)
except ValueError:
chapter_order = None
if chapter_order is not None:
conditions.append("c.order_no=%s")
query_params.append(chapter_order)
list_cols, list_rows = q(
"""SELECT d.id AS draft_id,d.entity_id,d.draft_type,d.status,d.confidence,d.source_type,d.source_id,
c.id AS chapter_id,c.order_no AS chapter_order,c.title AS chapter_title,
d.update_time,d.create_time,
COALESCE(e.normalized_name,d.draft_payload->>'name',d.draft_payload->>'名称',
d.draft_payload->'source'->>'name',d.draft_payload->'target'->>'name','未命名卡') AS entity_name,
COALESCE(e.entity_type,d.draft_payload->>'type',d.draft_payload->>'型',d.draft_type) AS entity_type,
COALESCE(e.description,d.draft_payload->>'brief',d.draft_payload->>'description','') AS summary,
EXISTS (SELECT 1 FROM example_knowledge_embedding emb
WHERE emb.deleted=false AND (emb.draft_id=d.id OR emb.entity_id=d.entity_id)) AS has_embedding
FROM muse_knowledge_draft d
LEFT JOIN muse_knowledge_entity e ON e.id=d.entity_id AND e.deleted=false
LEFT JOIN muse_content_chapter c ON c.id=d.source_id AND c.work_id=d.work_id AND c.deleted=false
WHERE """ + " AND ".join(conditions) +
" ORDER BY d.update_time DESC NULLS LAST,d.create_time DESC,d.id DESC",
tuple(query_params),
)
rows = _row_dicts(list_cols, list_rows)
detail = ("
卡详情
"
"
点击左侧实体/卡列表，右侧显示完整内容和来源证据。
")
if selected_card_id is not None:
detail_cols, detail_rows = q(
"""SELECT d.id AS draft_id,d.entity_id,d.draft_type,d.status,d.confidence,d.source_type,d.source_id,
d.revision,d.create_time,d.update_time,d.draft_payload,d.proposed_changes,
d.current_canonical_snapshot,d.source_snapshot_id,d.authorization_snapshot_id,
d.source_status,d.source_action_policy,
c.id AS chapter_id,c.order_no AS chapter_order,c.title AS chapter_title,
e.id AS entity_id_joined,e.normalized_name AS entity_name,e.entity_type AS entity_type,
e.description AS entity_description,e.attributes AS entity_attributes,e.status AS entity_status,
e.update_time AS entity_update_time,e.lineage_payload AS entity_lineage_payload
FROM muse_knowledge_draft d
LEFT JOIN muse_content_chapter c ON c.id=d.source_id AND c.work_id=d.work_id AND c.deleted=false
LEFT JOIN muse_knowledge_entity e ON e.id=d.entity_id AND e.deleted=false
WHERE d.id=%s AND d.work_id=%s AND d.deleted=false""",
(selected_card_id, wid),
)
if detail_rows:
card = dict(zip(detail_cols, detail_rows[0]))
card["entity_id"] = card.get("entity_id_joined")
detail = _knowledge_card_detail(wid, card)
else:
detail = ("
卡详情
"
"
该卡不属于当前作品，或已经被软删除。
")
def option(value, label, selected):
return f""
type_opts = "".join(option(value, _ENUM_LABELS.get("entity_type", {}).get(value, value), type_filter)
for value in type_options)
status_opts = "".join(option(value, _ENUM_LABELS.get("status", {}).get(value, value), status_filter)
for value in status_options)
chapter_opts = "".join(option(row[1], f"第{row[1]}章《{row[2]}》" if row[2] else f"第{row[1]}章", chapter_filter)
for row in chapter_rows)
filter_form = (
""
)
card_rows = []
for row in rows:
selected = " selected" if row["draft_id"] == selected_card_id else ""
current = dict(params or {})
href = _query_href(f"/works/{wid}", current, card=row["draft_id"])
summary_text = short_text(row["summary"], 120) if row["summary"] else "暂无摘要"
source_text = _chapter_text(row["chapter_id"], row["chapter_order"], row["chapter_title"])
card_rows.append(
f""
f"
"
) if runs else empty_gap("作品暂无运行", "运行登记后会显示模型、回执和原始证据下钻入口。")
return f"
知识闭环
抽取 → 草稿 → 向量 → 用户确认 → 正式实体/关系；看板只读，确认仍走确认通道。
{cards}{filter_form}{knowledge_grid}{runs_html}"
_SCHEMA_LABELS = {
"outline": "大纲结构", "novel_work": "作品设定", "narrative_state": "叙事状态", "assembly": "装配结构",
}
def _planning_title(payload, section_type, target_chapter):
payload = payload if isinstance(payload, dict) else {}
if payload.get("title"):
return str(payload["title"])
if payload.get("chapterGoal"):
return f"第{target_chapter}章目标" if target_chapter is not None else "章节目标"
if isinstance(payload.get("state"), dict) and payload["state"].get("currentSituation"):
return "当前叙事状态"
if section_type == "assembly":
return "范式装配结果"
return _ENUM_LABELS.get("section_type", {}).get(section_type, "规划内容")
def _planning_summary(payload):
payload = payload if isinstance(payload, dict) else {}
candidates = [
payload.get("chapterGoal"), payload.get("overall_arc"), payload.get("core_premise"),
payload.get("currentSituation"), payload.get("chapterEndHook"), payload.get("handoff"),
]
state = payload.get("state")
if isinstance(state, dict):
candidates.insert(3, state.get("currentSituation"))
return next((str(value) for value in candidates if value), "暂无摘要")
def _planning_detail(wid, section):
payload = section.get("payload") or {}
section_type = section.get("section_type")
title = _planning_title(payload, section_type, section.get("target_chapter"))
summary_rows = []
summary_keys = [
("title", "标题"), ("chapterGoal", "章节目标"), ("overall_arc", "总体弧线"),
("core_premise", "核心设定"), ("currentSituation", "当前局势"),
("chapterEndHook", "章末钩子"), ("handoff", "交接信息"),
]
for key, label in summary_keys:
value = payload.get(key)
if value:
summary_rows.append(f"
{label}
{esc(_json_text(value))}
")
state = payload.get("state")
if isinstance(state, dict) and state.get("currentSituation") and not payload.get("currentSituation"):
summary_rows.append(f"
"
)
list_body = "".join(items) if items else empty_gap("没有匹配的规划", "调整搜索或筛选条件后重试。")
pager_parts = [f"第 {page}/{page_count} 页 · 共 {total} 条"]
if page > 1:
pager_parts.append(f"上一页")
page_start = max(1, page - 2)
page_end = min(page_count, page + 2)
if page_start > 1:
pager_parts.append(f"1")
if page_start > 2:
pager_parts.append("…")
pager_parts.extend(
f"{number}" if number == page else
f"{number}"
for number in range(page_start, page_end + 1)
)
if page_end < page_count:
if page_end < page_count - 1:
pager_parts.append("…")
pager_parts.append(f"{page_count}")
if page < page_count:
pager_parts.append(f"下一页")
pager = "
" + "".join(pager_parts) + "
"
list_html = (f"
规划列表（{len(rows)} / {total}）默认按更新时间倒序
"
f"
{list_body}
{pager}
")
detail = ("
规划详情
"
"
点击左侧规划列表，右侧显示完整规划内容和版本信息。
")
if selected_id is not None:
detail_cols, detail_rows = q(
"""SELECT p.id AS planning_id,p.section_type,p.schema_type,p.target_chapter,p.version,p.state,p.confirmed_at,
p.create_time,p.update_time,p.payload,c.id AS chapter_id,c.order_no AS chapter_order,c.title AS chapter_title
FROM example_planning_section p
LEFT JOIN muse_content_chapter c ON c.work_id=p.work_id AND c.order_no=p.target_chapter AND c.deleted=false
WHERE p.id=%s AND p.work_id=%s AND p.deleted=false""",
(selected_id, wid),
)
if detail_rows:
detail = _planning_detail(wid, dict(zip(detail_cols, detail_rows[0])))
else:
detail = ("
规划详情
"
"
该规划不属于当前作品，或已经被软删除。
")
return f"{filter_form}
{list_html}{detail}
"
def view_work(wid, params=None):
_, wrows = q(
"SELECT w.id,w.title,w.status,w.genre,w.summary, "
"COALESCE(SUM(b.word_count),0)::bigint AS word_count, "
"COUNT(DISTINCT c.id)::bigint AS chapter_count "
"FROM muse_content_work w "
"LEFT JOIN muse_content_chapter c ON c.work_id=w.id AND c.deleted=false "
"LEFT JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=false "
"WHERE w.id=%s AND w.deleted=false GROUP BY w.id,w.title,w.status,w.genre,w.summary",
(wid,),
)
if not wrows:
return layout("作品", "
作品不存在
", "/works")
wid_, title, status, genre, summary, wc, cc = wrows[0]
_, chs = q("SELECT id, order_no, title, status, revision FROM muse_content_chapter "
"WHERE work_id=%s AND deleted=false ORDER BY order_no", (wid,))
if chs:
trs = "".join(
f"
{plan}{knowledge}")
return layout(title, body, "/works")
def view_chapter(wid, order_no):
_, wrow = q("SELECT title FROM muse_content_work WHERE id=%s AND deleted=false", (wid,))
if not wrow:
return layout("章节", "
作品不存在
", "/works")
wtitle = wrow[0][0]
_, ch = q("SELECT id,title,order_no FROM muse_content_chapter "
"WHERE work_id=%s AND order_no=%s AND deleted=false", (wid, order_no))
if not ch:
return layout("章节", "
章节不存在
", "/works")
cid, ctitle, _ = ch[0]
_, blocks = q("SELECT content_text, word_count FROM muse_content_block "
"WHERE chapter_id=%s AND deleted=false ORDER BY order_no", (cid,))
text = "\n\n".join((b[0] or "") for b in blocks)
body = (f"
{esc(ctitle) or f'第{order_no}章'}{fmt_num(sum(b[1] or 0 for b in blocks))} 字 · 正文全文（实时查库）
"
f"{esc(text) if text.strip() else '（本章正文为空）'}")
return layout(f"{wtitle} · 第{order_no}章", body, "/works")
# ── 知识库（02 实体 + 03 范式：知识卡 + 参考书·拆书·清洗）─────────────────────
def _barset(rows, slots, group=None):
if not rows:
return "
无数据
"
mx = max(r[1] for r in rows)
total = sum(r[1] for r in rows)
out = []
for i, (name, n) in enumerate(rows):
color = f"var(--s{slots[i % len(slots)]})"
display_name = enum_text(name, group) if group else esc(name)
out.append(f"
")
return "".join(out)
def view_knowledge(params=None):
params = params or {}
selected = (params.get("work") or [""])[0]
_, works = q("SELECT id,title FROM muse_content_work WHERE deleted=false ORDER BY id")
selector = _selector("按作品：", "work", [(w[0], w[1]) for w in works], selected)
if selected:
try:
work_id = int(selected)
except ValueError:
return layout("知识库", "
作品参数非法
", "/knowledge")
_, work = q("SELECT title FROM muse_content_work WHERE id=%s AND deleted=false", (work_id,))
if not work:
return layout("知识库", "
"
f"{selector}{_work_knowledge_panel(work_id)}")
return layout("知识库 · " + str(work[0][0]), body, "/knowledge")
drafts = one("SELECT count(*) FROM muse_knowledge_draft WHERE deleted=false") or 0
entities = one("SELECT count(*) FROM muse_knowledge_entity WHERE deleted=false") or 0
public = one("SELECT count(*) FROM muse_knowledge_draft WHERE work_id=0 AND deleted=false") or 0
work_cards = one("SELECT count(*) FROM muse_knowledge_draft WHERE work_id<>0 AND deleted=false") or 0
# 型别看载荷字段，不看 draft_type 列（库内约定，防误导）：公共卡用中文键「型」，作品卡用 type 编码。
_, pub_types = q("SELECT COALESCE(draft_payload->>'型','(无型)') t, count(*) FROM muse_knowledge_draft "
"WHERE work_id=0 AND deleted=false GROUP BY 1 ORDER BY 2 DESC")
_, wk_types = q("SELECT COALESCE(draft_payload->>'type','(无型)') t, count(*) FROM muse_knowledge_draft "
"WHERE work_id<>0 AND deleted=false GROUP BY 1 ORDER BY 2 DESC")
cards = f"""
"
for r in runs)
def _runs_query(where_sql="r.deleted=false", params=(), limit=50):
candidate_join = (
"LEFT JOIN LATERAL (SELECT ec.run_type,ec.source_role FROM example_candidate ec "
"WHERE ec.run_id=r.run_id AND ec.deleted=false ORDER BY ec.create_time,ec.id LIMIT 1) cand ON TRUE"
if table_exists("example_candidate") else
"LEFT JOIN LATERAL (SELECT NULL::text AS run_type,NULL::text AS source_role) cand ON TRUE"
)
raw_join = (
"LEFT JOIN LATERAL (SELECT rl.purpose FROM example_raw_lease rl "
"WHERE rl.run_id=r.run_id ORDER BY rl.create_time,rl.id LIMIT 1) raw ON TRUE"
if table_exists("example_raw_lease") else
"LEFT JOIN LATERAL (SELECT NULL::text AS purpose) raw ON TRUE"
)
receipt_join = (
"LEFT JOIN LATERAL (SELECT er.adapter_role,er.stage_kind FROM example_run_receipt er "
"WHERE er.run_id=r.run_id ORDER BY er.create_time,er.id LIMIT 1) first_receipt ON TRUE"
if table_exists("example_run_receipt") else
"LEFT JOIN LATERAL (SELECT NULL::text AS adapter_role,NULL::text AS stage_kind) first_receipt ON TRUE"
)
sql = (
"SELECT r.run_id,r.work_id,w.title AS work_title,r.target_chapter,r.trigger_source,r.terminal_state,r.started_at,"
"c.id AS chapter_id,c.order_no AS chapter_order,c.title AS chapter_title,"
"first_receipt.adapter_role,first_receipt.stage_kind,cand.run_type,cand.source_role,raw.purpose "
"FROM example_run r LEFT JOIN muse_content_work w ON w.id=r.work_id "
"LEFT JOIN muse_content_chapter c ON c.work_id=r.work_id AND c.order_no=r.target_chapter AND c.deleted=false "
f"{receipt_join} {candidate_join} {raw_join} WHERE {where_sql} "
"ORDER BY r.started_at DESC LIMIT %s"
)
_, runs = q(sql, tuple(params) + (limit,))
return runs
def _all_runs(limit=50):
return _runs_query(limit=limit)
def _runs_for_work(work_id, limit=50):
return _runs_query("r.deleted=false AND r.work_id=%s", (work_id,), limit)
def _selector(label, name, options, sel):
"""服务端渲染的下拉筛选（GET 表单，无 JS）。options=[(值,显示名)]。"""
opts = "".join(f""
for v, t in options)
return (f"")
def view_runs():
runs = _all_runs()
entries = ("
") if runs else \
empty_gap("暂无运行记录", "创作链落库后，每次运行登记一行（example_run）。")
if table_exists("example_candidate"):
_, cands = q("SELECT id, work_id, target_chapter, candidate_version, run_type, state, source_role "
"FROM example_candidate WHERE deleted=false AND state IN ('draft','checking','passed') "
"ORDER BY create_time DESC LIMIT 30")
cand = ("
待审候选（活跃）
"
"
候选编号 candidate_id
作品
目标章
版本
运行类型
状态
来源角色
"
+ "".join(f"
{esc(c[0])}
{esc(c[1])}
{esc(c[2])}
"
f"
{esc(c[3])}
{enum_text(c[4], 'run_type')}
{chip(c[5])}
{enum_text(c[6], 'source_role')}
" for c in cands)
+ "
") if cands else "
候选表已建 · 暂无待审候选
"
else:
cand = empty_gap("待审候选 · 待落库", "候选表尚未建立（编码：99），模型候选正文暂无持久化登记。")
if table_exists("example_user_decision"):
_, decs = q("SELECT work_id, target_chapter, decision, rationale, decided_by, create_time "
"FROM example_user_decision ORDER BY create_time DESC LIMIT 30")
dec = ("
用户决策（最近 30）
"
"
作品
章
决策
理据
决策人
时间
"
+ "".join(f"
{esc(d[0])}
{esc(d[1])}
{enum_chip(d[2], 'decision')}
"
f"
{annotated_text((d[3] or '')[:40])}
{esc(d[4])}
{esc(d[5])}
" for d in decs)
+ "
") if decs else "
决策表已建 · 暂无决策
"
else:
dec = empty_gap("用户决策 · 待落库", "决策记录表尚未建立（编码：99），接受/合并/丢弃暂以版本库提交留痕。")
body = (f"
运行记录智能体的操作与产出 · 流程 05 · 质量 06
"
f"{entries}{runs_html}
待审候选
{cand}
用户决策
{dec}")
return layout("运行记录", body, "/runs")
def view_runs_by_work(params):
sel = (params.get("work") or [""])[0]
_, works = q("SELECT id, title FROM muse_content_work WHERE deleted=false ORDER BY id")
selector = _selector("选作品：", "work", [(w[0], w[1]) for w in works], sel)
if sel:
runs = _runs_for_work(sel, 50)
else:
runs = _all_runs()
tbl = (_RUN_COLS + _run_rows(runs) + "") if runs else "
")
return layout("按作品看运行", body, "/runs")
def view_runs_by_role(params):
sel = (params.get("role") or [""])[0]
_, roles = q("SELECT role, display_name FROM example_agent_role WHERE deleted=false ORDER BY role")
selector = _selector("选角色：", "role", [(r[0], r[1] or r[0]) for r in roles], sel)
if sel:
runs = _runs_query(
"r.deleted=false AND EXISTS (SELECT 1 FROM example_run_receipt role_receipt "
"WHERE role_receipt.run_id=r.run_id AND role_receipt.adapter_role=%s)",
(sel,), 50,
)
else:
runs = _all_runs()
tbl = (_RUN_COLS + _run_rows(runs) + "") if runs else "
")
return layout("按角色看运行", body, "/runs")
def view_run_detail(run_id):
_, rr = q(
"""SELECT r.run_id,r.work_id,r.target_chapter,r.trigger_source,r.started_at,r.finished_at,r.terminal_state,
w.title AS work_title,c.id AS chapter_id,c.order_no AS chapter_order,c.title AS chapter_title
FROM example_run r
LEFT JOIN muse_content_work w ON w.id=r.work_id
LEFT JOIN muse_content_chapter c ON c.work_id=r.work_id AND c.order_no=r.target_chapter AND c.deleted=false
WHERE r.run_id=%s AND r.deleted=false""",
(run_id,),
)
if not rr:
return layout("运行", "
运行不存在
", "/runs")
rid, wid, chap, trig, started, finished, state, work_title, chapter_id, chapter_order, chapter_title = rr[0]
rc_cols, rc_rows = q(
"""SELECT id,sample_id,revision,stage_kind,adapter_role,attempt,loop_seq,candidate_version,is_error,
stop_reason,terminal_reason,safe_summary,raw_content_id,requested_model_id,actual_model_id,
total_cost_usd,usage,create_time
FROM example_run_receipt WHERE run_id=%s ORDER BY sample_id,revision""",
(run_id,),
)
rc = _row_dicts(rc_cols, rc_rows)
q_cols, q_rows = q(
"""SELECT judge_kind,dimension,scale_version,score,conclusion,failure_class,detail,raw_content_id,create_time
FROM example_quality_result WHERE run_id=%s ORDER BY create_time""",
(run_id,),
)
qres = _row_dicts(q_cols, q_rows)
call_cols, call_rows = q(
"""SELECT id,call_time,caller,requested_model_id,actual_model_id,in_tokens,cached_tokens,out_tokens,
cost_usd,stop_reason,duration_ms,prompt_sha256,raw_content_id
FROM example_llm_call WHERE run_id=%s ORDER BY call_time""",
(run_id,),
)
calls = _row_dicts(call_cols, call_rows)
candidates = []
if table_exists("example_candidate"):
candidate_cols, candidate_rows = q(
"SELECT id, target_chapter, candidate_version, run_type, state, source_role, "
"candidate_sha256, context_sha256 FROM example_candidate "
"WHERE run_id=%s AND deleted=false ORDER BY target_chapter, candidate_version",
(run_id,),
)
candidates = _row_dicts(candidate_cols, candidate_rows)
cand = (
"
候选（本运行）
"
"
候选编号 candidate_id
目标章
版本编码
运行类型
"
"
状态
来源角色
正文哈希
上下文哈希
"
+ "".join(
f"
{esc(c['id'])}
{esc(c['target_chapter'])}
"
f"
{esc(c['candidate_version'])}
{enum_text(c['run_type'], 'run_type')}
{chip(c['state'])}
"
f"
{enum_text(c['source_role'], 'source_role')}
{esc(c['candidate_sha256'])}
{esc(c['context_sha256'])}
"
for c in candidates
)
+ "
"
if candidates else "
候选表已建 · 该运行暂无候选
"
)
else:
cand = empty_gap("待审候选 · 待落库", "候选表尚未建立（编码：99）。")
if table_exists("example_user_decision") and table_exists("example_candidate"):
decision_cols, decision_rows = q(
"SELECT d.id,d.decision,d.rationale,d.decided_by,d.canonical_block_id,d.create_time "
"FROM example_user_decision d JOIN example_candidate c ON c.id=d.candidate_id "
"WHERE c.run_id=%s ORDER BY d.create_time",
(run_id,),
)
decisions = _row_dicts(decision_cols, decision_rows)
dec = (
"
用户决策（本运行）
"
"
决策编号 decision_id
决策
理据
决策人
"
"
正式正文块
时间
"
+ "".join(
f"
{esc(d['id'])}
{enum_chip(d['decision'], 'decision')}
"
f"
{annotated_text(d['rationale'])}
{esc(d['decided_by'])}
{esc(d['canonical_block_id'])}
{esc(d['create_time'])}
"
for d in decisions
)
+ "
"
if decisions else "
决策表已建 · 该运行暂无决策
"
)
else:
dec = empty_gap("用户决策 · 待落库", "决策表(99)或候选表未建。")
raw_rows = []
if table_exists("example_raw_lease"):
raw_cols, raw_data = q(
"SELECT l.id,l.purpose,l.status,count(c.id),max(c.create_time) "
"FROM example_raw_lease l LEFT JOIN example_raw_content c ON c.lease_id=l.id "
"WHERE l.run_id=%s GROUP BY l.id,l.purpose,l.status ORDER BY l.id",
(run_id,),
)
raw_rows = [dict(zip(raw_cols, row)) for row in raw_data]
raw_html = (
"
原始证据（本运行）
"
"
租约编号 lease_id
用途
状态
内容数
最近写入
"
+ "".join(
f"
{esc(r['id'])}
{enum_text(r['purpose'], 'purpose')}
{enum_chip(r['status'], 'status')}
"
f"
{esc(r['count'])}
{esc(r['max'])}
" for r in raw_rows
)
+ "
"
if raw_rows else empty_gap("原始证据 · 该运行暂无记录", "模型输入输出落库后在此显示。")
)
else:
raw_html = empty_gap("原始证据 · 待落库", "原始证据表尚未建立（编码：example_raw_lease）。")
raw_content = []
if table_exists("example_raw_content"):
raw_content_cols, raw_content_rows = q(
"SELECT id,kind,role,content_sha256,content,create_time FROM example_raw_content "
"WHERE run_id=%s ORDER BY create_time,id", (run_id,)
)
raw_content = _row_dicts(raw_content_cols, raw_content_rows)
raw_details = []
for index, content in enumerate(raw_content):
open_attr = " open" if index == len(raw_content) - 1 else ""
raw_details.append(
f"{_raw_kind_label(content['kind'])} · 内容编号 {esc(content['id'])} · "
f"{len(content['content'] or '')} 字 · 哈希 {esc(content['content_sha256'])}"
f"
{esc(_content_text(content['content']))}
"
)
raw_detail_html = (
"
模型输入 / 输出正文
"
+ "".join(raw_details) + "
"
if raw_details else empty_gap("暂无模型输入输出", "原始内容落库后在此显示完整正文。")
)
if table_exists("example_context_freeze") and wid is not None and chap is not None:
_, freezes = q(
"SELECT id,as_of_chapter,manifest_sha256,context_sha256,token_budget,create_time "
"FROM example_context_freeze WHERE work_id=%s AND target_chapter=%s "
"ORDER BY create_time DESC",
(wid, chap),
)
freeze_html = (
"
") if rc else empty_gap("暂无回执", "该运行的回执落库后在此显示（修订链编码：revision）。")
q_html = ("
质量评判
"
"
评判类型
维度编号
量表版本编码
分数
结论
失败类别编码
详细信息
"
+ "".join(
f"
{enum_chip(u['judge_kind'], 'judge_kind')}
{esc(u['dimension'])}
{esc(u['scale_version'])}
"
f"
{esc(u['score'])}
{enum_chip(u['conclusion'], 'conclusion')}
{esc(u['failure_class'])}
"
f"
{('查看' + _data_block(u['detail']) + '') if u['detail'] else '—'}
"
for u in qres)
+ "
") if qres else empty_gap("暂无质量评判", "")
call_html = ("
模型调用
"
"
时间
调用方
请求模型编码
实际模型编码
输入令牌
缓存令牌
输出令牌
停止原因
耗时毫秒
原始内容
"
+ "".join(
f"
{esc(c['call_time'])}
{_operator_text(c['caller'])}
{esc(c['requested_model_id'])}
{esc(c['actual_model_id'])}
"
f"
{fmt_num(c['in_tokens'])}
{fmt_num(c['cached_tokens'])}
{fmt_num(c['out_tokens'])}
"
f"
{_reason_text(c['stop_reason'])}
{fmt_num(c['duration_ms'])}
{esc(c['raw_content_id'])}
"
for c in calls)
+ "
") if calls else empty_gap("暂无模型调用", "")
error_items = []
for r in rc:
abnormal_stop = r["stop_reason"] not in (None, "", "completed", "tool_use")
abnormal_terminal = r["terminal_reason"] not in (None, "", "completed")
if r["is_error"] or abnormal_stop or abnormal_terminal:
summary = _data_block(r["safe_summary"]) if r["safe_summary"] else "—"
error_items.append(
f"
{esc(disp or _ENUM_LABELS['agent_role'].get(role_, '未登记'))} {esc(role_)}
"
f"
"
f"
模型归属
{esc(model)}
"
f"
工具
{_fmt_tools(tools)}
"
f"
可配置项
工作强度 effort：{esc(effort)} · 预算：{esc(budget)}
"
f"
出处编码
{esc(src)} · 同步于 {esc(synced)}
"
f"
职责
{annotated_text(resp)}
"
f"
")
_, runs = q("SELECT run_id, stage_kind, revision, create_time FROM example_run_receipt "
"WHERE adapter_role=%s ORDER BY create_time DESC LIMIT 20", (role,))
runs_html = ("
本角色运行（回执，最近 20）
"
"
运行编号 run_id
阶段
修订号 revision
时间
"
+ "".join(f"
{esc(r[0])}
{enum_text(r[1], 'stage_kind')}
{esc(r[2])}
{esc(r[3])}
" for r in runs)
+ "
") if runs else empty_gap("本角色暂无运行回执", "创作链落库后，该角色的运行会显示在这里。")
_, calls = q("SELECT call_time, requested_model_id, out_tokens, cost_usd FROM example_llm_call "
"WHERE caller=%s ORDER BY call_time DESC LIMIT 20", (role,))
calls_html = ("
本角色模型调用（最近 20）
"
"
时间
模型编码
输出令牌数
成本（美元）
"
+ "".join(f"
{esc(c[0])}
{esc(c[1])}
{fmt_num(c[2])}
${float(c[3] or 0):.5f}
" for c in calls)
+ "
") if calls else empty_gap("本角色暂无模型调用", "")
_, quals = q("SELECT q.judge_kind, q.dimension, q.conclusion, q.failure_class FROM example_quality_result q "
"JOIN example_run_receipt r ON r.id=q.receipt_id WHERE r.adapter_role=%s ORDER BY q.create_time DESC LIMIT 20", (role,))
quals_html = ("
本角色质量评判（最近 20）
"
"
类型
维度
结论
失败类别
"
+ "".join(f"
{enum_text(u[0], 'judge_kind')}
{esc(u[1])}
{enum_chip(u[2], 'conclusion')}
{esc(u[3])}
" for u in quals)
+ "
") if quals else empty_gap("本角色暂无质量评判", "")
body = (f"
智能体 / {esc(disp or _ENUM_LABELS['agent_role'].get(role_, '未登记'))}
"
f"
{esc(disp or _ENUM_LABELS['agent_role'].get(role_, '未登记'))}
{info}
运行与产出
{runs_html}{quals_html}{calls_html}")
return layout(f"角色 · {disp or _ENUM_LABELS['agent_role'].get(role_, '未登记')}", body, "/agents")
def view_skill(name):
_, ss = q("SELECT skill_name, purpose, consumer, reads, writes, red_lines, model_used, source_ref, synced_at "
"FROM example_skill WHERE skill_name=%s AND deleted=false", (name,))
if not ss:
return layout("技能", "