Skip to content

Commit 68f41bb

Browse files
SunnyHazeSunnyHaze
committed
[update] add batch inference guide
1 parent c9efe99 commit 68f41bb

File tree

8 files changed

+270
-1
lines changed

8 files changed

+270
-1
lines changed

‎docs/.vuepress/notes/en/guide.ts‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,17 @@ export const Guide: ThemeNote = defineNoteConfig({
4848
],
4949

5050
},
51-
51+
{
52+
text: 'Preview Features',
53+
collapsed: false,
54+
icon: 'carbon:idea',
55+
prefix: 'new_feature',
56+
items: [
57+
"resume",
58+
"batch"
59+
],
60+
},
61+
5262
// {
5363
// text: 'Dataflow Agent',
5464
// collapsed: false,

‎docs/.vuepress/notes/zh/guide.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,16 @@ export const Guide: ThemeNote = defineNoteConfig({
5555
'speech_transcription',
5656
],
5757
},
58+
{
59+
text: '内测功能',
60+
collapsed: false,
61+
icon: 'carbon:idea',
62+
prefix: 'new_feature',
63+
items: [
64+
"resume",
65+
"batch"
66+
],
67+
},
5868

5969
{
6070
text:"流水线教程",

‎docs/en/notes/guide/new_feature/batch.md‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
---
2+
title: Preview-Batched Inference
3+
createTime: 2025/12/30 11:47:05
4+
permalink: /en/guide/batch/
5+
---
6+
7+
8+
# Beta: Checkpoint Resume
9+
10+
> This feature is currently in beta and may contain bugs. If you encounter any issues, please report them via issues. Thank you for your understanding.
11+
12+
## Overview
13+
14+
During inference, if an operator holds a large amount of data—such as thousands of records—and an unexpected interruption occurs midway, the portion that has already been inferred will be lost, resulting in wasted API calls.
15+
16+
To address this problem, we designed a **batched inference** interface with the following workflow:
17+
18+
1. Start inference for the first operator
19+
2. The first operator processes only one batch at a time
20+
3. The output of the current step is stored in an appendable output step file (e.g., JSONL or CSV)
21+
4. Once the entire dataset has been processed by the current operator, proceed to the next operator
22+
5. The entire pipeline is completed using this approach
23+
24+
## Usage
25+
26+
This is similar to [Framework Design – Resume Pipeline](/en/guide/basicinfo/framework/#breakpoint-resume-pipelines-resume), except that the pipeline must inherit from another base class, `BatchedPipelineABC`. It also needs to be compiled, and the `forward` function requires additional parameters. In addition, `Storage` currently needs to inherit from a special `BatchedFileStorage` class.
27+
28+
```python
29+
import re
30+
from dataflow.pipeline import BatchedPipelineABC # [!code highlight]
31+
from dataflow.operators.general_text import (
32+
LLMLanguageFilter,
33+
)
34+
from dataflow.operators.text_pt import MetaSampleEvaluator
35+
from dataflow.operators.core_text import PromptedGenerator
36+
from dataflow.serving import APILLMServing_request
37+
from dataflow.utils.storage import BatchedFileStorage # [!code highlight]
38+
39+
class AutoOPPipeline(BatchedPipelineABC): # [!code highlight]
40+
41+
def __init__(self):
42+
super().__init__()
43+
self.storage = BatchedFileStorage( # [!code highlight]
44+
first_entry_file_name="./dataflow/example/GeneralTextPipeline/pt_input.jsonl",
45+
cache_path="./cache_autoop",
46+
file_name_prefix="dataflow_cache_auto_run",
47+
cache_type="jsonl",
48+
)
49+
self.llm_serving = APILLMServing_request(
50+
api_url="http://api.openai.com/v1/chat/completions",
51+
model_name="gpt-4o",
52+
max_workers=30
53+
)
54+
self.op1 = PromptedGenerator(
55+
llm_serving=self.llm_serving1,
56+
system_prompt="Please translate the following content into Chinese:",
57+
)
58+
self.op2 = PromptedGenerator(
59+
llm_serving=self.llm_serving1,
60+
system_prompt="Please translate the following content into Korean:",
61+
)
62+
self.op3 = PromptedGenerator(
63+
llm_serving=self.llm_serving1,
64+
system_prompt="Please translate the following content into Japanese:"
65+
)
66+
67+
def forward(self):
68+
self.op1.run(
69+
self.storage.step(),
70+
input_key='raw_content',
71+
output_key='content_cn1'
72+
)
73+
self.op2.run(
74+
self.storage.step(),
75+
input_key='raw_content',
76+
output_key='content_cn2'
77+
)
78+
self.op3.run(
79+
self.storage.step(),
80+
input_key='raw_content',
81+
output_key='content_cn3'
82+
)
83+
84+
if __name__ == "__main__":
85+
pipeline = AutoOPPipeline()
86+
pipeline.compile() # [!code highlight]
87+
pipeline.forward(
88+
batch_size=2, # [!code highlight]
89+
resume_from_last=True
90+
)
91+
```
92+
93+
`resume_from_last` allows the pipeline to automatically continue from the last step file found in the current cache path.
94+
95+
Alternatively, you can use `resume_step` to resume from a specific previous operator step. Note that **only one of these parameters can be set at a time** to avoid logical conflicts.
96+
97+
```python
98+
# The following invocation is also valid
99+
...
100+
pipeline.compile()
101+
pipeline.forward(
102+
batch_size=2,
103+
resume_step=2 # Resume from the operator with index 2; the first operator has index 0
104+
)
105+
```
106+
107+
---

‎docs/en/notes/guide/new_feature/resume.md‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
---
2+
title: Preview-Checkpoint Resume
3+
createTime: 2025/12/30 11:19:40
4+
permalink: /en/guide/resume/
5+
---
6+
7+
8+
# Beta: Checkpoint Resume
9+
10+
> This feature is currently in beta and may contain bugs. If you encounter any issues, please report them via issues. Thank you for your understanding.
11+
12+
## Overview
13+
14+
During inference, if the process is interrupted at a certain operator, the default workflow requires rerunning the entire pipeline from the previous operator. This typically involves commenting out all preceding operators in the pipeline and manually renaming intermediate cached step files as the new entry point, which is cumbersome and error-prone.
15+
16+
To address this, we provide a feature that allows inference to be resumed directly from a specific operator step.
17+
18+
## Usage
19+
20+
Please refer to the section
21+
[Framework Design – Resume Pipelines](/en/guide/basicinfo/framework/#breakpoint-resume-pipelines-resume) for details.

‎docs/zh/notes/guide/new_feature/batch.md‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
---
2+
title: 内测：Batch化推理
3+
createTime: 2025/12/30 11:47:05
4+
permalink: /zh/guide/batch/
5+
---
6+
# 内测：断点Resume
7+
> 此功能处于内测阶段，可能有bug，发现问题请及时通过issue反馈，感谢理解
8+
9+
## 概述
10+
推理时，如果一个算子持有大量数据，比如几千条数据，中间发生了异常中断，这几千条中已经推理的部分会原地丢失，相应的API也会被浪费掉。
11+
12+
为了解决这个问题，我们设计了Batch化推理的接口，流程如下：
13+
14+
1. 开始推理第一个算子
15+
2. 第一个算子一次只给输入一个batch
16+
3. 当前step输出的结果存放在一个可以append的输出step文件中（比如jsonl或者csv）
17+
4. 直到整个数据集的数据都在当前算子推理完成，继续推理下一个算子。
18+
5. 整个pipeline都按照这个方式推理完成
19+
20+
## 使用方法
21+
与[框架设计-resume流水线](/zh/guide/basicinfo/framework/#断点恢复流水线-resume)相似，不过pipeline需要继承另一个基类`BatchedPipelineABC`来实现，也同样需要compile，并且在forward函数中传入更多的形参。此外目前`Storage`也需要继承另一个特殊的`BatchedFileStorage`类。
22+
```python
23+
import re
24+
from dataflow.pipeline import BatchedPipelineABC # [!code highlight]
25+
from dataflow.operators.general_text import (
26+
LLMLanguageFilter,
27+
)
28+
from dataflow.operators.text_pt import MetaSampleEvaluator
29+
from dataflow.operators.core_text import PromptedGenerator
30+
from dataflow.serving import APILLMServing_request
31+
from dataflow.utils.storage import BatchedFileStorage # [!code highlight]
32+
33+
class AutoOPPipeline(BatchedPipelineABC): # [!code highlight]
34+
35+
def __init__(self):
36+
super().__init__()
37+
self.storage = BatchedFileStorage( # [!code highlight]
38+
first_entry_file_name="./dataflow/example/GeneralTextPipeline/pt_input.jsonl",
39+
cache_path="./cache_autoop",
40+
file_name_prefix="dataflow_cache_auto_run",
41+
cache_type="jsonl",
42+
)
43+
self.llm_serving = APILLMServing_request(
44+
api_url="http://api.openai.com/v1/chat/completions",
45+
model_name="gpt-4o",
46+
max_workers=30
47+
)
48+
self.op1 = PromptedGenerator(
49+
llm_serving=self.llm_serving1,
50+
system_prompt="请将以下内容翻译成中文：",
51+
)
52+
self.op2 = PromptedGenerator(
53+
llm_serving=self.llm_serving1,
54+
system_prompt="请将以下内容翻译成韩文：",
55+
)
56+
self.op3 = PromptedGenerator(
57+
llm_serving=self.llm_serving1,
58+
system_prompt="请将以下内容翻译成日语："
59+
)
60+
61+
def forward(self):
62+
self.op1.run(
63+
self.storage.step(),
64+
input_key='raw_content',
65+
output_key='content_cn1'
66+
)
67+
self.op2.run(
68+
self.storage.step(),
69+
input_key='raw_content',
70+
output_key='content_cn2'
71+
)
72+
self.op3.run(
73+
self.storage.step(),
74+
input_key='raw_content',
75+
output_key='content_cn3'
76+
)
77+
78+
if __name__ == "__main__":
79+
pipeline = AutoOPPipeline()
80+
pipeline.compile() # [!code highlight]
81+
pipeline.forward(
82+
batch_size=2, # [!code highlight]
83+
resume_from_last=True
84+
)
85+
```
86+
87+
`resume_from_last`可以直接从当前cache路径下读取到的最后一个step文件，自动继续训练。
88+
特别的，也可以使用`resume_step`来从具体的一个之前的某一个具体的算子step恢复。不过这两个形参只能设置其中一个避免逻辑冲突。
89+
```python
90+
# 以下调用方式也是可以的
91+
...
92+
pipeline.compile()
93+
pipeline.forward(
94+
batch_size=2,
95+
resume_step=2 # 从index为2的算子恢复，第一个算子的index是0
96+
)
97+
```
98+

‎docs/zh/notes/guide/new_feature/resume.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
---
2+
title: 内测：断点恢复resume
3+
createTime: 2025/12/30 11:19:40
4+
permalink: /zh/guide/resume/
5+
---
6+
# 内测：断点Resume
7+
> 此功能处于内测阶段，可能有bug，发现问题请及时通过issue反馈，感谢理解
8+
9+
## 概述
10+
当推理时，于某个算子断掉了，此时按照默认流程，想从前一个算子重新跑整个流程，要把整个pipeline前面的算子注释掉，并且修改中间缓存的step文件的名称作为入口，这很麻烦。所以我们提供了一个可以从算子step进行恢复推理的功能。
11+
12+
## 使用方法
13+
请参考[框架设计-resume流水线](/zh/guide/basicinfo/framework/#断点恢复流水线-resume)章节。

‎docs/zh/notes/guide/quickstart/write_operator.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
title: write_operator
3+
createTime: 2025/12/29 11:05:15
4+
permalink: /zh/guide/livgzy4s/
5+
---

‎docs/zh/notes/guide/quickstart/write_pipeline.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
title: write_pipeline
3+
createTime: 2025/12/29 11:05:05
4+
permalink: /zh/guide/r8flkmy0/
5+
---

0 commit comments

Comments
 (0)