kg_agent/knowledge_v2/visualization/workflow_visualization.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
知识获取工作流可视化
1. 读取知识获取工作流详细过程数据文件（参考 .cache/9f510b2a8348/execution_record.json）
2. 文件路径可设置，每个文件表示一个输入信息执行过程，可能有多个。多个文件路径硬编码在代码中，用list表示
3. 将知识获取工作流用html页面展示出来
4. HTML 文件输出路径在当前目录下，文件名称为 workflow_visualization_datetime.html
'''

import json
import os
from datetime import datetime
from pathlib import Path


# 硬编码的文件路径列表（相对于项目根目录）
DATA_FILE_PATHS = [
    "../.cache/9f510b2a8348/execution_record.json",
    # 可以在这里添加更多文件路径
]


def load_data_files(file_paths):
    """读取并解析JSON文件"""
    data_list = []
    script_dir = Path(__file__).parent
    for file_path in file_paths:
        # 将相对路径转换为绝对路径
        abs_path = (script_dir / file_path).resolve()
        if not abs_path.exists():
            print(f"警告: 文件不存在: {abs_path}")
            continue
        try:
            with open(abs_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
                data_list.append(data)
        except (json.JSONDecodeError, IOError) as e:
            print(f"错误: 读取文件失败 {abs_path}: {e}")
    return data_list


def parse_workflow_data(data):
    """解析工作流数据，提取关键信息"""
    workflow = {
        'input': {},
        'steps': [],
        'output': {}
    }
    
    # 提取输入信息
    if 'input' in data:
        workflow['input'] = {
            'question': data['input'].get('question', ''),
            'post_info': data['input'].get('post_info', ''),
            'persona_info': data['input'].get('persona_info', '')
        }
    
    # 提取执行流程
    if 'execution' in data and 'modules' in data['execution']:
        function_knowledge = data['execution']['modules'].get('function_knowledge', {})
        
        # 步骤1: 生成query
        if 'generate_query' in function_knowledge:
            generate_query = function_knowledge['generate_query']
            workflow['steps'].append({
                'step': 'generate_query',
                'name': '生成查询',
                'query': generate_query.get('query', ''),
                'prompt': generate_query.get('prompt', '')
            })
        
        # 步骤2: 选择工具
        if 'select_tool' in function_knowledge:
            select_tool = function_knowledge['select_tool']
            response = select_tool.get('response', {})
            workflow['steps'].append({
                'step': 'select_tool',
                'name': '选择工具',
                'prompt': select_tool.get('prompt', ''),
                'tool_name': response.get('工具名', ''),
                'tool_id': response.get('工具调用ID', ''),
                'tool_usage': response.get('使用方法', '')
            })
        
        # 步骤3: 提取参数
        if 'extract_params' in function_knowledge:
            extract_params = function_knowledge['extract_params']
            workflow['steps'].append({
                'step': 'extract_params',
                'name': '提取参数',
                'prompt': extract_params.get('prompt', ''),
                'params': extract_params.get('params', {})
            })
        
        # 步骤4: 执行工具
        if 'execute_tool' in function_knowledge:
            execute_tool = function_knowledge['execute_tool']
            workflow['steps'].append({
                'step': 'execute_tool',
                'name': '执行工具',
                'response': execute_tool.get('response', '')
            })
    
    # 提取输出信息
    if 'output' in data:
        workflow['output'] = {
            'result': data['output'].get('result', '')
        }
    
    return workflow


def escape_html(text):
    """转义HTML特殊字符"""
    if not isinstance(text, str):
        text = str(text)
    return (text.replace('&', '&')
                .replace('<', '&lt;')
                .replace('>', '&gt;')
                .replace('"', '&quot;')
                .replace("'", '&#39;'))


def format_json_for_display(obj):
    """格式化JSON对象用于显示"""
    if isinstance(obj, dict):
        return json.dumps(obj, ensure_ascii=False, indent=2)
    elif isinstance(obj, str):
        try:
            parsed = json.loads(obj)
            return json.dumps(parsed, ensure_ascii=False, indent=2)
        except (json.JSONDecodeError, ValueError):
            return obj
    return str(obj)


def generate_html(workflows):
    """生成HTML页面"""
    html = '''<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>知识获取工作流可视化</title>
    <style>
        * {
            margin: 0;
            padding: 0;
            box-sizing: border-box;
        }
        
        body {
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC', 'Hiragino Sans GB', 'Microsoft YaHei', sans-serif;
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: #333;
            line-height: 1.6;
            min-height: 100vh;
        }
        
        .container {
            max-width: 1400px;
            margin: 0 auto;
            padding: 30px 20px;
        }
        
        h1 {
            text-align: center;
            color: white;
            margin-bottom: 40px;
            font-size: 32px;
            font-weight: 600;
            text-shadow: 0 2px 10px rgba(0,0,0,0.2);
            letter-spacing: 1px;
        }
        
        .tabs {
            display: flex;
            background: white;
            border-radius: 12px 12px 0 0;
            box-shadow: 0 4px 20px rgba(0,0,0,0.15);
            overflow-x: auto;
            padding: 5px;
        }
        
        .tab {
            padding: 16px 28px;
            cursor: pointer;
            border: none;
            background: transparent;
            color: #666;
            font-size: 14px;
            transition: all 0.3s cubic-bezier(0.4, 0, 0.2, 1);
            white-space: nowrap;
            border-radius: 8px;
            margin: 0 4px;
            position: relative;
            font-weight: 500;
        }
        
        .tab:hover {
            background: #f0f0f0;
            color: #333;
        }
        
        .tab.active {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white;
            box-shadow: 0 4px 12px rgba(102, 126, 234, 0.4);
        }
        
        .tab-content {
            display: none;
            background: white;
            padding: 40px;
            border-radius: 0 0 12px 12px;
            box-shadow: 0 4px 20px rgba(0,0,0,0.15);
            margin-bottom: 20px;
            animation: fadeIn 0.3s ease-in;
        }
        
        @keyframes fadeIn {
            from { opacity: 0; transform: translateY(10px); }
            to { opacity: 1; transform: translateY(0); }
        }
        
        .tab-content.active {
            display: block;
        }
        
        .input-section {
            background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);
            padding: 28px;
            border-radius: 12px;
            margin-bottom: 35px;
            box-shadow: 0 4px 15px rgba(0,0,0,0.1);
            border: 1px solid rgba(255,255,255,0.5);
        }
        
        .input-section h3 {
            color: #2c3e50;
            margin-bottom: 20px;
            font-size: 20px;
            font-weight: 600;
            display: flex;
            align-items: center;
            gap: 10px;
        }
        
        .input-section h3::before {
            content: '📋';
            font-size: 24px;
        }
        
        .input-item {
            margin-bottom: 16px;
            padding: 12px;
            background: rgba(255,255,255,0.7);
            border-radius: 8px;
            transition: all 0.3s;
        }
        
        .input-item:hover {
            background: rgba(255,255,255,0.9);
            transform: translateX(5px);
        }
        
        .input-item strong {
            color: #495057;
            display: inline-block;
            width: 110px;
            font-weight: 600;
        }
        
        .input-item .placeholder {
            color: #999;
            font-style: italic;
        }
        
        .workflow {
            position: relative;
        }
        
        .workflow-step {
            background: white;
            border: 2px solid #e0e0e0;
            border-radius: 12px;
            margin-bottom: 25px;
            transition: all 0.3s cubic-bezier(0.4, 0, 0.2, 1);
            overflow: hidden;
            box-shadow: 0 2px 8px rgba(0,0,0,0.08);
        }
        
        .workflow-step.active {
            border-color: #667eea;
            box-shadow: 0 8px 24px rgba(102, 126, 234, 0.25);
            transform: translateY(-2px);
        }
        
        .step-header {
            padding: 20px 24px;
            background: linear-gradient(135deg, #f8f9fa 0%, #e9ecef 100%);
            cursor: pointer;
            display: flex;
            justify-content: space-between;
            align-items: center;
            user-select: none;
            transition: all 0.3s;
        }
        
        .step-header:hover {
            background: linear-gradient(135deg, #e9ecef 0%, #dee2e6 100%);
        }
        
        .workflow-step.active .step-header {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white;
        }
        
        .workflow-step.active .step-name {
            color: white;
        }
        
        .workflow-step.active .step-toggle {
            color: white;
        }
        
        .step-title {
            display: flex;
            align-items: center;
            gap: 15px;
        }
        
        .step-number {
            display: inline-flex;
            align-items: center;
            justify-content: center;
            width: 36px;
            height: 36px;
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white;
            border-radius: 50%;
            font-size: 16px;
            font-weight: bold;
            box-shadow: 0 4px 12px rgba(102, 126, 234, 0.3);
        }
        
        .workflow-step.active .step-number {
            background: white;
            color: #667eea;
            box-shadow: 0 4px 12px rgba(255,255,255,0.3);
        }
        
        .step-name {
            font-size: 18px;
            font-weight: 600;
            color: #2c3e50;
        }
        
        .step-toggle {
            color: #6c757d;
            font-size: 20px;
            transition: transform 0.3s cubic-bezier(0.4, 0, 0.2, 1);
        }
        
        .step-toggle.expanded {
            transform: rotate(180deg);
        }
        
        .step-content {
            padding: 0 20px;
            max-height: 0;
            overflow: hidden;
            transition: max-height 0.3s ease-out, padding 0.3s;
        }
        
        .step-content.expanded {
            max-height: 5000px;
            padding: 20px;
        }
        
        .step-detail {
            margin-bottom: 20px;
        }
        
        .step-detail-label {
            font-weight: 600;
            color: #495057;
            margin-bottom: 10px;
            display: block;
            font-size: 14px;
            text-transform: uppercase;
            letter-spacing: 0.5px;
        }
        
        .step-detail-content {
            background: #f8f9fa;
            padding: 16px;
            border-radius: 8px;
            border-left: 4px solid #667eea;
            font-size: 14px;
            line-height: 1.8;
            white-space: pre-wrap;
            word-wrap: break-word;
            max-height: 400px;
            overflow-y: auto;
            box-shadow: 0 2px 8px rgba(0,0,0,0.05);
        }
        
        .json-content {
            font-family: 'SF Mono', 'Monaco', 'Courier New', monospace;
            background: #1e1e1e;
            color: #d4d4d4;
            padding: 20px;
            border-radius: 8px;
            overflow-x: auto;
            border-left: 4px solid #667eea;
            box-shadow: 0 4px 12px rgba(0,0,0,0.15);
        }
        
        .prompt-toggle-btn {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white;
            border: none;
            padding: 10px 20px;
            border-radius: 6px;
            cursor: pointer;
            font-size: 13px;
            font-weight: 500;
            margin-top: 15px;
            transition: all 0.3s;
            box-shadow: 0 2px 8px rgba(102, 126, 234, 0.3);
        }
        
        .prompt-toggle-btn:hover {
            transform: translateY(-2px);
            box-shadow: 0 4px 12px rgba(102, 126, 234, 0.4);
        }
        
        .prompt-content {
            display: none;
            margin-top: 15px;
            padding: 16px;
            background: #fff3cd;
            border-radius: 8px;
            border-left: 4px solid #ffc107;
            font-size: 13px;
            line-height: 1.8;
            white-space: pre-wrap;
            word-wrap: break-word;
            max-height: 500px;
            overflow-y: auto;
        }
        
        .prompt-content.show {
            display: block;
            animation: slideDown 0.3s ease-out;
        }
        
        @keyframes slideDown {
            from {
                opacity: 0;
                max-height: 0;
            }
            to {
                opacity: 1;
                max-height: 500px;
            }
        }
        
        .output-section {
            background: linear-gradient(135deg, #e0f2fe 0%, #bae6fd 100%);
            padding: 28px;
            border-radius: 12px;
            margin-top: 35px;
            border-left: 4px solid #0ea5e9;
            box-shadow: 0 4px 15px rgba(14, 165, 233, 0.2);
        }
        
        .output-section h3 {
            color: #0369a1;
            margin-bottom: 20px;
            font-size: 20px;
            font-weight: 600;
            display: flex;
            align-items: center;
            gap: 10px;
        }
        
        .output-section h3::before {
            content: '✨';
            font-size: 24px;
        }
        
        .arrow {
            text-align: center;
            color: #667eea;
            font-size: 32px;
            margin: -15px 0;
            position: relative;
            z-index: 1;
            filter: drop-shadow(0 2px 4px rgba(102, 126, 234, 0.3));
        }
        
        .arrow::before {
            content: '↓';
        }
        
        @media (max-width: 768px) {
            .container {
                padding: 10px;
            }
            
            .tab {
                padding: 12px 15px;
                font-size: 13px;
            }
            
            .tab-content {
                padding: 20px;
            }
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>知识获取工作流可视化</h1>
        <div class="tabs" id="tabs">
'''
    
    # 生成Tab标签
    for i, workflow in enumerate(workflows):
        question = workflow['input'].get('question', f'问题 {i+1}')
        active_class = 'active' if i == 0 else ''
        html += f'            <button class="tab {active_class}" onclick="switchTab({i})">{escape_html(question)}</button>\n'
    
    html += '        </div>\n'
    
    # 生成Tab内容
    for i, workflow in enumerate(workflows):
        active_class = 'active' if i == 0 else ''
        html += f'        <div class="tab-content {active_class}" id="tab-{i}">\n'
        
        # 输入信息
        html += '            <div class="input-section">\n'
        html += '                <h3>输入信息</h3>\n'
        html += f'                <div class="input-item"><strong>问题:</strong> {escape_html(workflow["input"].get("question", ""))}</div>\n'
        post_info = workflow['input'].get('post_info', '')
        post_info_display = escape_html(post_info) if post_info else '<span class="placeholder">（无）</span>'
        html += f'                <div class="input-item"><strong>帖子信息:</strong> {post_info_display}</div>\n'
        persona_info = workflow['input'].get('persona_info', '')
        persona_info_display = escape_html(persona_info) if persona_info else '<span class="placeholder">（无）</span>'
        html += f'                <div class="input-item"><strong>人设信息:</strong> {persona_info_display}</div>\n'
        html += '            </div>\n'
        
        # 工作流程
        html += '            <div class="workflow">\n'
        for j, step in enumerate(workflow['steps']):
            step_id = f"step-{i}-{j}"
            html += f'                <div class="workflow-step" id="{step_id}">\n'
            html += '                    <div class="step-header" onclick="toggleStep(\'' + step_id + '\')">\n'
            html += '                        <div class="step-title">\n'
            html += f'                            <span class="step-number">{j+1}</span>\n'
            html += f'                            <span class="step-name">{escape_html(step["name"])}</span>\n'
            html += '                        </div>\n'
            html += '                        <span class="step-toggle">▼</span>\n'
            html += '                    </div>\n'
            html += '                    <div class="step-content" id="content-' + step_id + '">\n'
            
            # 根据步骤类型显示不同内容（prompt放在最后，默认隐藏）
            prompt_id = f"prompt-{step_id}"
            
            if step['step'] == 'generate_query':
                if step.get('query'):
                    html += '                        <div class="step-detail">\n'
                    html += '                            <span class="step-detail-label">生成的Query:</span>\n'
                    html += f'                            <div class="step-detail-content">{escape_html(step["query"])}</div>\n'
                    html += '                        </div>\n'
            
            elif step['step'] == 'select_tool':
                if step.get('tool_name'):
                    html += '                        <div class="step-detail">\n'
                    html += '                            <span class="step-detail-label">工具名称:</span>\n'
                    html += f'                            <div class="step-detail-content">{escape_html(step["tool_name"])}</div>\n'
                    html += '                        </div>\n'
                if step.get('tool_id'):
                    html += '                        <div class="step-detail">\n'
                    html += '                            <span class="step-detail-label">工具调用ID:</span>\n'
                    html += f'                            <div class="step-detail-content">{escape_html(step["tool_id"])}</div>\n'
                    html += '                        </div>\n'
                if step.get('tool_usage'):
                    html += '                        <div class="step-detail">\n'
                    html += '                            <span class="step-detail-label">使用方法:</span>\n'
                    html += f'                            <div class="step-detail-content">{escape_html(step["tool_usage"])}</div>\n'
                    html += '                        </div>\n'
            
            elif step['step'] == 'extract_params':
                if step.get('params'):
                    html += '                        <div class="step-detail">\n'
                    html += '                            <span class="step-detail-label">提取的参数:</span>\n'
                    params_str = format_json_for_display(step['params'])
                    html += f'                            <div class="step-detail-content json-content">{escape_html(params_str)}</div>\n'
                    html += '                        </div>\n'
            
            elif step['step'] == 'execute_tool':
                if step.get('response'):
                    html += '                        <div class="step-detail">\n'
                    html += '                            <span class="step-detail-label">执行结果:</span>\n'
                    response_str = format_json_for_display(step['response'])
                    html += f'                            <div class="step-detail-content json-content">{escape_html(response_str)}</div>\n'
                    html += '                        </div>\n'
            
            # Prompt放在最后，默认隐藏
            if step.get('prompt'):
                html += f'                        <button class="prompt-toggle-btn" onclick="togglePrompt(\'{prompt_id}\')">显示 Prompt</button>\n'
                html += f'                        <div class="prompt-content" id="{prompt_id}">{escape_html(step["prompt"])}</div>\n'
            
            html += '                    </div>\n'
            html += '                </div>\n'
            
            # 添加箭头（除了最后一步）
            if j < len(workflow['steps']) - 1:
                html += '                <div class="arrow"></div>\n'
        
        html += '            </div>\n'
        
        # 输出信息
        if workflow['output'].get('result'):
            html += '            <div class="output-section">\n'
            html += '                <h3>最终输出</h3>\n'
            result_str = format_json_for_display(workflow['output']['result'])
            html += f'                <div class="step-detail-content json-content">{escape_html(result_str)}</div>\n'
            html += '            </div>\n'
        
        html += '        </div>\n'
    
    html += '''    </div>
    
    <script>
        function switchTab(index) {
            // 隐藏所有tab内容
            const contents = document.querySelectorAll('.tab-content');
            contents.forEach(content => content.classList.remove('active'));
            
            // 移除所有tab的active状态
            const tabs = document.querySelectorAll('.tab');
            tabs.forEach(tab => tab.classList.remove('active'));
            
            // 显示选中的tab内容
            document.getElementById('tab-' + index).classList.add('active');
            tabs[index].classList.add('active');
        }
        
        function toggleStep(stepId) {
            const step = document.getElementById(stepId);
            const content = document.getElementById('content-' + stepId);
            const toggle = step.querySelector('.step-toggle');
            
            if (content.classList.contains('expanded')) {
                content.classList.remove('expanded');
                toggle.classList.remove('expanded');
                step.classList.remove('active');
            } else {
                content.classList.add('expanded');
                toggle.classList.add('expanded');
                step.classList.add('active');
            }
        }
        
        function togglePrompt(promptId) {
            const promptContent = document.getElementById(promptId);
            const btn = promptContent.previousElementSibling;
            
            if (promptContent.classList.contains('show')) {
                promptContent.classList.remove('show');
                btn.textContent = '显示 Prompt';
            } else {
                promptContent.classList.add('show');
                btn.textContent = '隐藏 Prompt';
            }
        }
        
        // 页面加载时高亮第一个步骤
        window.addEventListener('load', function() {
            const firstSteps = document.querySelectorAll('.workflow-step');
            firstSteps.forEach((step, index) => {
                if (index === 0 || index % (firstSteps.length / document.querySelectorAll('.tab-content').length) === 0) {
                    step.classList.add('active');
                    const content = step.querySelector('.step-content');
                    const toggle = step.querySelector('.step-toggle');
                    if (content) {
                        content.classList.add('expanded');
                        toggle.classList.add('expanded');
                    }
                }
            });
        });
    </script>
</body>
</html>'''
    
    return html


def main():
    """主函数"""
    # 获取当前脚本所在目录
    script_dir = Path(__file__).parent
    os.chdir(script_dir)
    
    # 读取数据文件
    print("正在读取数据文件...")
    data_list = load_data_files(DATA_FILE_PATHS)
    
    if not data_list:
        print("错误: 没有成功读取任何数据文件")
        return
    
    print(f"成功读取 {len(data_list)} 个数据文件")
    
    # 解析工作流数据
    print("正在解析工作流数据...")
    workflows = [parse_workflow_data(data) for data in data_list]
    
    # 生成HTML
    print("正在生成HTML页面...")
    html = generate_html(workflows)
    
    # 保存HTML文件
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_filename = f"workflow_visualization_{timestamp}.html"
    
    with open(output_filename, 'w', encoding='utf-8') as f:
        f.write(html)
    
    print(f"HTML页面已生成: {output_filename}")
    print(f"文件路径: {os.path.abspath(output_filename)}")


if __name__ == '__main__':
    main()