Dify 代码执行节点代码规范

在使用 Dify 工作流的过程中，代码执行节点是我最常用的节点之一。它让我能在工作流中嵌入自定义的 Python 逻辑，处理一些标准节点无法完成的任务。但起初我经常遇到各种问题：返回值格式不对导致报错、数据类型超出限制被截断、异常分支没有处理导致工作流中断……每次都要反复调试才能找到原因。

后来我意识到，这些问题大多源于对 Dify 代码执行节点的运行机制和限制不够了解。于是我查阅官方文档、踩了无数坑后，总结出了一套相对完善的编码规范。这套规范帮助我显著减少了错误，也让代码更健壮、更易维护。如果你也在使用代码执行节点时遇到过类似困扰，希望这篇笔记能帮到你。

本文将从基本结构、变量命名、数据类型限制、返回结果格式、异常处理、JSON 数据处理等方面，详细介绍代码执行节点的编码规范和最佳实践，并提供大量实际代码示例供参考。

跳转到你感兴趣的章节

  1. 基本结构
  2. 变量命名
  3. 数据类型限制
  4. 返回结果格式要求
  5. 异常处理
  6. 常见报错处理
  7. JSON 数据类型的处理
  8. 更好的代码结构

1. 基本结构

代码执行节点必须定义一个 main 函数作为入口，输入参数对应节点配置中的输入变量，返回值必须是字典（dict），键名对应节点配置中的输出变量。

1
2
3
4
5
6
7
8
9
10
11
def main(input_var1: str, input_var2: list) -> dict:
    import json

    # 处理逻辑
    result = process(input_var1, input_var2)

    # 返回字典，键名必须与输出变量声明一致
    return {
        'output_var1': result,
        'output_var2': 'other_value'
    }

关键规则

  • 入口函数名必须是 main：Dify 只识别 main 函数作为执行入口，当然你可以定义其他工具函数，但是入口Dify只认main函数
  • 输入参数名必须与节点配置一致：Dify就是靠参数名传递数据的
  • 返回值必须是字典：返回其他类型（字符串、列表、数字）将直接报错
  • 返回字典的键名必须与输出变量声明一致：未声明的键会被忽略，缺少声明的键会报错

    tips: 与Python不同的是Python中直接 return 返回值，而Dify中必须返回字典，将返回值作为字典的值，字典的键名必须与输出变量声明一致，数据类型也要保持一致，这种设计让Dify能够一个函数返回多个返回值。

2. 变量命名

良好的变量命名能显著提升代码可读性，尤其在多人协作或后期维护时更为重要。

2.1 命名规范

  • 变量：使用 snake_case，如 user_query、filtered_results、item_count
  • 常量：使用 UPPER_SNAKE_CASE，如 MAX_RETRY、DEFAULT_TIMEOUT

2.2 命名建议

1
2
3
4
5
6
7
8
9
10
11
12
13
# 好的命名：语义清晰
def main(user_query: str, search_results: list) -> dict:
    filtered_results = [r for r in search_results if r['score'] > 0.5]
    return {'filtered_results': filtered_results}

# 不好的命名：含义模糊
def main(q: str, sr: list) -> dict:
    fr = [r for r in sr if r['score'] > 0.5]
    return {'fr': fr}

# 默认的命名：不容易区分不同节点的变量
def main(arg1: str, arg2: str) -> dict:
    return {'result': arg1 + arg2}

3. 数据类型限制

Dify 代码执行节点对输出数据有严格的限制，超出限制的数据会被截断或报错。

3.1 输出值限制

  • String：最大 80,000 字符，null 字节会被自动移除
  • Number：范围 -999999999 ~ 999999999，浮点数最多 10 位小数
  • Object/Array：最大嵌套深度 5 层，超出深度的部分会被截断

3.2 输入变量支持的数据类型

  • String → str：普通文本
  • Number → int / float：整数或浮点数
  • Object → dict：JSON 对象
  • Array → list：JSON 数组
  • Boolean → bool：布尔值

3.3 注意事项

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 数值超出范围会被截断
def main() -> dict:
    big_number = 9999999999  # 超出限制
    return {'result': big_number}  # 会被截断为 999999999

