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.pyCodeNodeError: 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 沙箱模式限制
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 代码编写
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
