#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import math
from html import escape
from pathlib import Path
from typing import Any
from bayesian_decision_report import build_report, load_request, localize_text
ROOT = Path(__file__).resolve().parent.parent
CSS_PATH = ROOT / "templates" / "report-theme.css"
TOP_NAV = [
("summary", "先看结论", "Summary"),
("action-plan", "行动建议", "Action"),
("why", "为什么这样判断", "Why"),
("hygiene", "先验检查", "Prior checks"),
("conversation", "对话过程", "Conversation"),
("decision", "问题定义", "Decision"),
("evidence", "证据", "Evidence"),
("update", "概率判断", "Update"),
("prior", "先验", "Prior"),
("sensitivity", "敏感性", "Sensitivity"),
("warnings", "注意事项", "Warnings"),
("workflow", "附录", "Appendix"),
]
SKILL_FLOW = [
{"zh": "问题结构化", "en": "Structure the decision"},
{"zh": "基于当前信息给初步先验", "en": "Set an initial prior from the current information"},
{"zh": "通过多轮追问补齐关键缺口", "en": "Close the key gaps through iterative questioning"},
{"zh": "检索与分级证据", "en": "Collect and grade evidence"},
{"zh": "构造先验与参考类", "en": "Build priors and reference classes"},
{"zh": "执行贝叶斯更新", "en": "Run the Bayesian update"},
{"zh": "做敏感性与阈值分析", "en": "Stress-test thresholds and sensitivity"},
{"zh": "给出行动与下一步信息建议", "en": "Recommend the next action and information step"},
{"zh": "记录每轮变化并输出过程日志", "en": "Record each round and export a process log"},
]
SKILL_CAPABILITIES = [
{
"zh_title": "问题定义与参考类选择",
"zh_body": "把模糊问题压缩成可计算假设、时间范围、成功标准和行动集合,并显式选择参考类。",
"en_title": "Problem framing and reference classes",
"en_body": "Turns a fuzzy choice into a computable hypothesis, time horizon, success metric, and action set with an explicit reference class.",
},
{
"zh_title": "证据分级与先验构造",
"zh_body": "区分强证据、弱证据和类比信号,输出带来源、可信度和等效样本量的先验。",
"en_title": "Evidence grading and prior construction",
"en_body": "Separates strong evidence from weak signals and produces priors with source notes, quality, and equivalent sample size.",
},
{
"zh_title": "赔率更新与共轭更新",
"zh_body": "支持似然比赔率更新、Beta-binomial 计数更新,以及二者的组合路径。",
"en_title": "Odds updates and conjugate updates",
"en_body": "Supports likelihood-ratio odds updates, Beta-binomial count updates, and the combined path when both are needed.",
},
{
"zh_title": "行动阈值与期望值比较",
"zh_body": "把后验概率映射到行动阈值和期望值排序,避免只停留在“概率高低”的描述。",
"en_title": "Action thresholds and expected value",
"en_body": "Converts posterior beliefs into action thresholds and expected-value ranking instead of stopping at raw probability.",
},
{
"zh_title": "敏感性分析与防幻觉提醒",
"zh_body": "用不同先验、证据强度和依赖折扣重跑结果,显式标记结论是否稳健。",
"en_title": "Sensitivity analysis and fragility checks",
"en_body": "Re-runs the decision under alternative priors, evidence strength, and dependency discounts to show whether the conclusion is robust.",
},
{
"zh_title": "自动化 HTML / Markdown 报告",
"zh_body": "从同一个结构化输入自动生成 Markdown 与双语 HTML,并在 HTML 中支持打印和存储为 PDF。",
"en_title": "Automated HTML / Markdown reports",
"en_body": "Automatically generates synchronized Markdown and bilingual HTML reports from the same structured input, with print and save-as-PDF support inside HTML.",
},
{
"zh_title": "多轮对话决策循环",
"zh_body": "先基于当前信息给弱先验和初步判断,再通过多轮对话追问缺失信息,持续更新先验、后验和决策准备度。",
"en_title": "Multi-turn decision loop",
"en_body": "Starts with a weak prior and an initial read, then iteratively asks for missing information and updates the prior, posterior, and decision readiness over multiple turns.",
},
{
"zh_title": "过程日志与变化图表",
"zh_body": "自动记录每一轮对话里新增了什么信息、如何完成贝叶斯更新、判断为什么改变,并在报告中输出轨迹图和过程表。",
"en_title": "Process log and change chart",
"en_body": "Records what changed in each round, how the Bayesian update was applied, why the judgment moved, and renders both a trajectory chart and a process table in the report.",
},
]
WARNING_TRANSLATIONS = {
"未提供先验等效样本量,先验强度相关的敏感性判断会更弱。": "No prior equivalent sample size was provided, so sensitivity about prior strength is weaker.",
"在 Beta-binomial 更新之后又追加了证据更新,请确认这些证据没有被重复包含在已观测计数中。": "Additional evidence was applied after the Beta-binomial update. Confirm that these signals are not already embedded in the observed counts.",
"部分证据较弱或较间接,后验判断应视为更脆弱。": "Some evidence is weak or indirect, so the posterior judgment should be treated as more fragile.",
"至少有一条证据因依赖关系被折扣处理,不应把这次更新理解为完全独立证据的简单叠加。": "At least one evidence item was discounted for dependence, so this update should not be read as a simple stack of fully independent signals.",
"本报告仅用于辅助决策,不替代持证专业人士的正式判断。": "This report is decision support only and does not replace judgment from a licensed professional.",
"这里的后验区间来自赔率更新下的敏感性包络,并非严格统计意义上的可信区间。": "The posterior interval here is a sensitivity envelope from odds updates, not a strict statistical credible interval.",
}
CONFIDENCE_EN = {
"low": "Low",
"medium": "Medium",
"medium-high": "Medium-high",
}
STABILITY_EN = {
"stable": "Stable",
"mixed": "Mixed",
"unstable": "Unstable",
}
READINESS_EN = {
"可以决策": "Ready to decide",
"接近可决策": "Nearly ready",
"仍需补信息": "Need more information",
"未评估": "Not evaluated",
}
DECISION_STATUS_EN = {
"ready": "Ready to decide",
"needs-more-info": "Need more information",
"in-progress": "Keep iterating",
"blocked": "Blocked",
}
DIRECTION_EN = {
"support": "Supports",
"against": "Weakens",
}
DECISION_STATES = {
"proceed": {
"label_zh": "可以推进",
"label_en": "Proceed",
"tone_zh": "当前判断偏向正面,可以往前走,但仍建议按推荐节奏推进。",
"tone_en": "The current read is positive enough to move forward, but you should still follow the recommended pace.",
},
"confirm": {
"label_zh": "先确认,再推进",
"label_en": "Confirm first",
"tone_zh": "现在不是不能做,而是还差最后一两个关键信息,先确认再推进更稳。",
"tone_en": "This is not a no. It means one or two key inputs are still missing, so confirm them before moving forward.",
},
"hold": {
"label_zh": "暂时别做",
"label_en": "Hold off",
"tone_zh": "当前把握还不够,先暂缓更稳,等拿到更强证据再决定。",
"tone_en": "The current case is not strong enough, so holding off is safer until stronger evidence appears.",
},
}
def request_text(value: Any, lang: str) -> str:
text = localize_text(value, lang)
return text.strip() if isinstance(text, str) and text.strip() else "-"
def localize_items(values: list[Any] | None, lang: str) -> list[str]:
items = []
for value in values or []:
text = request_text(value, lang)
if text != "-":
items.append(text)
return items
def fmt_pct(value: Any) -> str:
if value is None:
return "-"
return f"{float(value) * 100:.1f}%"
def fmt_num(value: Any) -> str:
if value is None:
return "-"
number = float(value)
if abs(number - round(number)) < 1e-9:
return f"{int(round(number)):,}"
return f"{number:,.1f}"
def fmt_interval(interval: list[float] | None) -> str:
if not interval:
return "-"
return f"{fmt_pct(interval[0])} - {fmt_pct(interval[1])}"
def html_text(value: Any) -> str:
return escape(str(value))
def dual_html(zh: str, en: str) -> str:
zh_text = html_text(zh or "-")
en_text = html_text(en or zh or "-")
return f'{zh_text}{en_text}'
def dual_tag(tag: str, zh: str, en: str, class_name: str = "") -> str:
attrs = f' class="{class_name}"' if class_name else ""
return f"<{tag}{attrs}>{dual_html(zh, en)}{tag}>"
def fold_section(section_id: str, zh_title: str, en_title: str, body_html: str, kicker_zh: str = "默认折叠", kicker_en: str = "Collapsed by default") -> str:
return f"""
{dual_html(kicker_zh, kicker_en)}
{dual_html(zh_title, en_title)}
{dual_html("展开", "Open")}
{body_html}
"""
def action_name_maps(request: dict) -> tuple[dict[str, str], dict[str, str]]:
zh_to_en: dict[str, str] = {}
en_to_zh: dict[str, str] = {}
for action in request.get("actions", []):
zh_name = request_text(action.get("name"), "zh")
en_name = request_text(action.get("name"), "en")
zh_to_en[zh_name] = en_name
en_to_zh[en_name] = zh_name
return zh_to_en, en_to_zh
def recommended_action_en(request: dict, recommended_zh: str) -> str:
zh_to_en, _ = action_name_maps(request)
return zh_to_en.get(recommended_zh, recommended_zh)
def summary_en(request: dict, report: dict) -> str:
summary = report["summary"]
action_en = recommended_action_en(request, summary["recommendation"])
return f"The current posterior probability is about {fmt_pct(summary['posterior_probability'])}; given the available evidence, the better action is {action_en}."
def prior_confidence_en(request: dict, report: dict) -> str:
raw = request.get("prior", {}).get("confidence")
text = request_text(raw, "en")
if text != "-":
return text.replace("-", " ").title()
code = report["summary"].get("confidence_code")
return CONFIDENCE_EN.get(code, code or "-")
def stability_en(report: dict) -> str:
code = report["sensitivity"].get("conclusion_stability_code")
return STABILITY_EN.get(code, code or "-")
def direction_en(item: dict) -> str:
code = item.get("direction_code")
return DIRECTION_EN.get(code, code or "-")
def warning_en(text: str) -> str:
return WARNING_TRANSLATIONS.get(text, text)
def natural_frequency_en(report: dict) -> str:
natural = report["natural_frequency"]
base = natural["base_population"]
prior_cases = natural["prior_expected_successes"]
posterior_cases = natural["posterior_expected_successes"]
return (
f"Out of {base} similar cases, the prior suggests about {prior_cases} would succeed. "
f"After the current evidence, about {posterior_cases} would succeed."
)
def next_information_reason_en(report: dict) -> str:
if report["next_information"].get("high_value"):
return "The current result is close to an action threshold, so one more strong signal could flip the recommendation."
return "More information still helps, but the current recommendation is less sensitive to boundary changes."
def first_next_information(request: dict, lang: str) -> str:
candidates = (request.get("next_information") or {}).get("candidates") or []
if not candidates:
return "-"
return request_text(candidates[0], lang)
def decision_reason_en(report: dict) -> str:
if report["decision"]["expected_value_ranking"]:
return "Expected-value comparison based on the current posterior places this action first."
return "No action-utility inputs were provided, so the skill cannot rank actions directly."
def conversation_raw_rounds(request: dict) -> list[dict]:
return ((request.get("conversation") or {}).get("rounds") or [])
def conversation_round_en(raw_round: dict, key: str, default: str = "-") -> str:
value = request_text(raw_round.get(key), "en")
return value if value != "-" else default
def readiness_en(label_zh: str) -> str:
return READINESS_EN.get(label_zh, label_zh)
def decision_status_en(report: dict) -> str:
process = report.get("conversation_process") or {}
code = process.get("status_code")
return DECISION_STATUS_EN.get(code, process.get("status_en") or code or "-")
def conversation_dual_list_items(round_payload: dict, raw_round: dict, primary_key: str, fallback_key: str | None = None) -> str:
zh_items = list(round_payload.get(primary_key) or [])
if not zh_items and fallback_key:
zh_items = list(round_payload.get(fallback_key) or [])
raw_items = list(raw_round.get(primary_key) or [])
if not raw_items and fallback_key:
raw_items = list(raw_round.get(fallback_key) or [])
count = max(len(zh_items), len(raw_items))
if count == 0:
return ""
items = []
for index in range(count):
zh_value = zh_items[index] if index < len(zh_items) else "-"
en_source = raw_items[index] if index < len(raw_items) else zh_value
items.append(f"
")
return "".join(items)
def conversation_chart_svg(report: dict) -> str:
process = report.get("conversation_process") or {}
points = process.get("trajectory") or []
if not points:
return ""
width = 760
height = 260
margin_left = 48
margin_right = 20
margin_top = 20
margin_bottom = 42
inner_width = width - margin_left - margin_right
inner_height = height - margin_top - margin_bottom
count = len(points)
step = inner_width / max(count - 1, 1)
def x_at(index: int) -> float:
return margin_left + step * index
def y_at(value: float | None) -> float:
safe = 0.0 if value is None else max(0.0, min(1.0, float(value)))
return margin_top + (1.0 - safe) * inner_height
def polyline(values: list[float | None]) -> str:
coords = [f"{x_at(index):.1f},{y_at(value):.1f}" for index, value in enumerate(values) if value is not None]
return " ".join(coords)
grid_lines = []
for ratio in (0.0, 0.25, 0.5, 0.75, 1.0):
y = y_at(ratio)
grid_lines.append(f'')
grid_lines.append(
f'{ratio * 100:.0f}%'
)
prior_values = [item.get("prior_probability") for item in points]
posterior_values = [item.get("posterior_probability") for item in points]
readiness_values = [item.get("decision_readiness") for item in points]
readiness_bars = []
point_markers = []
for index, point in enumerate(points):
x = x_at(index)
readiness = point.get("decision_readiness")
if readiness is not None:
y = y_at(readiness)
height_value = margin_top + inner_height - y
readiness_bars.append(
f''
)
for value, class_name in (
(point.get("prior_probability"), "chart-point-prior"),
(point.get("posterior_probability"), "chart-point-posterior"),
):
if value is not None:
point_markers.append(f'')
grid_lines.append(
f'R{point["round"]}'
)
return f"""
"""
def confidence_phrase_zh(code: str | None) -> str:
return {
"low": "把握偏低",
"medium": "把握一般",
"medium-high": "把握较高",
}.get(code, code or "把握一般")
def confidence_phrase_en(code: str | None) -> str:
return {
"low": "lower confidence",
"medium": "moderate confidence",
"medium-high": "stronger confidence",
}.get(code, code or "moderate confidence")
def stability_phrase_zh(code: str | None) -> str:
return {
"stable": "换一组合理假设,结论大体不变",
"mixed": "换一组合理假设,结论可能有轻微摆动",
"unstable": "换一组合理假设,结论可能明显变化",
}.get(code, code or "结论稳定性一般")
def stability_phrase_en(code: str | None) -> str:
return {
"stable": "The conclusion mostly holds under reasonable alternative assumptions.",
"mixed": "The conclusion shifts a bit under reasonable alternative assumptions.",
"unstable": "The conclusion can change materially under reasonable alternative assumptions.",
}.get(code, code or "The conclusion has moderate stability.")
def classify_decision_state(report: dict) -> str:
recommendation = report["decision"]["recommended_action"]
if any(token in recommendation for token in ("暂停", "暂缓", "放弃", "延后", "改到")):
return "hold"
if any(token in recommendation for token in ("先", "确认", "测试", "试点", "验证", "再订")):
return "confirm"
return "proceed"
def evidence_ranked_pairs(request: dict, report: dict) -> list[tuple[float, dict, dict]]:
pairs = []
for zh_item, source_item in zip(report["evidence"], request.get("evidence", [])):
lr = max(float(zh_item.get("likelihood_ratio") or 1.0), 1e-6)
discount = max(float(zh_item.get("dependency_discount") or 1.0), 0.0)
strength = abs(math.log(lr)) * discount
pairs.append((strength, zh_item, source_item))
return sorted(pairs, key=lambda item: item[0], reverse=True)
def top_reason_cards(request: dict, report: dict, direction_code: str, limit: int = 2) -> list[dict[str, str]]:
cards: list[dict[str, str]] = []
for _, zh_item, source_item in evidence_ranked_pairs(request, report):
if zh_item.get("direction_code") != direction_code:
continue
cards.append(
{
"title_zh": zh_item["name"],
"title_en": request_text(source_item.get("name"), "en"),
"body_zh": zh_item.get("summary") or zh_item["name"],
"body_en": request_text(source_item.get("summary"), "en") or request_text(source_item.get("name"), "en"),
}
)
if len(cards) >= limit:
break
return cards
def alternative_cards(request: dict, report: dict) -> list[dict[str, str]]:
ranking = report["decision"]["expected_value_ranking"]
if len(ranking) <= 1:
return []
top_value = float(ranking[0]["expected_value"])
cards = []
for item in ranking[1:3]:
gap = top_value - float(item["expected_value"])
cards.append(
{
"name_zh": item["name"],
"name_en": recommended_action_en(request, item["name"]),
"reason_zh": f"它目前的综合收益比首选方案低 {fmt_num(gap)},所以暂时排在后面。",
"reason_en": f"It currently trails the recommended option by {fmt_num(gap)} in expected value, so it stays behind for now.",
}
)
return cards
def plain_language_pack(request: dict, report: dict) -> dict[str, Any]:
state = classify_decision_state(report)
state_copy = DECISION_STATES[state]
confidence_code = report["summary"].get("confidence_code")
stability_code = report["sensitivity"].get("conclusion_stability_code")
process = report.get("conversation_process") or {}
deadline = report["question"].get("decision_deadline")
success_metric = report["question"].get("success_metric") or "-"
time_horizon = report["question"].get("time_horizon") or "-"
next_info_zh = report["next_information"]["recommended_next_information"]
next_info_en = first_next_information(request, "en")
recommendation_zh = report["decision"]["recommended_action"]
recommendation_en = recommended_action_en(request, recommendation_zh)
if state == "proceed":
step3_zh = (
f"如果到 {deadline} 前没有出现新的明显负面信息,就按这个方案执行。"
if deadline and deadline != "-"
else "如果关键条件都确认无误,就按这个方案执行。"
)
step3_en = (
f"If no major new negative signal shows up before {deadline}, execute this plan."
if deadline and deadline != "-"
else "If the key conditions check out, go ahead and execute this plan."
)
elif state == "confirm":
step3_zh = (
f"如果到 {deadline} 前关键条件确认无误,就推进;如果确认不了,就先不要做更重的动作。"
if deadline and deadline != "-"
else "如果关键条件确认无误,就推进;如果确认不了,就先不要做更重的动作。"
)
step3_en = (
f"If the key conditions are confirmed before {deadline}, move ahead; if not, do not escalate to a heavier commitment yet."
if deadline and deadline != "-"
else "If the key conditions are confirmed, move ahead; if not, do not escalate to a heavier commitment yet."
)
else:
step3_zh = "除非拿到新的强证据,否则先不要投入更多时间或预算。"
step3_en = "Do not commit more time or budget unless stronger evidence appears."
summary_zh = (
f"{state_copy['tone_zh']} 目前的建议属于{confidence_phrase_zh(confidence_code)},而且{stability_phrase_zh(stability_code)}。"
)
summary_en = (
f"{state_copy['tone_en']} The current recommendation carries {confidence_phrase_en(confidence_code)}, and {stability_phrase_en(stability_code)}"
)
if process:
gate_zh = (
f"多轮对话后的决策准备度约为 {fmt_pct(process.get('final_readiness'))},"
f"{'已经' if process.get('decision_ready') else '还没有'}达到可决策状态。"
)
gate_en = (
f"After the multi-turn loop, decision readiness is about {fmt_pct(process.get('final_readiness'))}, "
f"and the case has {'reached' if process.get('decision_ready') else 'not yet reached'} a ready-to-decide state."
)
else:
gate_zh = "这份判断基于当前一次性输入,没有额外的多轮过程记录。"
gate_en = "This judgment is based on the current single-pass input and does not include a multi-turn process log."
return {
"state": state,
"label_zh": state_copy["label_zh"],
"label_en": state_copy["label_en"],
"summary_zh": summary_zh,
"summary_en": summary_en,
"recommendation_zh": recommendation_zh,
"recommendation_en": recommendation_en,
"action_steps_zh": [
recommendation_zh,
next_info_zh,
step3_zh,
],
"action_steps_en": [
recommendation_en,
next_info_en,
step3_en,
],
"premises_zh": [
f"时间窗口:{time_horizon}",
f"成立条件:{success_metric}",
f"最晚决策时间:{deadline}" if deadline and deadline != "-" else "最晚决策时间:尽快确认关键条件",
],
"premises_en": [
f"Time window: {request_text(request.get('time_horizon'), 'en')}",
f"Condition for acting: {request_text(request.get('success_metric'), 'en')}",
f"Decision deadline: {deadline}" if deadline and deadline != "-" else "Decision deadline: confirm the key conditions soon",
],
"support_cards": top_reason_cards(request, report, "support"),
"risk_cards": top_reason_cards(request, report, "against"),
"alternatives": alternative_cards(request, report),
"decision_gate_zh": gate_zh,
"decision_gate_en": gate_en,
}
def build_markdown(request: dict, report: dict) -> str:
plain = plain_language_pack(request, report)
process = report.get("conversation_process") or {}
lines = [
f"# 贝叶斯决策报告:{report['title']}",
"",
"> 本报告由 `yao-bayesian-skill` 基于同一结构化输入自动生成;Markdown 与 HTML 版本保持同步。",
"",
"## 1. 先说结论",
f"- 结论标签:{plain['label_zh']}",
f"- 一句话判断:{report['summary']['one_sentence']}",
f"- 人话解释:{plain['summary_zh']}",
"",
"## 2. 你现在该怎么做",
f"1. {plain['action_steps_zh'][0]}",
f"2. {plain['action_steps_zh'][1]}",
f"3. {plain['action_steps_zh'][2]}",
"",
"## 3. 这份建议成立的前提",
*[f"- {item}" for item in plain["premises_zh"]],
"",
"## 4. 为什么不是另外两个选项",
]
if plain["alternatives"]:
lines.extend([f"- {item['name_zh']}:{item['reason_zh']}" for item in plain["alternatives"]])
else:
lines.append("- 当前没有足够的备选行动用于比较。")
hygiene = report.get("prior_hygiene") or {}
hygiene_checks = hygiene.get("checks") or []
lines.extend(
[
"",
"## 5. 贝叶斯先验检查",
f"- 选择规则:{hygiene.get('selection_rule') or '从默认先验检查表中选择本次最相关的项目。'}",
]
)
if hygiene_checks:
for item in hygiene_checks:
lines.extend(
[
f"- {item['principle']}:{item['core_sentence']}",
f" - 触发原因:{item['trigger']}",
f" - 对决策的影响:{item['decision_use']}",
]
)
else:
lines.append("- 当前没有单独触发的先验检查项。")
if process:
lines.extend(
[
"",
"## 6. 多轮对话过程与决策准备度",
f"- 初始现状:{process.get('initial_state') or '-'}",
f"- 对话轮次:{process.get('round_count')}",
f"- 当前状态:{process.get('status')}",
f"- 决策准备度:{fmt_pct(process.get('final_readiness'))}",
f"- 决策阈值:{fmt_pct(process.get('decision_ready_threshold'))}",
f"- 是否可以进入正式决策:{'可以' if process.get('decision_ready') else '还不可以'}",
f"- 过程分析:{process.get('analysis')}",
"",
"| 轮次 | 阶段 | 起点概率 | 更新后概率 | 概率变化 | 决策准备度 | 中间判断 |",
"|---:|---|---:|---:|---:|---:|---|",
]
)
for item in process.get("rounds", []):
lines.append(
f"| {item['round']} | {item['stage']} | {fmt_pct(item.get('prior_probability_before'))} | {fmt_pct(item.get('posterior_probability_after'))} | {fmt_pct(item.get('delta_probability'))} | {fmt_pct(item.get('decision_readiness'))} | {item.get('interim_judgment') or '-'} |"
)
if process.get("open_questions"):
lines.extend(
[
"",
"- 当前仍需关注的问题:",
*[f" - {item}" for item in process["open_questions"]],
]
)
lines.extend(
[
"",
"- 贝叶斯过程日志:",
]
)
for item in process.get("rounds", []):
formula = (item.get("bayes_update") or {}).get("formula_zh")
lines.append(f" - 第 {item['round']} 轮:{formula or '本轮只记录了结果,没有完整公式参数。'}")
lines.extend(
[
"",
"## 7. 决策问题",
f"- 决策问题:{report['question']['decision_question']}",
f"- 要判断的假设:{report['question']['hypothesis']}",
f"- 时间范围:{report['question']['time_horizon']}",
f"- 决策截止时间:{report['question']['decision_deadline']}",
f"- 成功标准:{report['question']['success_metric']}",
f"- 领域:{report['question']['domain']}",
"",
"## 8. 先验设置",
f"- 先验概率:{fmt_pct(report['prior']['probability'])}",
f"- 先验分布:{report['prior']['distribution']}",
f"- 先验区间:{fmt_interval(report['prior']['credible_interval'])}",
f"- 先验来源等级:{report['prior']['source_quality']}",
f"- 先验信心:{report['prior']['confidence']}",
f"- 等效样本量:{fmt_num(report['prior']['equivalent_sample_size'])}",
"",
]
)
if report["prior"]["source_summary"]:
lines.append("- 先验来源:")
lines.extend([f" - {item}" for item in report["prior"]["source_summary"]])
lines.append("")
lines.extend(
[
"## 9. 证据摘要",
"| 证据 | 可信度 | 方向 | 似然比 | 依赖折扣 | 摘要 |",
"|---|---|---|---:|---:|---|",
]
)
for item in report["evidence"]:
lines.append(
f"| {item['name']} | {item['quality']} | {item['direction']} | {fmt_num(item['likelihood_ratio'])} | {fmt_num(item['dependency_discount'])} | {item['summary'] or '-'} |"
)
lines.extend(
[
"",
"## 10. 贝叶斯更新",
f"- 更新方法:{report['posterior']['method']}",
f"- 后验概率:{fmt_pct(report['posterior']['probability'])}",
f"- 后验区间:{fmt_interval(report['posterior']['credible_interval'])}",
f"- 自然频率解释:{report['natural_frequency']['explanation']}",
"",
"## 11. 行动比较",
f"- 推荐行动:{report['decision']['recommended_action']}",
f"- 推荐理由:{report['decision']['reason']}",
f"- 决策状态:{report['decision'].get('decision_status') or '未单独评估'}",
f"- 决策准备度:{fmt_pct(report['decision'].get('decision_readiness'))}",
"",
"| 行动 | H 为真时效用 | H 为假时效用 | 期望值 | 行动阈值 |",
"|---|---:|---:|---:|---:|",
]
)
for action in report["decision"]["expected_value_ranking"]:
lines.append(
f"| {action['name']} | {fmt_num(action['utility_if_h_true'])} | {fmt_num(action['utility_if_h_false'])} | {fmt_num(action['expected_value'])} | {fmt_pct(action['action_threshold'])} |"
)
lines.extend(
[
"",
"## 12. 敏感性分析",
f"- 后验范围:{fmt_interval(report['sensitivity']['posterior_range'])}",
f"- 结论稳定性:{report['sensitivity']['conclusion_stability']}",
"",
"| 先验概率 | LR 强度系数 | 后验概率 | 推荐行动 |",
"|---:|---:|---:|---|",
]
)
for scenario in report["sensitivity"]["scenarios"]:
lines.append(
f"| {fmt_pct(scenario.get('prior_probability'))} | {fmt_num(scenario.get('lr_power'))} | {fmt_pct(scenario.get('posterior_probability'))} | {scenario.get('recommended_action') or '-'} |"
)
lines.extend(
[
"",
"## 13. 下一步最有价值的信息",
f"- 推荐下一步信息:{report['next_information']['recommended_next_information']}",
f"- 判断原因:{report['next_information']['reason']}",
"",
"## 14. 风险与注意事项",
]
)
if report["warnings"]:
lines.extend([f"- {warning}" for warning in report["warnings"]])
else:
lines.append("- 当前没有额外警示项。")
lines.extend(
[
"",
"## 15. Skill 流程",
]
)
for index, item in enumerate(SKILL_FLOW, start=1):
lines.append(f"- {index}. {item['zh']}")
lines.extend(
[
"",
"## 16. Skill 能力",
]
)
for capability in SKILL_CAPABILITIES:
lines.append(f"- {capability['zh_title']}:{capability['zh_body']}")
lines.extend(
[
"",
"## 17. 自动生成说明",
"- 本报告不是手写示例,而是由同一个结构化输入自动渲染出来的正式输出。",
"- HTML 提供中英双语一键切换,并支持直接打印或在浏览器打印面板里存储为 PDF。",
f"- {plain['decision_gate_zh']}",
]
)
return "\n".join(lines).strip() + "\n"
def build_html(request: dict, report: dict) -> str:
css = CSS_PATH.read_text(encoding="utf-8")
plain = plain_language_pack(request, report)
process = report.get("conversation_process") or {}
raw_rounds = conversation_raw_rounds(request)
title_zh = report["title"]
title_en = request_text(request.get("title"), "en")
decision_zh = report["question"]["decision_question"]
decision_en = request_text(request.get("question"), "en")
hypothesis_zh = report["question"]["hypothesis"]
hypothesis_en = request_text(request.get("hypothesis"), "en")
horizon_zh = report["question"]["time_horizon"]
horizon_en = request_text(request.get("time_horizon"), "en")
success_zh = report["question"]["success_metric"]
success_en = request_text(request.get("success_metric"), "en")
domain_zh = report["question"]["domain"]
domain_en = request_text(request.get("domain"), "en")
prior_sources_zh = report["prior"]["source_summary"]
prior_sources_en = localize_items(request.get("prior", {}).get("source_summary"), "en")
recommended_zh = report["decision"]["recommended_action"]
recommended_en = recommended_action_en(request, recommended_zh)
ranking = report["decision"]["expected_value_ranking"]
top_threshold = ranking[0].get("action_threshold") if ranking else None
support_cards_html = "".join(
"""
{label}
{title}
{body}
""".format(
label=dual_html("支持判断", "Supports the call"),
title=dual_html(item["title_zh"], item["title_en"]),
body=dual_html(item["body_zh"], item["body_en"]),
)
for item in plain["support_cards"]
) or """
支持判断Supports the call
当前没有足够的正向证据卡片。There are not enough positive evidence cards at the moment.
"""
risk_cards_html = "".join(
"""
{label}
{title}
{body}
""".format(
label=dual_html("拉低判断", "Pulls the call down"),
title=dual_html(item["title_zh"], item["title_en"]),
body=dual_html(item["body_zh"], item["body_en"]),
)
for item in plain["risk_cards"]
) or """
拉低判断Pulls the call down
当前没有足够的负向证据卡片。There are not enough negative evidence cards at the moment.
"""
alternative_cards_html = "".join(
"""
{name}{reason}
""".format(
name=dual_html(item["name_zh"], item["name_en"]),
reason=dual_html(item["reason_zh"], item["reason_en"]),
)
for item in plain["alternatives"]
) or """
当前没有足够的备选动作可比较。There are not enough alternate actions to compare right now.
"""
hygiene = report.get("prior_hygiene") or {}
hygiene_cards_html = "".join(
"""
{score}{label}
{title}
{core}
{trigger_label} {trigger}
{decision_label} {decision_use}
""".format(
score=html_text(f"{item.get('score', '-')}/25"),
label=dual_html("本次触发", "Triggered"),
title=dual_html(item.get("principle") or "-", item.get("principle_en") or "-"),
core=dual_html(item.get("core_sentence") or "-", item.get("core_sentence_en") or "-"),
trigger_label=dual_html("触发原因:", "Why it matters:"),
trigger=dual_html(item.get("trigger") or "-", item.get("trigger_en") or "-"),
decision_label=dual_html("对决策的影响:", "Decision effect:"),
decision_use=dual_html(item.get("decision_use") or "-", item.get("decision_use_en") or "-"),
)
for item in hygiene.get("checks", [])
) or f"""
{dual_html("没有单独触发的先验检查", "No dedicated prior checks triggered")}
{dual_html("当前报告没有识别出需要单独展示的贝叶斯先验原则。", "The report did not identify prior hygiene checks that need a dedicated callout.")}
"
for index in range(len(plain["premises_zh"]))
)
process_badge_html = ""
if process:
process_badge_html = f"""
{dual_html(process.get("status") or "-", decision_status_en(report))}{dual_html(f"对话 {process.get('round_count')} 轮", f"{process.get('round_count')} rounds")}
"""
evidence_rows = []
for zh_item, source_item in zip(report["evidence"], request.get("evidence", [])):
evidence_rows.append(
"""
{name}
{quality}
{direction}
{lr}
{discount}
{summary}
{notes}
""".format(
name=dual_html(zh_item["name"], request_text(source_item.get("name"), "en")),
quality=html_text(zh_item["quality"]),
direction=dual_html(zh_item["direction"], direction_en(zh_item)),
lr=html_text(fmt_num(zh_item["likelihood_ratio"])),
discount=html_text(fmt_num(zh_item["dependency_discount"])),
summary=dual_html(zh_item.get("summary") or "-", request_text(source_item.get("summary"), "en")),
notes=dual_html(zh_item.get("notes") or "-", request_text(source_item.get("notes"), "en")),
)
)
action_rows = []
for item in ranking:
action_rows.append(
"""
{name}
{utility_true}
{utility_false}
{expected_value}
{threshold}
""".format(
name=dual_html(item["name"], recommended_action_en(request, item["name"])),
utility_true=html_text(fmt_num(item["utility_if_h_true"])),
utility_false=html_text(fmt_num(item["utility_if_h_false"])),
expected_value=html_text(fmt_num(item["expected_value"])),
threshold=html_text(fmt_pct(item["action_threshold"])),
)
)
sensitivity_rows = []
for item in report["sensitivity"]["scenarios"]:
sensitivity_rows.append(
"""
{prior}
{lr_power}
{posterior}
{action}
""".format(
prior=html_text(fmt_pct(item.get("prior_probability"))),
lr_power=html_text(fmt_num(item.get("lr_power"))),
posterior=html_text(fmt_pct(item.get("posterior_probability"))),
action=dual_html(item.get("recommended_action") or "-", recommended_action_en(request, item.get("recommended_action") or "-")),
)
)
prior_source_items = "".join(
f"
{dual_html(prior_sources_zh[index], prior_sources_en[index] if index < len(prior_sources_en) else prior_sources_zh[index])}
"
for index in range(len(prior_sources_zh))
)
warning_items = "".join(f"
{dual_html(item, warning_en(item))}
" for item in report["warnings"]) or f"
{dual_html('当前没有额外警示项。', 'There are no additional warnings at the moment.')}
"
workflow_cards = "".join(
"""
{index}
{label}
""".format(index=html_text(f"Step {index}"), label=dual_html(item["zh"], item["en"]))
for index, item in enumerate(SKILL_FLOW, start=1)
)
capability_cards = "".join(
"""
{title}
{body}
""".format(title=dual_html(item["zh_title"], item["en_title"]), body=dual_html(item["zh_body"], item["en_body"]))
for item in SKILL_CAPABILITIES
)
conversation_rounds_html = ""
conversation_section_html = ""
if process:
for index, item in enumerate(process.get("rounds", [])):
raw_round = raw_rounds[index] if index < len(raw_rounds) else {}
formula = item.get("bayes_update") or {}
conversation_rounds_html += """
{round_label}
{stage}
{readiness}
{prior_label}{prior_value}
{posterior_label}{posterior_value}
{delta_label}{delta_value}
{ready_label}{ready_value}
{user_label} {user_value}
{focus_label} {focus_value}
{judgment_label} {judgment_value}
{formula_label} {formula_value}
{new_info_label}
{new_info_items}
{next_q_label}
{next_q_items}
""".format(
round_label=dual_html(f"第 {item['round']} 轮", f"Round {item['round']}"),
stage=dual_html(item.get("stage") or "-", conversation_round_en(raw_round, "stage")),
readiness=dual_html(item.get("decision_readiness_label") or "-", readiness_en(item.get("decision_readiness_label") or "-")),
prior_label=dual_html("起点概率", "Starting belief"),
prior_value=html_text(fmt_pct(item.get("prior_probability_before"))),
posterior_label=dual_html("更新后概率", "Updated belief"),
posterior_value=html_text(fmt_pct(item.get("posterior_probability_after"))),
delta_label=dual_html("概率变化", "Change"),
delta_value=html_text(fmt_pct(item.get("delta_probability"))),
ready_label=dual_html("准备度", "Readiness"),
ready_value=html_text(fmt_pct(item.get("decision_readiness"))),
user_label=dual_html("用户补充了什么:", "What the user added:"),
user_value=dual_html(item.get("user_input_summary") or "-", conversation_round_en(raw_round, "user_input_summary")),
focus_label=dual_html("这一轮我在判断什么:", "What the skill focused on:"),
focus_value=dual_html(item.get("assistant_focus") or "-", conversation_round_en(raw_round, "assistant_focus")),
judgment_label=dual_html("中间判断:", "Interim judgment:"),
judgment_value=dual_html(item.get("interim_judgment") or "-", conversation_round_en(raw_round, "interim_judgment")),
formula_label=dual_html("贝叶斯变化:", "Bayesian change:"),
formula_value=dual_html(formula.get("formula_zh") or "-", formula.get("formula_en") or "-"),
new_info_label=dual_html("新增信息", "New information"),
new_info_items=conversation_dual_list_items(item, raw_round, "new_information")
or f"
{dual_html('本轮没有新增结构化信息。', 'No new structured information was recorded for this round.')}
{dual_html(hygiene.get("selection_rule") or "从 20 条生活贝叶斯先验中选择本次最相关的项目。", hygiene.get("selection_rule_en") or "From 20 everyday Bayesian priors, show only the checks most relevant to this decision.")}
{dual_html("这份报告由 skill 自动把问题、证据、行动和阈值对齐成统一决策格式。", "This report is auto-structured by the skill so the question, evidence, actions, and thresholds stay aligned.")}
{fold_section("conversation", "多轮对话过程", "Conversation process", conversation_section_html or f"
{dual_html('当前没有记录多轮过程;报告只展示最终结果。', 'There is no multi-turn process log for this report, so only the final result is shown.')}
{fold_section("sensitivity", "敏感性分析", "Sensitivity analysis", sensitivity_section_html)}
{dual_tag("h2", "风险与注意事项", "Warnings and caveats")}
{warning_items}
{dual_html("本 HTML 与 Markdown 共享同一套结构化计算结果;如需 PDF,请使用右上角“下载 PDF”。", "This HTML report shares the same structured result as Markdown. If you need a PDF, use the save-as-PDF action in the top-right corner.")}
{fold_section("workflow", "附录", "Appendix", appendix_section_html)}
"""
return html
def generate_bundle(input_path: Path, output_dir: Path, basename: str | None) -> dict[str, str]:
request = load_request(input_path)
report = build_report(request)
output_dir.mkdir(parents=True, exist_ok=True)
stem = basename or input_path.stem.replace("_", "-")
md_path = output_dir / f"{stem}.md"
html_path = output_dir / f"{stem}.html"
md_path.write_text(build_markdown(request, report), encoding="utf-8")
html_path.write_text(build_html(request, report), encoding="utf-8")
stale_paths = [
output_dir / f"{stem}.json",
output_dir / f"{stem}.pdf",
output_dir / f"{stem}.docx",
output_dir / f"{stem}.print.html",
output_dir / f"{stem}.browser.pdf",
]
for stale_path in stale_paths:
if stale_path.exists():
stale_path.unlink()
generated = {
"md": str(md_path),
"html": str(html_path),
}
return generated
def main() -> None:
parser = argparse.ArgumentParser(description="Generate a synchronized Bayesian decision report in Markdown and HTML.")
parser.add_argument("input_json", help="Path to the structured decision request JSON file.")
parser.add_argument("output_dir", help="Directory where the report files should be written.")
parser.add_argument("--basename", help="Optional basename for output files. Defaults to the input filename stem.")
args = parser.parse_args()
generated = generate_bundle(Path(args.input_json), Path(args.output_dir), args.basename)
print(json.dumps({"ok": True, "generated": generated}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()