# 浮点数精度限制
def main() -> dict:
    precise = 3.14159265358979323846  # 超过10位小数
    return {'result': precise}  # 会被截断为 3.1415926536

# 嵌套深度限制
def main() -> dict:
    deep_data = {'a': {'b': {'c': {'d': {'e': {'f': 'too deep'}}}}}}  # 第6层，超出限制
    return {'result': deep_data}  # 第6层及更深层会被截断

4. 返回结果格式要求

4.1 必须返回字典

1
2
3
4
5
6
7
# 错误：直接返回字符串
def main(text: str) -> dict:
    return text  # 报错：result must be a dict

# 正确：返回字典
def main(text: str) -> dict:
    return {'result': text}

4.2 键名必须与输出变量声明一致

在节点配置中声明了输出变量后，返回字典的键名必须与之完全匹配：

1
2
3
4
5
6
7
8
9
# 假设节点配置中声明了输出变量：result 和 status

# 正确
def main(data: str) -> dict:
    return {'result': data, 'status': 'success'}

# 错误：键名不匹配
def main(data: str) -> dict:
    return {'output': data, 'state': 'success'}  # output 和 state 未声明，会被忽略

4.3 所有执行路径都必须返回字典

1
2
3
4
5
6
7
8
9
10
11
# 错误：异常分支没有返回值
def main(data: str) -> dict:
    if not data:
        return  # 缺少返回值，后续节点会报 "Output is missing"
    return {'result': data}

# 正确：所有分支都返回字典
def main(data: str) -> dict:
    if not data:
        return {'result': '', 'status': 'empty'}
    return {'result': data, 'status': 'success'}

5. 异常处理

5.1 基本原则

  • 所有可能出错的代码都应包裹在 try-except 中
  • 异常分支也必须返回字典，不能让程序崩溃
  • 异常信息应具有可读性，方便后续排查

5.2 标准异常处理模板

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def main(input_data: str) -> dict:
    try:
        # 核心业务逻辑
        result = process(input_data)
        return {
            'result': result,
            'status': 'success',
            'error': ''
        }
    except Exception as e:
        # 异常时也返回字典，包含错误信息
        return {
            'result': '',
            'status': 'error',
            'error': str(e)
        }

5.3 细粒度异常捕获

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def main(data: str) -> dict:
    try:
        import json
        parsed = json.loads(data)
    except json.JSONDecodeError as e:
        return {'result': None, 'error': f'JSON解析失败: {str(e)}'}
    except TypeError as e:
        return {'result': None, 'error': f'类型错误: {str(e)}'}
    except Exception as e:
        return {'result': None, 'error': f'未知错误: {str(e)}'}

    try:
        value = parsed['key']
    except KeyError:
        return {'result': None, 'error': "缺少必需字段 'key'"}

    return {'result': value, 'error': ''}

5.4 利用 Dify 的错误处理机制

Dify 代码执行节点支持配置级别的错误处理：

  • 无处理：直接抛出异常，中断流程（适用于关键节点，不允许失败）
  • 默认值：异常时使用预定义的默认输出（适用于允许降级的场景）
  • 失败分支：异常时走备选路径（适用于需要备用逻辑的场景）

同时支持重试配置：最大重试 10 次，重试间隔最大 5000ms。

6. 常见报错处理

6.1 常见报错速查

  • result must be a dict：返回值不是字典 → 确保 return 的是字典类型
  • Output is missing：返回字典缺少声明的输出变量键 → 检查所有分支是否都返回了完整的键
  • CodeNodeValidationFailed：输出值超出限制 → 检查字符串长度、数值范围、嵌套深度
  • Maximum execution time exceeded：代码执行超时 → 优化算法，减少循环次数
  • ModuleNotFoundError：导入了沙箱不支持的库 → 使用标准库或已安装的依赖
  • KeyError：访问字典中不存在的键 → 使用 .get() 方法并提供默认值
  • TypeError：类型不匹配 → 检查输入数据类型，必要时进行类型转换
  • json.JSONDecodeError：JSON 字符串格式错误 → 使用 try-except 捕获，检查原始数据

