Spaces:
Running
Running
File size: 9,851 Bytes
143d94c |
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 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 |
#!/usr/bin/env python3
"""
Скрипт для швидкого перегляду розмов MAI-DX сесії в терміналі
"""
import os
import json
import sys
import argparse
from datetime import datetime
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.syntax import Syntax
from rich.markdown import Markdown
# Спробуємо імпортувати Rich, якщо немає - використаємо простий вивід
try:
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.syntax import Syntax
from rich.markdown import Markdown
RICH_AVAILABLE = True
console = Console()
except ImportError:
RICH_AVAILABLE = False
print("⚠️ Rich не встановлено. Для кращого відображення: pip install rich")
def load_session(case_id, log_dir="mai_dx_logs"):
"""Завантажити конкретну сесію"""
json_path = os.path.join(log_dir, f"{case_id}.json")
if not os.path.exists(json_path):
# Спробуємо знайти файл з частковим збігом
for filename in os.listdir(log_dir):
if case_id in filename and filename.endswith('.json'):
json_path = os.path.join(log_dir, filename)
break
else:
return None
try:
with open(json_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"❌ Помилка читання файлу: {e}")
return None
def display_session_rich(session):
"""Відображення сесії з Rich форматуванням"""
# Заголовок
console.print(Panel.fit(
f"[bold blue]MAI-DX Diagnostic Session[/bold blue]\n"
f"[dim]{session['case_id']}[/dim]",
border_style="blue"
))
# Основна інформація
info_table = Table(show_header=False, box=None, padding=(0, 2))
info_table.add_column(style="bold cyan")
info_table.add_column()
info_table.add_row("📋 Випадок:", session['case_name'])
info_table.add_row("🕐 Час:", session['timestamp'])
info_table.add_row("🔧 Режим:", session['mode'])
info_table.add_row("💊 Діагноз:", f"[bold yellow]{session['diagnosis']}[/bold yellow]")
info_table.add_row("📊 Точність:", f"{session['confidence']}/5.0 {'✅' if session['confidence'] >= 3.0 else '⚠️'}")
info_table.add_row("💰 Вартість:", f"${session['cost']} (бюджет: ${session['budget']})")
info_table.add_row("⏱️ Тривалість:", f"{session['duration']:.1f} секунд")
console.print(info_table)
console.print()
# Розмови
conversations = session.get('conversations', [])
if not conversations:
console.print("[yellow]⚠️ Розмови не знайдено[/yellow]")
return
console.print(f"[bold]💬 Розмови ({len(conversations)} раундів):[/bold]\n")
for i, conv in enumerate(conversations):
# Заголовок раунду
console.print(Panel(
f"[bold]Раунд {conv['round_number']}[/bold]\n"
f"Час: {conv.get('start_time', 'N/A')} - {conv.get('end_time', 'N/A')}\n"
f"Повідомлень: {len(conv.get('messages', []))}"
+ (f"\nРішення: [yellow]{conv['decision']}[/yellow]" if conv.get('decision') else "")
+ (f"\nВартість: ${conv['cost_incurred']}" if conv.get('cost_incurred') else ""),
title=f"[cyan]Раунд {i+1}[/cyan]",
border_style="cyan"
))
# Повідомлення
for msg in conv.get('messages', []):
agent_color = {
'Dr. Hypothesis': 'blue',
'Dr. Test-Chooser': 'green',
'Dr. Challenger': 'red',
'Dr. Stewardship': 'yellow',
'Dr. Checklist': 'magenta',
'Consensus Coordinator': 'cyan',
'Judge': 'white',
'Gatekeeper': 'bright_black'
}.get(msg['agent_name'], 'white')
msg_type_icon = {
'reasoning': '🤔',
'decision': '💡',
'input': '❓',
'output': '📊',
'log': '📝'
}.get(msg['message_type'], '💬')
console.print(
f"\n[bold {agent_color}]{msg_type_icon} {msg['agent_name']}[/bold {agent_color}] "
f"[dim]({msg['message_type']})[/dim]"
)
# Обмежуємо довжину контенту для кращого відображення
content = msg['content']
if len(content) > 500:
content = content[:500] + "..."
console.print(Panel(
content,
border_style=agent_color,
padding=(0, 1)
))
def display_session_simple(session):
"""Відображення сесії без Rich"""
print("="*60)
print(f"MAI-DX Diagnostic Session: {session['case_id']}")
print("="*60)
print(f"\n📋 Випадок: {session['case_name']}")
print(f"🕐 Час: {session['timestamp']}")
print(f"💊 Діагноз: {session['diagnosis']}")
print(f"📊 Точність: {session['confidence']}/5.0")
print(f"💰 Вартість: ${session['cost']} (бюджет: ${session['budget']})")
conversations = session.get('conversations', [])
if not conversations:
print("\n⚠️ Розмови не знайдено")
return
print(f"\n💬 Розмови ({len(conversations)} раундів):")
for i, conv in enumerate(conversations):
print(f"\n{'='*40}")
print(f"Раунд {conv['round_number']}")
print(f"{'='*40}")
if conv.get('decision'):
print(f"Рішення: {conv['decision']}")
for msg in conv.get('messages', []):
print(f"\n[{msg['message_type']}] {msg['agent_name']}:")
content = msg['content']
if len(content) > 300:
content = content[:300] + "..."
print(content)
print("-"*40)
def export_markdown(session, output_file=None):
"""Експорт розмов в Markdown"""
if not output_file:
output_file = f"{session['case_id']}_conversation.md"
md_content = f"""# MAI-DX Diagnostic Session: {session['case_id']}
## 📋 Інформація про сесію
- **Випадок**: {session['case_name']}
- **Час**: {session['timestamp']}
- **Режим**: {session['mode']}
- **Діагноз**: **{session['diagnosis']}**
- **Точність**: {session['confidence']}/5.0
- **Вартість**: ${session['cost']} (бюджет: ${session['budget']})
- **Тривалість**: {session['duration']:.1f} секунд
## 💬 Розмови
"""
for i, conv in enumerate(session.get('conversations', [])):
md_content += f"\n### Раунд {conv['round_number']}\n\n"
if conv.get('decision'):
md_content += f"**Рішення**: {conv['decision']}\n\n"
for msg in conv.get('messages', []):
md_content += f"\n#### {msg['agent_name']} [{msg['message_type']}]\n\n"
md_content += f"{msg['content']}\n\n"
md_content += "---\n"
with open(output_file, 'w', encoding='utf-8') as f:
f.write(md_content)
print(f"\n✅ Експортовано в Markdown: {output_file}")
def main():
parser = argparse.ArgumentParser(description='Перегляд розмов MAI-DX сесії')
parser.add_argument('case_id', help='ID сесії або його частина')
parser.add_argument('--dir', default='mai_dx_logs', help='Директорія з логами')
parser.add_argument('--simple', action='store_true', help='Простий вивід без Rich')
parser.add_argument('--export-md', action='store_true', help='Експортувати в Markdown')
parser.add_argument('--raw', action='store_true', help='Показати сирий текст')
args = parser.parse_args()
# Завантажуємо сесію
session = load_session(args.case_id, args.dir)
if not session:
print(f"❌ Сесія '{args.case_id}' не знайдена в {args.dir}")
# Показуємо доступні сесії
print("\n📁 Доступні сесії:")
for filename in sorted(os.listdir(args.dir)):
if filename.endswith('.json') and not filename.endswith('_analytics.json'):
print(f" - {filename[:-5]}")
sys.exit(1)
# Показуємо сирий текст якщо потрібно
if args.raw:
raw_file = os.path.join(args.dir, f"{session['case_id']}_raw.txt")
if os.path.exists(raw_file):
with open(raw_file, 'r', encoding='utf-8') as f:
print(f.read())
else:
print("❌ Файл з сирим текстом не знайдено")
return
# Експортуємо в Markdown якщо потрібно
if args.export_md:
export_markdown(session)
return
# Відображаємо сесію
if args.simple or not RICH_AVAILABLE:
display_session_simple(session)
else:
display_session_rich(session)
if __name__ == "__main__":
main() |