工作流JSON实战技巧

一、压缩JSON的格式化与结构统一

从PandaAI导出的JSON文件有时是压缩格式（单行），在IDE中编辑极其困难。同时，某些导出格式可能包含冗余的 nodes 和 litegraph.nodes 双重结构，需要统一处理。

问题现象：

• JSON文件被压缩成单行，无法阅读和编辑
• 存在 litegraph.nodes 和顶层 nodes 两套节点数据，结构不一致
• 需要保留 litegraph 结构，清理冗余数据

格式化脚本：

import json
import sys

def format_workflow_json(input_file, output_file=None):
    """
    格式化工作流JSON文件，统一结构
    
    功能：
    1. 格式化压缩的JSON为可读格式
    2. 统一节点结构（只保留litegraph.nodes）
    3. 清理冗余的顶层nodes和links
    4. 验证JSON有效性
    """
    # 读取文件
    try:
        with open(input_file, 'r', encoding='utf-8') as f:
            content = f.read()
            workflow = json.loads(content)
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")
        return False
    
    # 统一结构：确保只使用litegraph结构
    if 'litegraph' not in workflow:
        print("错误：未找到litegraph结构")
        return False
    
    litegraph = workflow['litegraph']
    
    # 如果存在顶层nodes/links，检查是否需要合并
    if 'nodes' in workflow and workflow['nodes']:
        print("警告：发现顶层nodes结构，将使用litegraph.nodes")
        # 可以选择合并，但通常litegraph.nodes是标准格式
        # 这里选择保留litegraph.nodes，忽略顶层nodes
    
    # 清理冗余字段（可选）
    # 移除顶层nodes和links（如果存在且与litegraph重复）
    if 'nodes' in workflow:
        del workflow['nodes']
    if 'links' in workflow:
        del workflow['links']
    
    # 确保litegraph结构完整
    if 'nodes' not in litegraph:
        litegraph['nodes'] = []
    if 'links' not in litegraph:
        litegraph['links'] = []
    if 'version' not in litegraph:
        litegraph['version'] = 0.4
    
    # 输出文件
    if output_file is None:
        output_file = input_file.replace('.json', '_格式化.json')
    
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(workflow, f, indent=2, ensure_ascii=False)
    
    print(f"格式化完成: {output_file}")
    print(f"节点数量: {len(litegraph['nodes'])}")
    print(f"连接数量: {len(litegraph['links'])}")
    
    return True

# 使用示例
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python format_json.py <输入文件> [输出文件]")
        sys.exit(1)
    
    input_file = sys.argv[1]
    output_file = sys.argv[2] if len(sys.argv) > 2 else None
    
    format_workflow_json(input_file, output_file)

IDE快速格式化（VS Code）：

1. 安装扩展：JSON Tools 或使用内置格式化（Shift+Alt+F）
2. 对于压缩JSON，先使用 JSON Tools: Minify 再 Format Document
3. 使用命令面板：Format Document With... → 选择 JSON

命令行一键格式化：

# 使用Python
python -c "import json, sys; json.dump(json.load(open(sys.argv[1])), open(sys.argv[2], 'w'), indent=2, ensure_ascii=False)" input.json output.json

# 使用jq（需要安装）
jq . input.json > output.json

# 使用Node.js
node -e "const fs=require('fs'); const data=JSON.parse(fs.readFileSync(process.argv[2])); fs.writeFileSync(process.argv[3], JSON.stringify(data, null, 2))" input.json output.json

二、JSON结构差异的兼容处理

不同版本或不同导出方式产生的JSON结构可能不同，需要编写兼容性处理脚本，确保工作流能在各种环境下正常运行。

常见结构差异：

1. 节点ID类型：有些是数字，有些是字符串
2. 连接格式：数组格式 [id, src, slot, tgt, slot, type] vs 对象格式 {id, source, target, ...}
3. 字段缺失：某些节点缺少 widgets_values、flags 等字段
4. 嵌套结构：litegraph.nodes vs 顶层 nodes

兼容性处理脚本：

import json
from typing import Any, Dict, List