6.2 典型问题与解决方案

问题1：返回值类型错误

1
2
3
4
5
6
7
8
9
# 错误写法
def main(items: list) -> dict:
    total = sum(items)
    return total  # 返回了 int，不是 dict

# 正确写法
def main(items: list) -> dict:
    total = sum(items)
    return {'total': total}

问题2：异常分支未返回值

1
2
3
4
5
6
7
8
9
10
11
12
# 错误写法
def main(data: str) -> dict:
    if not data:
        print("数据为空")  # 没有返回值
        # 函数隐式返回 None，导致 "Output is missing"
    return {'result': data}

# 正确写法
def main(data: str) -> dict:
    if not data:
        return {'result': '', 'status': 'empty'}
    return {'result': data, 'status': 'success'}

问题3：字典键访问报错

1
2
3
4
5
6
7
# 错误写法
def main(data: dict) -> dict:
    name = data['name']  # 如果 name 不存在则 KeyError

# 正确写法
def main(data: dict) -> dict:
    name = data.get('name', '')  # 不存在时返回空字符串

问题4：导入不支持的库

1
2
3
4
5
6
7
8
9
10
11
12
# 错误写法
def main() -> dict:
    import tensorflow as tf  # 沙箱不支持
    return {'result': 'ok'}

# 正确写法：使用标准库或已安装的依赖
def main() -> dict:
    import json
    import math
    import re
    from datetime import datetime
    return {'result': 'ok'}

你也可以在 sandbox 中安装依赖包然后就可以在代码执行节点调用使用了。

7. JSON 数据类型的处理

JSON 是 Dify 工作流中最常用的数据交换格式，正确处理 JSON 是代码执行节点的核心技能。

7.1 JSON 与 Python 类型映射

  • object → dict：取值用 data['key'] 或 data.get('key')，赋值用 data['key'] = value
  • array → list：取值用 data[0] 或 data[-1]，赋值用 data.append(item)
  • string → str：直接使用
  • number (int) → int：直接使用
  • number (float) → float：直接使用
  • true/false → bool：直接使用
  • null → None：判断用 is None，赋值用 None

7.2 JSON 取值

基本取值

1
2
3
4
5
6
7
8
9
10
11
def main(json_data: dict) -> dict:
    # 安全取值：使用 .get() 避免 KeyError
    name = json_data.get('name', '')
    age = json_data.get('age', 0)
    hobbies = json_data.get('hobbies', [])

    return {
        'name': name,
        'age': age,
        'hobbies': hobbies
    }

嵌套取值

1
2
3
4
5
6
7
8
9
def main(json_data: dict) -> dict:
    # 逐层安全取值
    address = json_data.get('address', {})
    city = address.get('city', '') if isinstance(address, dict) else ''

    # 或者使用更简洁的写法
    city = json_data.get('address', {}).get('city', '')

    return {'city': city}

从字符串解析 JSON

1
2
3
4
5
6
7
8
9
10
11
def main(json_str: str) -> dict:
    import json

    try:
        data = json.loads(json_str)
    except json.JSONDecodeError as e:
        return {'result': None, 'error': f'JSON解析失败: {str(e)}'}

    # 安全取值
    value = data.get('key', '')
    return {'result': value, 'error': ''}

遍历 JSON 数组

1
2
3
4
5
6
7
8
9
def main(items: list) -> dict:
    results = []
    for item in items:
        # 每个元素都安全取值
        name = item.get('name', '') if isinstance(item, dict) else ''
        score = item.get('score', 0) if isinstance(item, dict) else 0
        results.append({'name': name, 'score': score})

    return {'results': results}

7.3 JSON 赋值

构建字典

1
2
3
4
5
6
7
def main(name: str, age: int) -> dict:
    result = {
        'name': name,
        'age': age,
        'tags': ['user', 'active']
    }
    return {'result': result}

修改已有字典

1
2
3
4
5
6
7
def main(data: dict) -> dict:
    # 注意：不要直接修改输入，创建副本
    result = data.copy()
    result['processed'] = True
    result['timestamp'] = '2025-01-01'

    return {'result': result}

