148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
"""用于把模型生成的 Markdown 摘要做轻量解析并用更产品化的方式展示。
|
||
|
||
说明：不引入复杂依赖，不要求模型输出 JSON。
|
||
通过解析固定标题：
|
||
- # 一句话结论
|
||
- # 关键决策要点（Top 5）
|
||
- # 主要风险（Top 3）
|
||
- # 下一步行动清单
|
||
- # 需要进一步澄清的问题
|
||
|
||
另外提供：
|
||
- render_opinion_preview：观点一句话预览
|
||
- summarize_round_opinions：把一轮的多角色观点压缩为可扫读摘要（用于时间线）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import streamlit as st
|
||
|
||
|
||
SECTION_TITLES = [
|
||
"一句话结论",
|
||
"关键决策要点（Top 5）",
|
||
"主要风险（Top 3）",
|
||
"下一步行动清单",
|
||
"需要进一步澄清的问题",
|
||
]
|
||
|
||
|
||
def _split_sections(md: str) -> dict[str, str]:
|
||
if not md:
|
||
return {}
|
||
|
||
pattern = r"^#\s+(.*)$"
|
||
lines = md.splitlines()
|
||
sections: dict[str, list[str]] = {}
|
||
|
||
current = None
|
||
for line in lines:
|
||
m = re.match(pattern, line.strip())
|
||
if m:
|
||
title = m.group(1).strip()
|
||
current = title
|
||
sections.setdefault(current, [])
|
||
continue
|
||
if current is not None:
|
||
sections[current].append(line)
|
||
|
||
return {k: "\n".join(v).strip() for k, v in sections.items()}
|
||
|
||
|
||
def _extract_decision_badge(one_liner: str) -> tuple[str, str]:
|
||
t = (one_liner or "").strip()
|
||
if any(k in t for k in ["不建议", "否决", "不推荐"]):
|
||
return ("不建议", "error")
|
||
if any(k in t for k in ["谨慎", "有条件", "需要修改"]):
|
||
return ("谨慎推进", "warning")
|
||
if any(k in t for k in ["推荐", "可行", "建议推进"]):
|
||
return ("推荐", "success")
|
||
return ("结论", "info")
|
||
|
||
|
||
def render_decision_summary(md: str) -> None:
|
||
"""以更产品化的方式渲染结论摘要。"""
|
||
sections = _split_sections(md)
|
||
|
||
one_liner = sections.get("一句话结论", "").strip()
|
||
badge_text, badge_kind = _extract_decision_badge(one_liner)
|
||
|
||
col1, col2 = st.columns([1, 5])
|
||
with col1:
|
||
if badge_kind == "success":
|
||
st.success(badge_text)
|
||
elif badge_kind == "warning":
|
||
st.warning(badge_text)
|
||
elif badge_kind == "error":
|
||
st.error(badge_text)
|
||
else:
|
||
st.info(badge_text)
|
||
|
||
with col2:
|
||
if one_liner:
|
||
st.markdown(f"**{one_liner}**")
|
||
else:
|
||
st.markdown("**（未识别到一句话结论，请检查模型输出格式）**")
|
||
|
||
st.markdown("---")
|
||
|
||
c1, c2 = st.columns(2)
|
||
|
||
with c1:
|
||
st.subheader("🎯 关键决策要点")
|
||
content = sections.get("关键决策要点（Top 5）", "").strip()
|
||
st.markdown(content if content else "（暂无）")
|
||
|
||
st.subheader("✅ 下一步行动")
|
||
actions = sections.get("下一步行动清单", "").strip()
|
||
items = [
|
||
re.sub(r"^-\s*\[.\]\s*", "", l).strip()
|
||
for l in actions.splitlines()
|
||
if l.strip().startswith("-")
|
||
]
|
||
if items:
|
||
for i, it in enumerate(items):
|
||
st.checkbox(it, value=False, key=f"action_{i}")
|
||
else:
|
||
st.markdown(actions if actions else "（暂无）")
|
||
|
||
with c2:
|
||
st.subheader("⚠️ 主要风险")
|
||
risks = sections.get("主要风险（Top 3）", "").strip()
|
||
st.markdown(risks if risks else "（暂无）")
|
||
|
||
st.subheader("❓ 需要澄清")
|
||
qs = sections.get("需要进一步澄清的问题", "").strip()
|
||
st.markdown(qs if qs else "（暂无）")
|
||
|
||
|
||
def render_opinion_preview(opinion: str, limit: int = 160) -> str:
|
||
"""返回用于预览的短文本。"""
|
||
if not opinion:
|
||
return ""
|
||
t = opinion.strip().replace("\n", " ")
|
||
if len(t) <= limit:
|
||
return t
|
||
return t[:limit].rstrip() + "..."
|
||
|
||
|
||
def summarize_round_opinions(opinions: dict, role_descriptions: dict, limit_per_role: int = 60) -> str:
|
||
"""把一轮 opinions 生成适合扫读的摘要（不调用模型，避免额外成本与不稳定）。
|
||
|
||
输出格式示例：
|
||
- 📊 产品经理：xxx...
|
||
- 💻 技术专家：xxx...
|
||
"""
|
||
if not opinions:
|
||
return ""
|
||
|
||
lines: list[str] = []
|
||
for role, text in opinions.items():
|
||
info = role_descriptions.get(role, {"icon": "👤", "name": role})
|
||
preview = render_opinion_preview(str(text or ""), limit=limit_per_role)
|
||
if not preview:
|
||
preview = "（本轮未生成/生成失败）"
|
||
lines.append(f"- {info['icon']} {info['name']}：{preview}")
|
||
|
||
return "\n".join(lines)
|