# 数据加工

数据[接入并登记](organize.md)后仍是原始形态——PDF、Word、音视频无法被直接检索。加工将它们解析、分块、向量化为 AI 可检索的数据；承担加工的机制是工作流。

工作流（Workflow）是 MOI 中定义数据加工过程的基本单位：一组算子（WorkItem），以及数据在算子之间的流转顺序。解析、清洗、分块、向量化，每个加工步骤由一个算子执行；MOI 按工作流定义调度算子，并在算子之间传递数据。

一条工作流的一生：被定义，被发布，被启动；运行中，数据沿定义在节点间流转；成功或失败，都留下一条运行记录。本页按这条生命线展开，示例采用平台内置的文档入库工作流 `catalog-parse-index-lineage`——从源文件到向量索引的完整加工链：

| 节点 | 算子 | 作用 |
|---|---|---|
| read_source | `moi:catalog.source.read.v2` | 按源引用读取文件清单 |
| parse_documents | `moi:document.parse` | 将文件解析为结构化文本 |
| split_documents | `moi:parser.split.documents.length` | 将文本按长度分块 |
| build_knowledge_index | `moi:knowledge.index.build` | 将分块向量化，写入向量表 |
| write_parsed_documents | `moi:files.write_documents` | 将解析产物写为文件 |
| save_result | `moi:catalog.sink.write` | 将结果写入目标表 |
| register_lineage | `moi:data.lineage.register` | 登记产物与源文件的血缘 |

## 创建与发布

您通过三种方式创建工作流，产物等价，均为同一份工作流定义：

| 方式 | 说明 |
|---|---|
| 画布 | 拖拽算子并连线，可视化编排 |
| 代码 | 直接编写工作流定义文件（YAML） |
| 自然语言 | 对话描述需求，由内置 AI 生成定义；结果可在画布中查看与调整 |

三种方式中，自然语言创建的门槛最低：您描述要加工什么、产出什么，AI 负责选择算子、连接顺序与配置参数。生成的定义与手工编排的定义完全同构，可随时切换到画布继续调整。

您编辑的始终是草稿；发布后，平台按发布版本调度执行。

## 启动

| 方式 | 说明 | 适用 |
|---|---|---|
| 手动运行 | 您直接发起一次执行 | 调试定义、一次性加工 |
| 定时运行 | 按 cron 表达式周期执行 | 周期性批量加工 |
| 文件卷触发 | 工作流绑定到文件卷后，您上传文件，它自动启动 | 无人值守的持续摄取 |

每次启动，平台生成一条运行记录，包含各节点的执行状态、输入输出与日志。

## 运行中的数据传递

您在启动时传入变量：示例工作流的变量是源引用、目标表与嵌入模型。运行开始后，引擎逐节点推进：按定义将输入派发给算子，收取输出，再派发下一个节点。

工作流运行时维护三类数据：

| 类别 | 读写 | 说明 |
|---|---|---|
| 变量（vars） | 全程只读 | 启动时注入的参数，如源引用、目标表名 |
| 状态（state） | 跨节点读写 | 各节点以 `save` 写入的中间结果；单个值上限 1 MB，超限自动转存为文件并保留引用 |
| 节点输出（data） | 每节点执行后整体替换 | 上一节点的直接输出，仅相邻节点可用 |

示例中 parse_documents 与 split_documents 的衔接，即状态传递的标准形态（定义原文节选）：

```yaml
- work_item:
    name: parse_documents
    id: moi:document.parse
    input:
      sources: "{{ .state.sources }}"
      file_ids: "{{ .state.source_file_ids }}"
    save:
      parsed_documents: .documents

- work_item:
    name: split_documents
    id: moi:parser.split.documents.length
    input:
      documents: "{{ .state.parsed_documents }}"
    save:
      chunked_documents: .documents
```

parse_documents 以 `save` 将解析结果存入状态键 `parsed_documents`；split_documents 的输入以 `{{ .state.parsed_documents }}` 引用它。跨越多个节点的传递使用 state；data 不跨节点保留。

工作流数据中不传递文件内容，只传递文件引用——示例中流转的 `file_ids` 即引用，需要文件内容的算子凭引用自行读取。大体量内容直接进入数据通道会压垮传输层，平台已据此禁用将文件正文读入工作流数据的算子。

## 失败排查

引擎只负责调度与数据传递，不校验任务语义；定义中的错误会被原样执行。因此排查失败，先分两类：

| 类型 | 例 | 修复位置 |
|---|---|---|
| 定义问题 | 节点选错、顺序错、参数绑定配置错误 | 修改工作流定义 |
| 执行问题 | 定义正确，算子对合法输入处理失败 | 查看该节点运行记录中的输入与日志 |

一个不报错的定义问题：节点引用了未注册的算子时，该节点持续等待，没有报错信息。可用算子及其输入输出约定，以接口 `GET /workspaces/:id/workitems` 的实时返回为准。

## 使用限制

- 子工作流仅支持在同一工作流定义内展开，不支持跨工作流引用或合并；工作流之间不共享状态，也不存在数据通道。
- SQL 算子的输入仅包含 SQL 语句本身，不支持自然语言转 SQL。

## 下一步

| 您想做的事 | 前往 |
|---|---|
| 把加工产物交给 AI 使用 | [知识库](../ai/knowledge-base.md) |
| 查询某个算子或术语的定义 | [参考](../reference/index.md) |

> 面向 AI 代理的说明：本页示例为平台内置示例 `catalog-parse-index-lineage` 的原文节选；「运行中的数据传递」与「使用限制」为权威约定；可用算子以实时接口为准，不要依据静态列表选择算子。