序列化为 JSON 字符串

1
2
3
4
5
6
7
8
9
10
def main(data: dict) -> dict:
    import json

    try:
        # ensure_ascii=False 保证中文等非ASCII字符正常显示，不被转义为 \uXXXX
        json_str = json.dumps(data, ensure_ascii=False)
    except TypeError as e:
        return {'result': '', 'error': f'序列化失败: {str(e)}'}

    return {'result': json_str}

tips: 序列化为JSON字符串也是解决输出内容嵌套过多问题的一种方式。

7.4 处理非标准 JSON 类型

Python 的 datetime、set 等类型不是 JSON 标准类型，直接序列化会报错：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def main() -> dict:
    import json
    from datetime import datetime

    data = {'created': datetime.now()}

    # 错误：直接序列化会抛出 TypeError
    # json_str = json.dumps(data)

    # 正确：使用自定义编码器
    class DateTimeEncoder(json.JSONEncoder):
        def default(self, obj):
            if isinstance(obj, datetime):
                return obj.isoformat()
            return super().default(obj)

    json_str = json.dumps(data, cls=DateTimeEncoder, ensure_ascii=False)
    return {'result': json_str}

7.5 JSON 处理最佳实践

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
def main(raw_data: str) -> dict:
    import json

    # 1. 解析时捕获异常
    try:
        data = json.loads(raw_data)
    except json.JSONDecodeError:
        return {'result': None, 'error': '输入不是有效的JSON'}

    # 2. 验证数据类型
    if not isinstance(data, dict):
        return {'result': None, 'error': f'期望dict，实际为{type(data).__name__}'}

    # 3. 使用 .get() 安全取值，提供合理默认值
    items = data.get('items', [])
    if not isinstance(items, list):
        return {'result': None, 'error': 'items字段不是数组'}

    # 4. 处理数据
    processed = []
    for item in items:
        if not isinstance(item, dict):
            continue
        processed.append({
            'name': item.get('name', ''),
            'value': item.get('value', 0)
        })

    # 5. 返回结构化结果
    return {
        'result': processed,
        'count': len(processed),
        'error': ''
    }

8. 更好的代码结构

8.1 模块导入放在函数内部

Dify 沙箱环境要求导入语句放在 main 函数内部，而非全局作用域：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 正确
def main(data: list) -> dict:
    import json
    import math
    from datetime import datetime

    result = math.ceil(sum(data) / len(data))
    return {'result': result}

# 错误
import json  # 不要放在函数外部
import math

def main(data: list) -> dict:
    result = math.ceil(sum(data) / len(data))
    return {'result': result}

8.2 推荐的代码组织模板

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
def main(input_data: str, config: dict) -> dict:
    # ---- 1. 导入依赖 ----
    import json
    import re

    # ---- 2. 参数校验与默认值 ----
    if not input_data:
        return {'result': '', 'status': 'empty', 'error': ''}

    # ---- 3. 数据解析 ----
    try:
        data = json.loads(input_data)
    except json.JSONDecodeError as e:
        return {'result': '', 'status': 'error', 'error': f'JSON解析失败: {str(e)}'}

    # ---- 4. 核心业务逻辑 ----
    try:
        processed = process_data(data, config)
    except Exception as e:
        return {'result': '', 'status': 'error', 'error': f'处理失败: {str(e)}'}

    # ---- 5. 结果校验 ----
    if not isinstance(processed, dict):
        return {'result': '', 'status': 'error', 'error': '处理结果格式异常'}

    # ---- 6. 返回结果 ----
    return {
        'result': processed,
        'status': 'success',
        'error': ''
    }


def process_data(data: dict, config: dict) -> dict:
    """核心处理逻辑抽离为独立函数，提升可读性"""
    # 具体业务逻辑
    return data

8.3 保持代码幂等性

代码执行节点可能因重试机制被多次执行，因此代码应具有幂等性：

1
2
3
4
5
6
7
8
9
10
# 好的做法：相同的输入总是产生相同的输出
def main(count: int) -> dict:
    result = count * 2
    return {'result': result}

