Spaces:
Sleeping
Sleeping
| import json | |
| import gradio as gr | |
| from pathlib import Path | |
| from counselor import PsychodynamicCounselor | |
| counselor = None | |
| def _depth_level(score): | |
| """将 1-10 分映射为探索层级""" | |
| if score <= 2: | |
| return "L1 · 表层", "社交性叙述，回避情感" | |
| elif score <= 4: | |
| return "L2 · 事件", "具体事件浮现，情绪初现" | |
| elif score <= 6: | |
| return "L3 · 情感", "情绪深化，开始反思模式" | |
| elif score <= 8: | |
| return "L4 · 核心", "触及核心冲突与早期经历" | |
| else: | |
| return "L5 · 突破", "深层揭露，防御松动" | |
| def _phase_label(turn): | |
| """会话阶段""" | |
| if turn <= 3: | |
| return "建立联盟" | |
| elif turn <= 8: | |
| return "探索展开" | |
| elif turn <= 15: | |
| return "深层工作" | |
| else: | |
| return "整合修通" | |
| def build_status_panel(c): | |
| """构建会话状态 Markdown 富文本面板""" | |
| score = c._last_disclosure_score | |
| turn = c.turn_number | |
| history = c._disclosure_history | |
| # ── 探索层级 ── | |
| level_name, level_desc = _depth_level(score) | |
| phase = _phase_label(turn) | |
| # 揭露深度进度条（渐变风格） | |
| bar_chars = "░▒▓█" | |
| filled_bar = "" | |
| for i in range(10): | |
| if i < score: | |
| ci = min(i // 3, 3) | |
| filled_bar += bar_chars[ci] | |
| else: | |
| filled_bar += "·" | |
| # 趋势计算 | |
| if len(history) >= 2: | |
| diff = history[-1] - history[-2] | |
| trend = "▲" if diff > 0 else ("▼" if diff < 0 else "━") | |
| trend_word = f"+{diff}" if diff > 0 else str(diff) if diff < 0 else "±0" | |
| else: | |
| trend = "·" | |
| trend_word = "—" | |
| # 均值 & 峰值 | |
| avg_score = sum(history) / len(history) if history else 0 | |
| peak_score = max(history) if history else 0 | |
| # 历史火花线 (sparkline) | |
| spark_chars = " ▁▂▃▄▅▆▇█" | |
| spark = "" | |
| for s in history[-20:]: | |
| idx = min(s, 9) | |
| spark += spark_chars[idx] | |
| if not spark: | |
| spark = "—" | |
| # 维度指示灯 | |
| dims = c._last_dimensions | |
| dim_labels = { | |
| "A": "具体事件", "B": "情绪表达", "C": "具体情绪", | |
| "D": "自我反思", "E": "回避触及", | |
| } | |
| dim_on = sum(1 for k in ["A", "B", "C", "D", "E"] if dims.get(k, False)) | |
| dim_parts = [] | |
| for k in ["A", "B", "C", "D", "E"]: | |
| on = dims.get(k, False) | |
| icon = "🟢" if on else "⚫" | |
| dim_parts.append(f"{icon} {dim_labels[k]}") | |
| dim_line = " · ".join(dim_parts) | |
| # ── 构建面板 ── | |
| lines = [] | |
| lines.append("### ◈ SESSION MONITOR") | |
| lines.append("") | |
| # 核心指标表 | |
| lines.append("| | |") | |
| lines.append("|:---|:---|") | |
| lines.append(f"| **轮次** | `T{turn}` · {phase} |") | |
| lines.append(f"| **探索层级** | **{level_name}** — {level_desc} |") | |
| lines.append(f"| **揭露深度** | `{filled_bar}` **{score}/10** {trend} ({trend_word}) |") | |
| lines.append(f"| **均值 / 峰值** | avg `{avg_score:.1f}` · peak `{peak_score}` |") | |
| lines.append(f"| **深度轨迹** | `{spark}` |") | |
| lines.append(f"| **维度命中** | **{dim_on}/5** |") | |
| lines.append("") | |
| # 维度详情 | |
| lines.append(f"> {dim_line}") | |
| lines.append("") | |
| # ── 督导模块 ── | |
| lines.append("---") | |
| if c.current_guidance: | |
| g = c.current_guidance | |
| direction = g.get("direction", "—") | |
| principles = g.get("principles", []) | |
| evidence = g.get("evidence", "") | |
| lines.append("#### ▸ 督导指令") | |
| lines.append("") | |
| lines.append(f"> 🎯 **{direction}**") | |
| if principles: | |
| lines.append(">") | |
| for i, p in enumerate(principles[:3], 1): | |
| lines.append(f"> {i}. {p}") | |
| if evidence: | |
| lines.append(">") | |
| lines.append(f"> 📌 {evidence[:100]}{'…' if len(evidence) > 100 else ''}") | |
| lines.append("") | |
| # ── PUCT 推理引擎 ── | |
| ts = c._last_trace_stats | |
| if ts: | |
| timing = ts.get("timing", {}) | |
| total_s = timing.get("total_seconds", 0) | |
| total_paths = ts.get("total_paths", 0) | |
| deep = ts.get("deep_paths", 0) | |
| seeds = ts.get("seeds", []) | |
| selected = ts.get("selected", "") | |
| best_score = ts.get("best_score", 0) | |
| best_delta = ts.get("best_delta", 0) | |
| predicted = ts.get("predicted_disclosure", "?") | |
| lines.append("---") | |
| lines.append("#### ▸ PUCT 推理引擎") | |
| lines.append("") | |
| lines.append("| | |") | |
| lines.append("|:---|:---|") | |
| lines.append(f"| **搜索架构** | `L1 → L2 → L3 → L4 → L5 → L6` (6层深度) |") | |
| lines.append(f"| **种子方向** | {' / '.join(seeds) if seeds else '—'} |") | |
| lines.append(f"| **路径探索** | {total_paths} 候选 → {deep} 深探 |") | |
| lines.append(f"| **最优路径** | `{selected}` |") | |
| lines.append(f"| **路径评分** | score=**{best_score}** · Δ=**{best_delta:+.1f}** |") | |
| lines.append(f"| **预测揭露** | **{predicted}**/10 |") | |
| lines.append(f"| **推理耗时** | `{total_s:.1f}s` |") | |
| else: | |
| lines.append("#### ▸ 推理引擎") | |
| lines.append("") | |
| lines.append("> ⏳ 同步推理中…") | |
| # ── 评估理由（折叠） ── | |
| if c._last_reasoning: | |
| lines.append("") | |
| lines.append(f"<details><summary>📋 评估理由</summary>\n\n{c._last_reasoning}\n\n</details>") | |
| return "\n".join(lines) | |
| def start_session(): | |
| global counselor | |
| counselor = PsychodynamicCounselor() | |
| return [], "### 🧠 SESSION MONITOR\n\n> 新会话已开始，等待来访者发言…" | |
| def chat(user_message, chat_history): | |
| global counselor | |
| if counselor is None: | |
| counselor = PsychodynamicCounselor() | |
| if not user_message.strip(): | |
| return chat_history, "", "", "" | |
| response = counselor.respond(user_message) | |
| chat_history = chat_history or [] | |
| chat_history.append({"role": "user", "content": user_message}) | |
| chat_history.append({"role": "assistant", "content": response}) | |
| # 构建富文本状态面板 | |
| status = build_status_panel(counselor) | |
| return chat_history, "", status, "" | |
| def end_session(): | |
| global counselor | |
| if counselor: | |
| path = counselor.get_session_filepath() | |
| counselor = None | |
| return f"会话已结束。日志保存于：{path}" | |
| return "当前无活跃会话。" | |
| def view_sessions(): | |
| files = sorted(Path("sessions").glob("session_*.json")) | |
| if not files: | |
| return "暂无会话记录" | |
| output = "" | |
| for f in files: | |
| with open(f, encoding="utf-8") as fp: | |
| data = json.load(fp) | |
| total = data.get("total_turns", len(data["turns"])) | |
| output += f"\n{'='*50}\n" | |
| output += f"Session: {data['session_id']} | 轮次: {total}\n" | |
| output += f"{'='*50}\n" | |
| for t in data["turns"]: | |
| score = t.get("disclosure_score", "?") | |
| output += f"\n[轮次 {t['turn_number']}] 揭露评分: {score}/5\n" | |
| output += f"来访者: {t['user_message']}\n" | |
| output += f"咨询师: {t['counselor_message']}\n" | |
| dims = t.get("dimension_score", {}) | |
| reason = t.get("reason", "") | |
| if dims: | |
| output += f"维度: {dims}\n" | |
| if reason: | |
| output += f"理由: {reason}\n" | |
| # 战略推理记录（每5轮） | |
| trace = t.get("mcts_trace") | |
| if trace and "selected_direction" in trace: | |
| output += f"\n === 战略推理（第{t['turn_number']}轮触发） ===\n" | |
| output += f" 总结: {trace.get('summary', '')[:80]}...\n" | |
| output += f" 选中: {trace['selected']} → {trace['selected_direction'][:50]}\n" | |
| output += f" 预测揭露: {trace.get('selected_score', '?')}/10\n" | |
| for d in trace.get("directions", []): | |
| marker = " ★" if d["id"] == trace["selected"] else "" | |
| output += f" [{d['id']}]{marker} {d.get('direction', '')[:40]} → 揭露={d.get('disclosure_level', '?')}/10\n" | |
| output += f" ========================\n" | |
| return output | |
| def download_all_sessions(): | |
| files = sorted(Path("sessions").glob("session_*.json")) | |
| if not files: | |
| return None | |
| return [str(f) for f in files] | |
| with gr.Blocks(title="Freud-Zero MVP") as app: | |
| gr.Markdown("# Freud-Zero MVP") | |
| gr.Markdown("精神动力学取向回应性咨询师 · 自我揭露深度追踪") | |
| with gr.Row(): | |
| btn_start = gr.Button("开始新会话", variant="primary") | |
| btn_end = gr.Button("结束会话", variant="stop") | |
| chatbot = gr.Chatbot(label="对话", height=480, type="messages") | |
| with gr.Row(): | |
| user_input = gr.Textbox(placeholder="说你想说的……", show_label=False, scale=4) | |
| btn_send = gr.Button("发送", scale=1) | |
| status_output = gr.Markdown(value="### 🧠 SESSION MONITOR\n\n> 等待开始会话…") | |
| with gr.Accordion("研究者面板", open=False): | |
| with gr.Row(): | |
| btn_view = gr.Button("查看所有会话记录") | |
| btn_download = gr.Button("下载日志文件") | |
| log_display = gr.Textbox(label="会话日志", lines=20, interactive=False) | |
| file_output = gr.File(label="日志文件") | |
| # 绑定事件 | |
| btn_start.click(start_session, outputs=[chatbot, status_output]) | |
| btn_end.click(end_session, outputs=[status_output]) | |
| btn_send.click(chat, inputs=[user_input, chatbot], outputs=[chatbot, user_input, status_output, log_display]) | |
| user_input.submit(chat, inputs=[user_input, chatbot], outputs=[chatbot, user_input, status_output, log_display]) | |
| btn_view.click(view_sessions, outputs=[log_display]) | |
| btn_download.click(download_all_sessions, outputs=[file_output]) | |
| if __name__ == "__main__": | |
| app.launch(server_name="0.0.0.0", server_port=7860, share=True) | |