Dify 集成-代码执行集成

1. 集成概述

Dify 提供安全的代码执行沙箱环境，支持在工作流中执行 Python 和 JavaScript 代码，实现复杂的数据处理和业务逻辑。

核心价值

• 安全隔离: 沙箱环境隔离代码执行
• 多语言支持: Python 3 和 JavaScript (Node.js)
• 变量传递: 支持工作流变量的输入输出
• 错误处理: 完善的错误捕获和报告机制

2. 支持的服务/产品

2.1 代码执行模式对比

3. 集成方式

3.1 架构设计

3.2 配置方式

# 沙箱服务配置
CODE_EXECUTION_ENDPOINT=http://sandbox:8194
CODE_EXECUTION_API_KEY=dify-sandbox

# 最大执行时间（秒）
CODE_MAX_EXECUTION_TIME=30

# 代码执行模式
CODE_EXECUTION_MODE=sandbox  # 或 local

4. 代码实现

4.1 核心代码路径

📌 源代码引用api/core/workflow/nodes/code/

api/core/workflow/nodes/code/
├── code_node.py                # 代码节点
├── entities.py                 # 实体定义
└── exc.py                      # 异常定义

api/core/helper/code_executor/
├── code_executor.py            # 代码执行器
├── python3/                    # Python 提供商
│   ├── python3_code_provider.py
│   └── python3_transformer.py
└── javascript/                 # JavaScript 提供商
    ├── javascript_code_provider.py
    └── javascript_transformer.py

4.2 代码节点定义

📌 源代码引用api/core/workflow/nodes/code/code_node.py

classCodeNode(Node):
"""代码节点"""

    node_type = NodeType.CODE

def_run(self) -> NodeRunResult:
# 获取代码语言和代码
        code_language = self._node_data.code_language
        code = self._node_data.code

# 准备变量
        variables = {}
for variable_selector inself._node_data.variables:
            variable_name = variable_selector.variable
            variable = self.graph_runtime_state.variable_pool.get(variable_selector.value_selector)
ifisinstance(variable, ArrayFileSegment):
                variables[variable_name] = [v.to_dict() for v in variable.value] if variable.value elseNone
else:
                variables[variable_name] = variable.to_object() if variable elseNone

# 执行代码
try:
            result = CodeExecutor.execute_workflow_code_template(
                language=code_language,
                code=code,
                inputs=variables,
            )

# 验证输出
            result = self._transform_result(result=result, output_schema=self._node_data.outputs)
except (CodeExecutionError, CodeNodeError) as e:
return NodeRunResult(
                status=WorkflowNodeExecutionStatus.FAILED,
                inputs=variables,
                error=str(e),
                error_type=type(e).__name__
            )

return NodeRunResult(
            status=WorkflowNodeExecutionStatus.SUCCEEDED,
            inputs=variables,
            outputs=result
        )

5. 使用示例

5.1 Python 代码执行

⚠️ 概念性示例（简化版，实际使用中需要在工作流节点中调用）

from core.helper.code_executor import CodeExecutor, CodeLanguage

# Python 代码模板
code = """
def main(arg1, arg2):
    # 数据处理
    result = arg1 + arg2
    return {
        'result': result,
        'message': f'Sum is {result}'
    }
"""

# 执行代码
result = CodeExecutor.execute_workflow_code_template(
    language=CodeLanguage.PYTHON3,
    code=code,
    inputs={
'arg1': 10,
'arg2': 20
    }
)

print(result)  # {'result': 30, 'message': 'Sum is 30'}

5.2 JavaScript 代码执行

⚠️ 概念性示例

# JavaScript 代码模板
code = """
function main(arg1, arg2) {
    // 数据处理
    const result = arg1 + arg2;
    return {
        result: result,
        message: `Sum is ${result}`
    };
}
"""

# 执行代码
result = CodeExecutor.execute_workflow_code_template(
    language=CodeLanguage.JAVASCRIPT,
    code=code,
    inputs={
'arg1': 10,
'arg2': 20
    }
)

5.3 工作流中使用

⚠️ 概念性示例（工作流节点配置）

{
"node_id":"code_1",
"node_type":"code",
"title":"数据处理",
"code_language":"python3",
"code":"def main(data):\n    return {'processed': data.upper()}",
"variables":[
{
"variable":"data",
"value_selector":["start","input"]
}
],
"outputs":{
"processed":"string"
}
}

5.4 复杂数据处理示例

⚠️ 概念性示例

# 处理 JSON 数据
code = """
import json

def main(json_data):
    data = json.loads(json_data)

    # 数据转换
    processed = []
    for item in data:
        processed.append({
            'id': item['id'],
            'name': item['name'].upper(),
            'score': item['score'] * 2
        })

    return {
        'processed_data': json.dumps(processed),
        'count': len(processed)
    }
"""

result = CodeExecutor.execute_workflow_code_template(
    language=CodeLanguage.PYTHON3,
    code=code,
    inputs={
'json_data': '[{"id":1,"name":"test","score":10}]'
    }
)

5.5 HTTP 请求示例

⚠️ 概念性示例（沙箱模式下支持）