# 不好的做法：依赖外部状态或随机性
def main(count: int) -> dict:
    import random
    result = count + random.randint(1, 100)  # 每次执行结果不同
    return {'result': result}

8.4 避免全局变量和状态

1
2
3
4
5
6
7
8
9
10
11
12
# 错误：使用全局变量
counter = 0

def main(data: str) -> dict:
    global counter
    counter += 1  # 沙箱环境可能无法保持状态
    return {'count': counter}

# 正确：基于输入计算，不依赖外部状态
def main(data: list) -> dict:
    count = len(data)
    return {'count': count}

8.5 合理使用辅助函数

当逻辑较复杂时，将核心逻辑抽离为辅助函数，保持 main 函数简洁：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def main(raw_items: list) -> dict:
    validated = validate_items(raw_items)
    filtered = filter_items(validated)
    sorted_items = sort_items(filtered)

    return {
        'items': sorted_items,
        'total': len(sorted_items)
    }


def validate_items(items: list) -> list:
    """过滤无效条目"""
    return [item for item in items if isinstance(item, dict) and 'id' in item]


def filter_items(items: list) -> list:
    """过滤低分条目"""
    return [item for item in items if item.get('score', 0) >= 60]


def sort_items(items: list) -> list:
    """按分数降序排列"""
    return sorted(items, key=lambda x: x.get('score', 0), reverse=True)

8.6 沙箱环境限制清单

编写代码时需牢记以下限制：

  • 禁止文件系统访问：不能读写本地文件
  • 禁止网络请求：不能在代码中调用外部 API
  • 禁止系统命令：不能使用 os.system、subprocess
  • 执行时间限制：通常为 30 秒（视部署配置）
  • 内存限制：通常为 512MB（视部署配置）
  • 可用标准库：json, math, datetime, re, collections 等
  • 可用第三方库：numpy, pandas, requests 等（视部署配置）

如果需要使用不支持的库，可考虑：将复杂计算迁移到外部服务，通过 HTTP 请求节点调用；或在自托管环境中扩展沙箱依赖（在sandbox容器中通过pip install xxx.whl安装依赖）。

附录：完整示例

示例1：安全解析 LLM 输出的 JSON

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
def main(llm_output: str) -> dict:
    import json
    import re

    # 1. 尝试从 LLM 输出中提取 JSON
    json_match = re.search(r'\{[\s\S]*\}', llm_output)
    if not json_match:
        return {'result': None, 'error': '未找到JSON内容'}

    # 2. 解析 JSON
    try:
        data = json.loads(json_match.group())
    except json.JSONDecodeError as e:
        return {'result': None, 'error': f'JSON解析失败: {str(e)}'}

    # 3. 安全取值
    summary = data.get('summary', '')
    keywords = data.get('keywords', [])
    confidence = data.get('confidence', 0.0)

    # 4. 类型校验
    if not isinstance(keywords, list):
        keywords = []

    # 5. 数值范围校验
    confidence = max(0.0, min(1.0, float(confidence)))

    return {
        'summary': summary,
        'keywords': keywords,
        'confidence': confidence,
        'error': ''
    }

示例2：数据聚合与格式转换

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
def main(records: list) -> dict:
    import json

    if not isinstance(records, list):
        return {'result': [], 'error': '输入不是数组'}

    # 按 category 分组统计
    groups = {}
    for record in records:
        if not isinstance(record, dict):
            continue
        category = record.get('category', 'unknown')
        amount = record.get('amount', 0)
        try:
            amount = float(amount)
        except (TypeError, ValueError):
            amount = 0.0

        if category not in groups:
            groups[category] = {'count': 0, 'total': 0.0}
        groups[category]['count'] += 1
        groups[category]['total'] += amount

    # 转换为列表格式
    summary = [
        {'category': k, 'count': v['count'], 'total': round(v['total'], 2)}
        for k, v in groups.items()
    ]

    return {
        'summary': summary,
        'total_categories': len(summary),
        'error': ''
    }

参考文档：Dify 官方文档 - Code Node