weichen
1c25d65ae9
Why: The U1 controlled trial finished; its fixtures, validator, runner, per-run results, report, and renderer spike must be reviewable from the task branch, and the task card needs the final PASS/FAIL record. Evidence: results/summary.json + report.md (SELF, MiniMax official channel, CNY 3 of 200 cap); make evidence receipts for precheck, task-sync, and make test follow in this branch. Impact: Task-scoped experiment artifacts only (.task/genui-u1/); no production code changes. Refs: Documents PR #62 (proposal r2); task card feature/generative-ui-u1
273 lines
15 KiB
Python
273 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""U1 试验共享库：MiniMax 调用、组件目录提示、规格校验器（原型）。"""
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import urllib.request
|
||
|
||
BASE = os.environ.get("MINIMAX_BASE", "https://api.minimaxi.com/v1")
|
||
KEY = os.environ.get("MINIMAX_API_KEY", "")
|
||
|
||
NODE_TYPES = {
|
||
"section", "list", "kv", "form", "markdown", "doc-ref", "actions",
|
||
"chart-lite", "avatar", "progress", "compare", "tree", "table",
|
||
"steps", "tabs", "timeline",
|
||
}
|
||
NODE_FIELDS = { # 各类型允许的顶层字段（公共字段 id/source 之外）
|
||
"section": {"type", "title", "children"}, "list": {"type", "items"},
|
||
"kv": {"type", "kv"}, "form": {"type", "fields"},
|
||
"markdown": {"type", "markdown"}, "doc-ref": {"type", "docId"},
|
||
"actions": {"type", "actions"}, "chart-lite": {"type", "chart"},
|
||
"avatar": {"type", "avatar"}, "progress": {"type", "progress"},
|
||
"compare": {"type", "compare"}, "tree": {"type", "tree"},
|
||
"table": {"type", "table"}, "steps": {"type", "steps"},
|
||
"tabs": {"type", "tabs"}, "timeline": {"type", "timeline"},
|
||
}
|
||
TONES = {"ok", "warn", "danger", "info", ""}
|
||
MAX_NODES, MAX_DEPTH, MAX_BYTES = 200, 8, 256 * 1024
|
||
|
||
CATALOG_PROMPT = """你是本产品中的页面生成助手：宿主给出材料与目标，你输出一页声明式页面规格（JSON），由宿主通用渲染器渲染。
|
||
组件目录（仅允许以下 16 种节点类型；字段必须严格匹配，禁止出现目录外的类型或字段）：
|
||
- section: {"title"?, "children"?:[节点]}
|
||
- list: {"items":["字符串"]}
|
||
- kv: {"kv":[["键","值"],...]}
|
||
- form: {"fields":[{"key","label","type":"text|textarea|number|boolean|select","options"?（select 必填）,"required"?,"placeholder"?,"help"?,"default"?}]}
|
||
- markdown: {"markdown":"正文"}
|
||
- doc-ref: {"docId":"工件ID"}（docId 必须是材料中出现过的 sourceId）
|
||
- actions: {"actions":[{"id","label","style"?:"primary|default|danger"}]}（不带 command，点击降级为发送消息）
|
||
- chart-lite: {"chart":{"kind":"bar|line","labels":["…"],"series":[数值]}}（labels 与 series 等长）
|
||
- avatar: {"avatar":{"name","sub"?,"tone"?:"ok|warn|danger|info","stack"?}}
|
||
- progress: {"progress":{"label"?,"value":0~1 之间的数值}}
|
||
- compare: {"compare":{"left":{"title"?,"tone"?,"nodes"?:[节点]},"right":{同 left}}}
|
||
- tree: {"tree":{"nodes":[{"id","label","tone"?,"badge"?,"count"?,"children"?}]}}
|
||
- table: {"table":{"columns":[{"key","label"}],"rows":[{"id"?,"cells":[["列key","值"],...]}]}}（cells 与 columns 等长同序）
|
||
- steps: {"steps":{"current":0 起下标,"steps":[{"label","status"?:"pending|active|done|failed","tone"?}]}}
|
||
- tabs: {"tabs":{"tabs":[{"id","label","nodes"?:[节点]}],"active"?}}
|
||
- timeline: {"timeline":{"items":[{"title","time"?,"tone"?,"markdown"?,"ref"?}]}}
|
||
规则：
|
||
1. 每个节点可以带 "id"（整页唯一且稳定，如 n1、n2a）与 "source"（数组：支撑该节点事实内容的材料 sourceId，只能使用给定材料中的 id）。
|
||
2. 事实性内容必须来自材料并在相应节点标注 source；你的分析、建议、推断只能放 markdown 节点，且不加 source。
|
||
3. 硬限制：整页 ≤200 节点、嵌套 ≤8 层、规格 ≤256KiB；未知类型/字段、引用不存在的 sourceId 都会被宿主拒绝。
|
||
4. 结构服务目标：横向比较→compare/table；风险→按严重度组织并标来源；步骤/顺序→steps/timeline；收集输入→form；概览→section/tabs 组织。
|
||
5. 只输出一个 JSON 对象 {"title":"…","nodes":[…]}，不要输出任何其他文字、解释或代码围栏。"""
|
||
|
||
BASELINE_PROMPT = """你是本产品中的助手。依据给定材料回答用户目标，输出为一张 Markdown 卡片（当前系统的卡片形态）。
|
||
只使用材料中的事实，不要编造；你的判断与建议需口头标明是分析。输出即卡片正文，不要额外说明。"""
|
||
|
||
|
||
def call_llm(model, system, user, max_tokens, temperature=0.3, retries=3):
|
||
body = json.dumps({
|
||
"model": model, "temperature": temperature, "max_tokens": max_tokens,
|
||
"messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],
|
||
}).encode()
|
||
last = None
|
||
for i in range(retries):
|
||
req = urllib.request.Request(
|
||
BASE + "/chat/completions", data=body,
|
||
headers={"Authorization": "Bearer " + KEY, "Content-Type": "application/json"})
|
||
t0 = time.time()
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=180) as r:
|
||
d = json.loads(r.read())
|
||
msg = (d.get("choices") or [{}])[0].get("message", {}) or {}
|
||
content = msg.get("content", "") or ""
|
||
raw_content = content
|
||
# MiniMax 会把 <think>…</think> 推理段内联进 content；剥离后才是可用的输出体。
|
||
content = re.sub(r"<think>.*?</think>\s*", "", content, flags=re.S).strip()
|
||
return {
|
||
"content": content, "raw_content": raw_content,
|
||
"usage": d.get("usage", {}),
|
||
"latency_s": round(time.time() - t0, 2),
|
||
}
|
||
except Exception as e: # 429/5xx/超时：退避重试
|
||
last = str(e)[:200]
|
||
time.sleep(3 * (i + 1))
|
||
raise RuntimeError(f"llm call failed: {last}")
|
||
|
||
|
||
def extract_json(text):
|
||
"""从回复中剥离围栏/杂文，解析唯一 JSON 对象。"""
|
||
t = text.strip()
|
||
m = re.search(r"```(?:json)?\s*(.+?)```", t, re.S)
|
||
if m:
|
||
t = m.group(1).strip()
|
||
if not t.startswith("{"):
|
||
s, e = t.find("{"), t.rfind("}")
|
||
if s >= 0 and e > s:
|
||
t = t[s:e + 1]
|
||
return json.loads(t)
|
||
|
||
|
||
def _err(errors, path, msg):
|
||
errors.append(f"{path}: {msg}")
|
||
|
||
|
||
def _check_str_list(v, path, errors, field):
|
||
if not isinstance(v, list) or any(not isinstance(x, str) for x in v):
|
||
_err(errors, path, f"{field} 必须为字符串数组")
|
||
|
||
|
||
def validate_spec(spec, source_ids):
|
||
"""严格校验：返回 (errors, stats)。stats 供评分（节点计数、深度、引用等）。"""
|
||
errors, stats = [], {"nodes": 0, "maxDepth": 0, "idsSeen": [], "refsUsed": set(), "unknownRefs": set(), "types": []}
|
||
|
||
def walk(node, path, depth):
|
||
stats["nodes"] += 1
|
||
stats["maxDepth"] = max(stats["maxDepth"], depth)
|
||
if not isinstance(node, dict):
|
||
_err(errors, path, "节点必须是对象"); return
|
||
t = node.get("type")
|
||
if t not in NODE_TYPES:
|
||
_err(errors, path, f"未知节点类型 {t!r}"); return
|
||
stats["types"].append(t)
|
||
extra = set(node) - NODE_FIELDS[t] - {"id", "source"}
|
||
if extra:
|
||
_err(errors, path, f"类型 {t} 不允许字段 {sorted(extra)}")
|
||
nid = node.get("id")
|
||
if nid is not None:
|
||
if not isinstance(nid, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,40}", nid):
|
||
_err(errors, path, f"非法 id {nid!r}")
|
||
elif nid in stats["idsSeen"]:
|
||
_err(errors, path, f"重复 id {nid!r}")
|
||
else:
|
||
stats["idsSeen"].append(nid)
|
||
src = node.get("source")
|
||
if src is not None:
|
||
if not isinstance(src, list) or any(not isinstance(x, str) for x in src):
|
||
_err(errors, path, "source 必须为字符串数组")
|
||
else:
|
||
for s in src:
|
||
(stats["refsUsed"] if s in source_ids else stats["unknownRefs"]).add(s)
|
||
if t == "list":
|
||
_check_str_list(node.get("items"), path, errors, "items")
|
||
elif t == "kv":
|
||
kv = node.get("kv")
|
||
if not isinstance(kv, list) or any(not (isinstance(p, list) and len(p) == 2 and all(isinstance(x, str) for x in p)) for p in kv):
|
||
_err(errors, path, "kv 必须为 [键,值] 对数组")
|
||
elif t == "markdown":
|
||
if not isinstance(node.get("markdown"), str):
|
||
_err(errors, path, "markdown 必须为字符串")
|
||
elif t == "doc-ref":
|
||
d = node.get("docId")
|
||
if not isinstance(d, str) or d not in source_ids:
|
||
_err(errors, path, f"docId 必须是材料 sourceId，得到 {d!r}")
|
||
else:
|
||
stats["refsUsed"].add(d)
|
||
elif t == "actions":
|
||
acts = node.get("actions")
|
||
if not isinstance(acts, list):
|
||
_err(errors, path, "actions 必须为数组")
|
||
else:
|
||
for a in acts:
|
||
if not isinstance(a, dict) or not (isinstance(a.get("id"), str) and isinstance(a.get("label"), str)) or a.get("style", "") not in {"primary", "default", "danger", ""}:
|
||
_err(errors, path, f"非法 action {a!r}")
|
||
elif "command" in a:
|
||
_err(errors, path, "试验目录不允许 command 字段")
|
||
elif t == "form":
|
||
fields = node.get("fields")
|
||
if not isinstance(fields, list) or not fields:
|
||
_err(errors, path, "form.fields 必须为非空数组")
|
||
else:
|
||
for f in fields:
|
||
ok = isinstance(f, dict) and isinstance(f.get("key"), str) and isinstance(f.get("label"), str) \
|
||
and f.get("type") in {"text", "textarea", "number", "boolean", "select"}
|
||
if not ok:
|
||
_err(errors, path, f"非法字段 {f!r}"); continue
|
||
if f["type"] == "select" and not isinstance(f.get("options"), list):
|
||
_err(errors, path, f"select 字段 {f['key']} 缺 options")
|
||
extraF = set(f) - {"key", "label", "type", "options", "required", "placeholder", "help", "default"}
|
||
if extraF:
|
||
_err(errors, path, f"字段 {f['key']} 不允许 {sorted(extraF)}")
|
||
elif t == "chart-lite":
|
||
c = node.get("chart")
|
||
if not isinstance(c, dict) or c.get("kind") not in {"bar", "line"} \
|
||
or not isinstance(c.get("labels"), list) or not isinstance(c.get("series"), list) \
|
||
or len(c["labels"]) != len(c["series"]) or any(not isinstance(x, (int, float)) for x in c["series"]):
|
||
_err(errors, path, "chart 需 kind∈{bar,line} 且 labels/series 等长数值")
|
||
elif t == "progress":
|
||
p = node.get("progress")
|
||
if not isinstance(p, dict) or not isinstance(p.get("value"), (int, float)) or not (0 <= p["value"] <= 1):
|
||
_err(errors, path, "progress.value 需 0~1")
|
||
elif t == "avatar":
|
||
a = node.get("avatar")
|
||
if not isinstance(a, dict) or not isinstance(a.get("name"), str) or a.get("tone", "") not in TONES:
|
||
_err(errors, path, "非法 avatar")
|
||
elif t == "tree":
|
||
ns = node.get("tree", {}).get("nodes")
|
||
if not isinstance(ns, list):
|
||
_err(errors, path, "tree.nodes 必须为数组")
|
||
else:
|
||
def wk(n):
|
||
if not isinstance(n, dict) or not (isinstance(n.get("id"), str) and isinstance(n.get("label"), str)) or n.get("tone", "") not in TONES:
|
||
_err(errors, path, f"非法树节点 {n!r}")
|
||
else:
|
||
for c in n.get("children") or []:
|
||
wk(c)
|
||
for n in ns:
|
||
wk(n)
|
||
elif t == "table":
|
||
tb = node.get("table")
|
||
cols = (tb or {}).get("columns") if isinstance(tb, dict) else None
|
||
rows = (tb or {}).get("rows") if isinstance(tb, dict) else None
|
||
if not isinstance(cols, list) or not cols or any(not (isinstance(c, dict) and isinstance(c.get("key"), str) and isinstance(c.get("label"), str)) for c in cols):
|
||
_err(errors, path, "table.columns 非法")
|
||
elif not isinstance(rows, list):
|
||
_err(errors, path, "table.rows 必须为数组")
|
||
else:
|
||
for r in rows:
|
||
cells = r.get("cells") if isinstance(r, dict) else None
|
||
if not isinstance(cells, list) or len(cells) != len(cols) or any(not (isinstance(c, list) and len(c) == 2 and all(isinstance(x, str) for x in c)) for c in cells):
|
||
_err(errors, path, "row.cells 必须与 columns 等长且为 [key,值] 对")
|
||
elif t == "steps":
|
||
st = node.get("steps")
|
||
steps = (st or {}).get("steps") if isinstance(st, dict) else None
|
||
if not isinstance(steps, list) or not steps or not isinstance(st.get("current"), int) or not (0 <= st["current"] < len(steps)) \
|
||
or any(not isinstance(s.get("label"), str) or s.get("status", "") not in {"pending", "active", "done", "failed", ""} for s in steps if isinstance(s, dict)):
|
||
_err(errors, path, "steps 需非空、current 在范围内、status 合法")
|
||
elif t == "tabs":
|
||
ts = node.get("tabs", {}).get("tabs") if isinstance(node.get("tabs"), dict) else None
|
||
if not isinstance(ts, list) or not ts or any(not (isinstance(x, dict) and isinstance(x.get("id"), str) and isinstance(x.get("label"), str)) for x in ts):
|
||
_err(errors, path, "tabs.tabs 非法")
|
||
else:
|
||
for x in ts:
|
||
for c in x.get("nodes") or []:
|
||
walk(c, path + f"/tab:{x['id']}", depth + 1)
|
||
elif t == "timeline":
|
||
items = (node.get("timeline") or {}).get("items")
|
||
if not isinstance(items, list) or not items or any(not (isinstance(i, dict) and isinstance(i.get("title"), str)) for i in items):
|
||
_err(errors, path, "timeline.items 非法")
|
||
elif t == "section":
|
||
for c in node.get("children") or []:
|
||
walk(c, path + "/child", depth + 1)
|
||
elif t == "compare":
|
||
cp = node.get("compare")
|
||
if not isinstance(cp, dict) or not isinstance(cp.get("left"), dict) or not isinstance(cp.get("right"), dict):
|
||
_err(errors, path, "compare 需 left/right")
|
||
else:
|
||
for side in ("left", "right"):
|
||
sd = cp[side]
|
||
if sd.get("tone", "") not in TONES:
|
||
_err(errors, path, f"compare.{side}.tone 非法")
|
||
for c in sd.get("nodes") or []:
|
||
walk(c, path + f"/{side}", depth + 1)
|
||
|
||
if not isinstance(spec, dict) or not isinstance(spec.get("title"), str) or not isinstance(spec.get("nodes"), list):
|
||
return ["顶层必须为 {title:string, nodes:[]}"], stats
|
||
if len(json.dumps(spec, ensure_ascii=False).encode()) > MAX_BYTES:
|
||
_err(errors, "$", "规格超过 256KiB")
|
||
for n in spec["nodes"]:
|
||
walk(n, "$", 1)
|
||
if stats["nodes"] > MAX_NODES:
|
||
_err(errors, "$", f"节点数 {stats['nodes']} 超过 200")
|
||
if stats["maxDepth"] > MAX_DEPTH:
|
||
_err(errors, "$", f"嵌套深度 {stats['maxDepth']} 超过 8")
|
||
if stats["unknownRefs"]:
|
||
_err(errors, "$", f"引用了不存在的 sourceId: {sorted(stats['unknownRefs'])}")
|
||
return errors, stats
|
||
|
||
|
||
def build_user_message(task):
|
||
mats = "\n\n".join(f"【{m['sourceId']}】\n{m['text']}" for m in task["materials"])
|
||
return f"材料（sourceId 如【】所示）：\n\n{mats}\n\n用户目标：{task['goal']}"
|