# Python 代码可以发起 HTTP 请求（在沙箱中）
code = """
import requests

def main(url):
    response = requests.get(url)
    return {
        'status_code': response.status_code,
        'content': response.text[:100]  # 前100字符
    }
"""

result = CodeExecutor.execute_workflow_code_template(
    language=CodeLanguage.PYTHON3,
    code=code,
    inputs={
'url': 'https://api.example.com/data'
    }
)

6. 错误处理

6.1 常见错误类型

⚠️ 概念性示例

📌 异常类定义：

• CodeExecutionError
  : api/core/helper/code_executor/code_executor.py
• CodeNodeError
  : api/core/workflow/nodes/code/exc.py

from core.helper.code_executor import CodeExecutionError
from core.workflow.nodes.code.exc import (
    CodeNodeError,
    DepthLimitError,
    OutputValidationError,
)

try:
    result = CodeExecutor.execute_workflow_code_template(...)
except CodeExecutionError as e:
# 代码执行错误（语法错误、运行时错误）
    logger.error(f"Code execution failed: {e}")
except DepthLimitError as e:
# 递归深度限制
    logger.error(f"Recursion depth exceeded: {e}")
except OutputValidationError as e:
# 输出验证失败
    logger.error(f"Output validation failed: {e}")

6.2 超时处理

⚠️ 概念性示例

# 代码执行会自动处理超时
# 超时时间由配置决定
try:
    result = CodeExecutor.execute_workflow_code_template(
        language=CodeLanguage.PYTHON3,
        code=long_running_code,
        inputs=inputs
    )
except TimeoutError:
    logger.error("Code execution timeout")

7. 安全限制

7.1 沙箱模式限制

• 网络访问受限（仅允许列表内的域名）
• 文件系统访问受限
• 内存使用限制
• CPU 时间限制
• 禁止访问系统资源

7.2 本地模式限制

⚠️ 概念性说明（本地模式使用 RestrictedPython 限制访问）

# 示例：限制访问某些内置函数和模块
ALLOWED_MODULES = ['json', 'math', 'datetime']
FORBIDDEN_FUNCTIONS = ['eval', 'exec', 'open', '__import__']

7.3 代码验证

⚠️ 概念性示例（实际代码执行器内部会进行验证）

from core.helper.code_executor import validate_python_code

# 验证代码安全性
try:
    validate_python_code(code)
except ValueError as e:
    logger.error(f"Code validation failed: {e}")

8. 性能优化

8.1 代码缓存

⚠️ 概念性示例

# 缓存编译后的代码
from functools import lru_cache

@lru_cache(maxsize=100)
defcompile_code(code, language):
if language == CodeLanguage.PYTHON3:
returncompile(code, '<string>', 'exec')
# ...

8.2 沙箱连接池

⚠️ 概念性示例

# 复用沙箱连接
import requests
from requests.adapters import HTTPAdapter

session = requests.Session()
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=20)
session.mount('http://', adapter)
session.mount('https://', adapter)

9. 监控与日志

⚠️ 概念性示例

import logging
import time

logger = logging.getLogger(__name__)

defexecute_with_monitoring(language, code, inputs):
    start_time = time.time()

try:
        result = CodeExecutor.execute_workflow_code_template(
            language=language,
            code=code,
            inputs=inputs
        )

        latency = time.time() - start_time
        logger.info(
f"Code executed - "
f"Language: {language}, "
f"Latency: {latency:.2f}s"
        )

return result
except Exception as e:
        logger.error(f"Code execution failed: {e}")
raise

10. 测试

⚠️ 概念性示例

import pytest

deftest_python_execution():
    code = """
def main(x, y):
    return {'sum': x + y}
"""

    result = CodeExecutor.execute_workflow_code_template(
        language=CodeLanguage.PYTHON3,
        code=code,
        inputs={'x': 1, 'y': 2}
    )

assert result['sum'] == 3

deftest_javascript_execution():
    code = """
function main(x, y) {
    return {sum: x + y};
}
"""

    result = CodeExecutor.execute_workflow_code_template(
        language=CodeLanguage.JAVASCRIPT,
        code=code,
        inputs={'x': 1, 'y': 2}
    )

assert result['sum'] == 3

11. 最佳实践

11.1 代码编写

• 使用 main 函数作为入口
• 返回字典类型的结果
• 处理可能的异常
• 避免无限循环
• 限制输出大小

11.2 性能考虑

• 避免复杂的计算
• 减少网络请求
• 优化算法复杂度
• 考虑使用缓存

11.3 安全性

• 验证输入数据
• 不要在代码中硬编码敏感信息
• 使用沙箱模式而非本地模式
• 限制执行时间

11.4 部署沙箱服务

11.4.1 Docker 部署

# 使用官方沙箱镜像
docker run -d \
  --name dify-sandbox \
  -p 8194:8194 \
  -e API_KEY=dify-sandbox \
  langgenius/dify-sandbox:latest

11.4.2 配置连接

# .env 配置
CODE_EXECUTION_ENDPOINT=http://sandbox:8194
CODE_EXECUTION_API_KEY=dify-sandbox