Spaces:
Sleeping
Sleeping
File size: 5,048 Bytes
408f650 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """战略推理可视化：将每次战略决策的完整路径渲染为可读的文本树。"""
from datetime import datetime
from pathlib import Path
class StrategyVisualizer:
"""生成战略推理的可视化报告，保存到 sessions/strategy_vis/ 目录。"""
def __init__(self, session_dir="sessions/strategy_vis"):
self.dir = Path(session_dir)
self.dir.mkdir(parents=True, exist_ok=True)
self.report_count = 0
def render(self, trace, turn_number=0):
"""将一次战略推理 trace 渲染为可视化文本并保存。"""
if not trace:
return
self.report_count += 1
lines = []
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
lines.append(f"{'=' * 70}")
lines.append(f"战略推理报告 | 第{turn_number}轮 | {ts}")
lines.append(f"{'=' * 70}")
# 总结
lines.append(f"\n📋 会话总结:")
lines.append(f" {trace.get('summary', '?')}")
# 当前揭露度
current = trace.get("current_disclosure", "?")
lines.append(f"\n📊 当前揭露水平: {current}/10")
# 种子
seeds = trace.get("seeds", {})
lines.append(f"\n🌱 种子视角:")
for sid, seed in seeds.items():
lines.append(f" {sid}: {seed}")
# 路径树
candidates = trace.get("candidates", [])
if not candidates:
lines.append("\n⚠️ 无候选路径")
else:
# 按种子分组
from collections import defaultdict
groups = defaultdict(list)
for c in candidates:
groups[c["id"]].append(c)
# 标记有效路径（delta>0）
effective_ids = set()
for c in candidates:
if c.get("delta", 0) > 0:
effective_ids.add((c["id"], c["branch"]))
lines.append(f"\n🌳 路径树 (共{len(candidates)}条，{len(effective_ids)}条有效):")
lines.append(f" {'─' * 66}")
for sid in sorted(groups.keys()):
items = groups[sid]
seed_text = seeds.get(sid, "?")
max_score = max(i["score"] for i in items)
lines.append(f"")
lines.append(f" ┌─ 种子{sid}: {seed_text[:50]}")
lines.append(f" │ L1咨询师: {items[0].get('l1_reply', '?')[:55]}")
lines.append(f" │ L2来访者: {items[0].get('l2_client', '?')[:55]}")
lines.append(f" │")
for item in sorted(items, key=lambda x: x["branch"]):
bid = item["branch"]
score = item["score"]
delta = item.get("delta", 0)
is_effective = (sid, bid) in effective_ids
is_selected = f"{sid}.{bid}" == trace.get("selected", "")
# 标记符号
if is_selected:
marker = "★"
elif is_effective:
marker = "✓"
else:
marker = "·"
delta_str = f"+{delta}" if delta > 0 else str(delta)
lines.append(f" │ {marker} 分叉{bid}: score={score}/10 (Δ{delta_str}) | {item.get('reason', '')[:30]}")
lines.append(f" │ L3咨询师: {item.get('l3_reply', '?')[:50]}")
lines.append(f" │ L4来访者: {item.get('l4_client', '?')[:50]}")
lines.append(f" │ ── 种子{sid}最高分: {max_score}/10")
lines.append(f" └{'─' * 65}")
# 蒸馏结果
guidance = trace.get("guidance", {})
distill_count = guidance.get("_distill_count", len(effective_ids))
distill_ids = guidance.get("_distill_ids", [])
distill_label = f"{distill_count}条路径（前30%）" + (f" [{', '.join(distill_ids)}]" if distill_ids else "")
lines.append(f"\n🎯 蒸馏结果 (从{distill_label}):")
lines.append(f" 方向: {guidance.get('direction', '?')}")
for p in guidance.get("principles", []):
lines.append(f" • {p}")
lines.append(f" 证据: {guidance.get('evidence', '?')}")
# 计时
timing = trace.get("timing", {})
if timing:
lines.append(f"\n⏱ 计时:")
for k, v in timing.items():
lines.append(f" {k}: {v}s")
# 图例
lines.append(f"\n图例: ★=最终选中 ✓=有效路径(Δ>0) ·=未提升 | 蒸馏仅用前30%有效路径")
lines.append(f"{'=' * 70}\n")
report_text = "\n".join(lines)
# 保存
filename = f"turn{turn_number:03d}_{datetime.now().strftime('%H%M%S')}.txt"
filepath = self.dir / filename
with open(filepath, "w", encoding="utf-8") as f:
f.write(report_text)
# 同时打印到终端
print(report_text)
return str(filepath)
|