# Python 处理（自定义算子）

使用 Python 编写自定义处理逻辑，将上游数据转换为后续节点需要的结构。

适用于字段转换、文本计算或其它需要代码表达的处理。Python 通过“自定义算子”入口创建，发布后以算子名称出现在自定义节点列表中。

## 创建 Python 算子

在工作流编辑器的节点面板中选择“新建算子”，将实现方式设为“Python 代码”，适用场景选择“普通工作流”。

| 配置 | 说明 |
| --- | --- |
| 名称 | 算子显示名称，例如“文本长度统计” |
| 标识 | 使用英文、数字和下划线，不能以数字开头，例如 `text_counter` |
| 功能描述 | 说明处理什么输入、返回什么结果 |
| 语言 | 选择 Python |
| 处理函数 | 默认填写 `main.handle`，对应代码中的 `handle` 函数 |
| 版本 | 标识当前算子版本，创建表单默认 `v1` |
| 输入 Schema | 定义输入字段、类型、必填要求及字段说明 |
| 输出 Schema | 定义返回字段、类型、必填要求及字段说明 |
| 代码 | 实现与输入输出约定一致的 Python 函数 |

填写字段级说明，便于理解和绑定数据。配置完成后发布算子，再从自定义节点列表选择相应算子和版本加入工作流。

## 编写处理函数

普通工作流使用以下函数签名：

```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`，后续节点可使用该字段取得字符数。

## 示例：统计文本字符数

以下示例展示如何创建一个文本字符数统计算子。

1. 创建“文本长度统计”，标识设为 `text_counter`，处理函数使用 `main.handle`。
2. 在输入 Schema 中添加必填字符串字段 `text`，说明为“待处理文本”。
3. 在输出 Schema 中添加必填整数字段 `length`，说明为“文本字符数”。
4. 使用上面的 Python 代码并发布。
5. 将节点加入工作流，把上游正文绑定到 `text`，在运行结果中检查 `length`。

输入 `{"text": "你好 MOI"}` 时，函数计算结果为 `{"length": 6}`。这里统计 Python 字符串长度，不是字节数或模型 Token 数。

## 运行条件

运行前需要启用 Python 自定义算子执行服务。节点无法调度时，请联系管理员检查该服务。

代码在服务端运行，使用服务端安装的依赖和可访问文件。网络访问与执行超时由服务端配置控制。

## 继续阅读

- [代码处理](index.md)
- [SQL 处理](sql-processing.md)
- [变量与数据传递](../../variables-and-data.md)
- [运行与调试](../../run-debug.md)