第 9 期：代码节点 — 在 Workflow 中执行 Python/JS

Code 节点概述

Code 节点是 Dify Workflow 中最灵活的瑞士军刀，支持在安全沙盒中执行 Python 3 或 JavaScript 代码。

graph LR
    Input["上游节点输出
(变量绑定)"] --> Code["Code 节点
(沙盒执行)"]
    Code --> Output["结构化输出
(dict/object)"]

Python 代码节点

基本模板

def main(arg1: str, arg2: int) -> dict:
    """
    参数: 从上游节点绑定的变量
    返回: 必须是 dict，key 作为输出变量名
    """
    result = arg1.upper() + str(arg2)
    return {
        "output_text": result,
        "length": len(result)
    }

实例：JSON 数据清洗

import json
import re

def main(raw_llm_output: str) -> dict:
    """从 LLM 的混杂输出中提取 JSON"""
    # LLM 可能输出 ```json ... ``` 包裹的内容
    json_match = re.search(r'```json\s*(.*?)\s*```', raw_llm_output, re.DOTALL)
    
    if json_match:
        clean_json = json_match.group(1)
    else:
        clean_json = raw_llm_output
    
    try:
        data = json.loads(clean_json)
        return {
            "parsed_data": data,
            "success": True,
            "error": ""
        }
    except json.JSONDecodeError as e:
        return {
            "parsed_data": {},
            "success": False,
            "error": str(e)
        }

实例：数据统计与排序

def main(items: list) -> dict:
    """对评分数据进行统计分析"""
    if not items:
        return {"avg": 0, "max_item": None, "sorted_items": []}
    
    scores = [item["score"] for item in items]
    avg_score = sum(scores) / len(scores)
    max_item = max(items, key=lambda x: x["score"])
    sorted_items = sorted(items, key=lambda x: x["score"], reverse=True)
    
    return {
        "average_score": round(avg_score, 2),
        "max_item": max_item,
        "top_3": sorted_items[:3],
        "total_count": len(items)
    }

JavaScript 代码节点

function main(userInput, threshold) {
    // 参数类型自动推断
    const words = userInput.split(/\s+/);
    const wordCount = words.length;
    
    return {
        word_count: wordCount,
        exceeds_limit: wordCount > threshold,
        truncated: wordCount > threshold 
            ? words.slice(0, threshold).join(' ') + '...'
            : userInput
    };
}

可用标准库

安全限制：不支持 import os、import subprocess、网络请求等系统操作。如需调用外部服务，请使用 HTTP Request 节点。

错误处理策略

graph TB
    Code[Code 节点执行] --> Check{执行结果}
    Check -->|成功| Next[下游节点]
    Check -->|异常| Error[错误处理]
    Error --> Retry[可配置重试]
    Error --> Default[使用默认值]
    Error --> Abort[终止工作流]

def main(data: str) -> dict:
    try:
        result = process_data(data)
        return {"status": "success", "data": result}
    except ValueError as e:
        # 返回错误信息而不是抛出异常
        return {"status": "error", "data": None, "error": str(e)}