Python 处理（自定义算子）¶
使用 Python 编写自定义处理逻辑，将上游数据转换为后续节点需要的结构。
适用于字段转换、文本计算或其它需要代码表达的处理。Python 通过“自定义算子”入口创建，发布后以算子名称出现在自定义节点列表中。
创建 Python 算子¶
在工作流编辑器的节点面板中选择“新建算子”，将实现方式设为“Python 代码”，适用场景选择“普通工作流”。
配置 |
说明 |
|---|---|
名称 |
算子显示名称，例如“文本长度统计” |
标识 |
使用英文、数字和下划线，不能以数字开头，例如 |
功能描述 |
说明处理什么输入、返回什么结果 |
语言 |
选择 Python |
处理函数 |
默认填写 |
版本 |
标识当前算子版本，创建表单默认 |
输入 Schema |
定义输入字段、类型、必填要求及字段说明 |
输出 Schema |
定义返回字段、类型、必填要求及字段说明 |
代码 |
实现与输入输出约定一致的 Python 函数 |
填写字段级说明，便于理解和绑定数据。配置完成后发布算子，再从自定义节点列表选择相应算子和版本加入工作流。
编写处理函数¶
普通工作流使用以下函数签名：
def handle(workspace_id, sdk, input):
return {"length": len(input["text"])}
workspace_id：当前执行所在的工作区标识。sdk：运行时提供的 Python SDK 客户端。input：工作流传给当前算子的输入字典。
使用 return 返回符合输出 Schema 的字典；print 用于输出日志。字典中的值需要支持 JSON 序列化，可使用字符串、数字、列表和对象等类型。
绑定输入与输出¶
执行前会按输入 Schema 校验数据，执行后按输出 Schema 校验返回值。字段缺失、类型不匹配、代码运行异常或输出无法序列化，都会导致节点失败。
例如输入字段声明为 text，就把上游解析节点的正文绑定到该字段。文件 ID 只是引用，不会自动变成文件正文；需要处理正文时，先连接解析节点。
下游直接绑定返回字典中的字段。例如上例返回 length，后续节点可使用该字段取得字符数。
示例：统计文本字符数¶
以下示例展示如何创建一个文本字符数统计算子。
创建“文本长度统计”，标识设为
text_counter，处理函数使用main.handle。在输入 Schema 中添加必填字符串字段
text，说明为“待处理文本”。在输出 Schema 中添加必填整数字段
length，说明为“文本字符数”。使用上面的 Python 代码并发布。
将节点加入工作流，把上游正文绑定到
text，在运行结果中检查length。
输入 {"text": "你好 MOI"} 时，函数计算结果为 {"length": 6}。这里统计 Python 字符串长度，不是字节数或模型 Token 数。
运行条件¶
运行前需要启用 Python 自定义算子执行服务。节点无法调度时，请联系管理员检查该服务。
代码在服务端运行，使用服务端安装的依赖和可访问文件。网络访问与执行超时由服务端配置控制。