def normalize_workflow(workflow: Dict[str, Any]) -> Dict[str, Any]:
    """
    标准化工作流JSON结构，处理各种格式差异
    
    返回:
        标准化后的工作流JSON
    """
    # 1. 统一节点结构
    if 'litegraph' not in workflow:
        # 如果没有litegraph，尝试从顶层nodes构建
        if 'nodes' in workflow:
            workflow['litegraph'] = {
                'id': workflow.get('id', '00000000-0000-0000-0000-000000000000'),
                'version': workflow.get('version', 0.4),
                'nodes': workflow['nodes'],
                'links': workflow.get('links', []),
                'groups': workflow.get('groups', []),
                'config': workflow.get('config', {}),
                'extra': workflow.get('extra', {})
            }
        else:
            raise ValueError("无法找到节点数据")
    
    litegraph = workflow['litegraph']
    nodes = litegraph.get('nodes', [])
    links = litegraph.get('links', [])
    
    # 2. 标准化节点ID（确保都是整数）
    node_id_map = {}
    for i, node in enumerate(nodes, start=1):
        old_id = node.get('id')
        if isinstance(old_id, str):
            # 字符串ID转为整数
            try:
                new_id = int(old_id)
            except ValueError:
                new_id = i
        else:
            new_id = old_id if old_id else i
        
        node_id_map[old_id] = new_id
        node['id'] = new_id
        node['order'] = node.get('order', new_id)
    
    # 3. 标准化连接格式
    normalized_links = []
    for link in links:
        if isinstance(link, list):
            # 数组格式：[link_id, src_id, src_slot, tgt_id, tgt_slot, type]
            link_id, src_id, src_slot, tgt_id, tgt_slot, link_type = link
            # 更新节点ID引用
            src_id = node_id_map.get(src_id, src_id)
            tgt_id = node_id_map.get(tgt_id, tgt_id)
            normalized_links.append([link_id, src_id, src_slot, tgt_id, tgt_slot, link_type])
        elif isinstance(link, dict):
            # 对象格式：转换为数组格式
            link_id = link.get('id', len(normalized_links) + 1)
            src_id = node_id_map.get(link.get('source'), link.get('source'))
            src_slot = link.get('source_slot', 0)
            tgt_id = node_id_map.get(link.get('target'), link.get('target'))
            tgt_slot = link.get('target_slot', 0)
            link_type = link.get('type', 'dataframe')
            normalized_links.append([link_id, src_id, src_slot, tgt_id, tgt_slot, link_type])
    
    litegraph['links'] = normalized_links
    
    # 4. 补充缺失的必需字段
    for node in nodes:
        # 确保有flags字段
        if 'flags' not in node:
            node['flags'] = {}
        if 'uuid' not in node['flags']:
            import uuid
            node['flags']['uuid'] = str(uuid.uuid4())
        
        # 确保有order字段
        if 'order' not in node:
            node['order'] = node.get('id', 0)
        
        # 确保有mode字段
        if 'mode' not in node:
            node['mode'] = 0
        
        # CodeControl节点确保有widgets_values
        if node.get('type') == 'CodeControl':
            if 'widgets_values' not in node:
                # 从properties.策略代码复制
                code = node.get('properties', {}).get('策略代码', '')
                node['widgets_values'] = [code] if code else []
    
    # 5. 清理冗余的顶层结构
    if 'nodes' in workflow and 'litegraph' in workflow:
        del workflow['nodes']
    if 'links' in workflow and 'litegraph' in workflow:
        del workflow['links']
    
    return workflow

# 使用示例
if __name__ == "__main__":
    import sys
    
    input_file = sys.argv[1] if len(sys.argv) > 1 else '工作流.json'
    output_file = sys.argv[2] if len(sys.argv) > 2 else input_file.replace('.json', '_标准化.json')
    
    with open(input_file, 'r', encoding='utf-8') as f:
        workflow = json.load(f)
    
    normalized = normalize_workflow(workflow)
    
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(normalized, f, indent=2, ensure_ascii=False)
    
    print(f"标准化完成: {output_file}")

使用场景：

• 合并来自不同版本导出的工作流
• 修复格式不规范的JSON文件
• 批量处理多个工作流文件
• 迁移旧版本工作流到新版本

验证脚本：

def validate_workflow_structure(workflow: Dict) -> tuple[bool, List[str]]:
    """
    验证工作流结构完整性
    
    返回:
        (是否有效, 错误列表)
    """
    errors = []
    
    # 检查必需字段
    if 'litegraph' not in workflow:
        errors.append("缺少litegraph字段")
        return False, errors
    
    litegraph = workflow['litegraph']
    nodes = litegraph.get('nodes', [])
    links = litegraph.get('links', [])
    
    # 检查节点ID唯一性
    node_ids = [node.get('id') for node in nodes]
    if len(node_ids) != len(set(node_ids)):
        errors.append("节点ID不唯一")
    
    # 检查连接有效性
    node_id_set = set(node_ids)
    for link in links:
        if isinstance(link, list) and len(link) >= 5:
            src_id, tgt_id = link[1], link[3]
            if src_id not in node_id_set:
                errors.append(f"连接引用不存在的源节点: {src_id}")
            if tgt_id not in node_id_set:
                errors.append(f"连接引用不存在的目标节点: {tgt_id}")
    
    return len(errors) == 0, errors