Spaces:
Sleeping
Sleeping
| """战略推理可视化：将每次战略决策的完整路径渲染为可读的文本树。""" | |
| 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) | |