#!/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('<', '<')
.replace('>', '>')
.replace('"', '"')
.replace("'", '''))
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 = '''
知识获取工作流可视化
知识获取工作流可视化
'''
# 生成Tab标签
for i, workflow in enumerate(workflows):
question = workflow['input'].get('question', f'问题 {i+1}')
active_class = 'active' if i == 0 else ''
html += f' \n'
html += '
\n'
# 生成Tab内容
for i, workflow in enumerate(workflows):
active_class = 'active' if i == 0 else ''
html += f'