# Generating Dataflow Pipeline

> Plan and write a standard DataFlow pipeline from a target and representative JSONL data. Use when a user asks to select DataFlow operators, trace field dependencies, generate runnable pipeline code, or repair a pipeline with schema or field-flow errors.

- Skill: `opendcai/generating-dataflow-pipeline` (Agent Skill, multi-file: 20 files)
- Install (CLI): `npx skillmds@latest add opendcai/generating-dataflow-pipeline`
- Raw SKILL.md: https://api.skillmd.com/api/skills/opendcai/generating-dataflow-pipeline/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: opendcai (https://skillmd.com/u/opendcai)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/opendcai/generating-dataflow-pipeline

---

# DataFlow Pipeline Code Generator

## Goal

This skill is used when users provide:

- **Target**: What the pipeline should achieve
- **Sample Data File**: Path to a JSONL file containing 1-5 representative data samples

The skill must:

1. **Read and analyze the JSONL file** at the provided path
2. Infer data structure, field types, and content characteristics
3. Determine task type based on file content (document processing, text transformation, multi-field composition)
4. Select appropriate operators from preferred primitives
5. Validate field dependencies
6. Output intermediate operator decision summary
7. Generate standard DataFlow pipeline code with `first_entry_file_name` set to the user-provided file path

## User Input Format

Users provide:

```
Target: [Clear task description]
Sample file: [Path to JSONL file, e.g., ./data/input.jsonl]
Expected outputs: [Optional field list]
```

**Important**: The sample file is a JSONL file (one JSON object per line), not a JSON array.

## Preferred Operator Strategy

**Six Core Primitives** (high-coverage operators for most data science tasks):

1. `PromptedGenerator` - Single-field LLM generation
2. `FormatStrPromptedGenerator` - Multi-field template generation
3. `Text2MultiHopQAGenerator` - Multi-hop QA pair construction
4. `PromptedFilter` - LLM-based quality filtering
5. `GeneralFilter` - Rule-based filtering
6. KBC trio (always used together in order): `FileOrURLToMarkdownConverterFlash` → `KBCChunkGenerator` → `KBCTextCleaner`

These are **preferred primitives**, not fixed workflows. They can be used repeatedly and combined flexibly.

## Operator Selection Priority Rule (MANDATORY)

When a specialized operator exists for the task, it MUST be used over generic operators. Do NOT use `PromptedGenerator` to replicate functionality that a dedicated operator already provides.

**Decision table** (check in order, use the first match):

| Task / Scenario                                 | Required Operator                                                                               | Do NOT use                                 |
| ----------------------------------------------- | ----------------------------------------------------------------------------------------------- | ------------------------------------------ |
| Generate QA pairs from text                     | `Text2MultiHopQAGenerator`                                                                    | `PromptedGenerator` with QA prompt       |
| Convert file path / URL to text                 | KBC trio (`FileOrURLToMarkdownConverterFlash` → `KBCChunkGenerator` → `KBCTextCleaner`) | `PromptedGenerator` to summarize files   |
| Score / evaluate using multiple fields          | `FormatStrPromptedGenerator` + `GeneralFilter`                                              | `PromptedFilter` (single input_key only) |
| Filter by deterministic rule on existing fields | `GeneralFilter`                                                                               | `PromptedFilter`                         |
| Generate new content from a single field        | `PromptedGenerator`                                                                           | —                                         |
| Generate new content from multiple fields       | `FormatStrPromptedGenerator`                                                                  | Multiple `PromptedGenerator` steps       |

**Key principle**: `PromptedGenerator` is the fallback for generic single-field generation. If the target mentions "QA", "question-answer", "问答" — always reach for `Text2MultiHopQAGenerator` first.
<!-- @if mcp==yes -->
In MCP mode, browse top-level category `core_text`; do not hallucinate query categories like `core_text/generate`.
If both `core_text` and another top-level category seem plausible, call `recommend_operator_categories` with the task description plus dataset columns before spending more MCP context. Treat its result as a hard budget: inspect at most the top 1-2 suggested categories, then switch to `get_operator_detail_by_name` instead of further category scans.
<!-- @endif -->

## Field Dependency Rules (MANDATORY)

<!-- @if mcp==yes -->
When this skill is used with MCP and a registered dataset, first call `get_dataset_columns` for the dataset id and treat those returned column names as the ground-truth initial field set.
<!-- @endif -->

1. **Inspect sample first**: Identify all available fields in user's sample data
2. **Field existence check**: If step N needs field X, then X must exist in original sample OR be output by step M where M < N
3. **Generate missing fields**: Use `PromptedGenerator` or `FormatStrPromptedGenerator` to create missing semantic fields
4. **Never reference before creation**: Cannot consume a field before it exists
5. **Avoid overwriting**: Do not overwrite original user fields unless explicitly requested
<!-- @if mcp==yes -->
6. **If committing through MCP**: call `validate_pipeline_config` before create/update; field-flow errors should be fixed before commit
7. If `validate_pipeline_config` returns `missing_input_field`, treat its `suggested_fields` and `repair_hint` as the first repair path. Fix the binding and re-validate; do **not** broaden MCP browsing or switch categories just because a field name was wrong.
<!-- @endif -->

```
✗ WRONG: Filter by "quality_score" before generating it
✓ CORRECT: Generate "quality_score" first, then filter by it
```

## Prompted Operator Usage Policy (MANDATORY)

- Don't mechanically create one prompted operator per tiny requirement. If one operator can handle multiple related transformations, prefer that over splitting.
- Multiple prompted operators are allowed when the task genuinely requires distinct semantic transformations. If using multiple, justify each step's role, input field, and output field.

## KBC Usage Constraint (MANDATORY)

The KBC trio must always be used in this exact order:

1. `FileOrURLToMarkdownConverterFlash` — converts file path / URL → Markdown text (field: `text_path`)
2. `KBCChunkGenerator` — splits Markdown into chunks (field: `raw_chunk`)
3. `KBCTextCleaner` — LLM-cleans each chunk (field: `cleaned_chunk`)

Rules:

- All three steps are required; never skip one.
- Input to step 1 must be a file path or URL, never plain text content.
- Each step's `output_key` becomes the next step's `input_key`.
- Use the default field names (`text_path`, `raw_chunk`, `cleaned_chunk`) unless explicitly requested otherwise.

## GeneralFilter Field Safety Rule (MANDATORY)

`GeneralFilter` lambda rules must ONLY reference fields that exist in sample data or are produced by upstream steps.

## Multi-Field Filtering Pattern (MANDATORY)

`PromptedFilter` only accepts a single `input_key`. For multi-field evaluation (e.g., scoring QA pairs), use `FormatStrPromptedGenerator` to score + `GeneralFilter` to filter.

**Important caveat for `Text2MultiHopQAGenerator` output**: The `QA_pairs` column is a nested list of dicts, not separate `question`/`answer` columns. You **cannot** directly pass `question` or `answer` as kwargs to `FormatStrPromptedGenerator` after `Text2MultiHopQAGenerator`. To score or filter individual QA pairs, use **post-processing** (explode the list into rows, then optionally score/filter in a second pipeline or in Python code).

## Output Contract (MANDATORY)

**Two-stage output required**:

### Stage 1: Intermediate Operator Decision (JSON)

Output this first:

```json
{
  "ops": ["OperatorA", "OperatorB", "OperatorC"],
  "field_flow": "field_a -> field_b -> field_c",
  "reason": "Why this ordered operator chain satisfies the target, how field dependencies are satisfied, and why prompted operators are or are not used."
}
```

### Stage 2: Complete Response (5 sections)

1. **Field Mapping**: Map sample fields to semantic roles, identify fields to generate
2. **Ordered Operator List**: List operators in execution order with justification
3. **Reasoning Summary**: Explain operator selection, field flow, why this design
4. **Complete Standard Pipeline Code**: Full executable Python following repository style
5. **Adjustable Parameters / Caveats**: Tunable parameters, fallback strategies, debugging tips

## LLM Serving Pre-check Rule (MANDATORY)

Before generating pipeline code, the agent MUST confirm the user's LLM serving configuration. If any LLM-dependent operator is used (e.g., `PromptedGenerator`, `FormatStrPromptedGenerator`, `PromptedFilter`, `Text2MultiHopQAGenerator`, `KBCTextCleaner`), the following information is required:

**Required information** (ask the user if not provided):

1. **`api_url`**: The LLM API endpoint (e.g., `https://api.openai.com/v1/chat/completions` or a self-hosted/proxy URL)
2. **`model_name`**: The model to use (e.g., `gpt-4o`, `gpt-4o-mini`, `deepseek-chat`)
3. **API key variable name**: Which environment variable holds the key — `OPENAI_API_KEY`, `DF_API_KEY`, etc. Ask for the **name only**. Never ask for, accept or repeat the key's value.

**When to ask**:

- If the user has NOT specified `api_url` or `model_name` in their request, ask before generating code
- If the pipeline uses LLM operators but the user only provided data + target, ask in one consolidated prompt:
  ```
  Pipeline 中使用了 LLM 算子，请确认以下配置：
  1. API 端点 (api_url)：例如 https://api.openai.com/v1/chat/completions
  2. 模型名称 (model_name)：例如 gpt-4o
  3. API Key 环境变量名（只要变量名，**不要粘贴密钥本身**）：例如 OPENAI_API_KEY（默认为 DF_API_KEY）
  ```
- Ask for the variable's **name**, never its value.
- To check it is set without revealing it, test for presence only — never print it:

  ```bash
  [ -n "${OPENAI_API_KEY:-}" ] && echo "OPENAI_API_KEY is set" || echo "OPENAI_API_KEY is NOT set"
  ```

  Do **not** run `echo $OPENAI_API_KEY`: that writes the secret into the
  transcript and into any log capturing it.

**When NOT to ask** (skip the pre-check):

- User explicitly provided all three pieces of info
- Pipeline uses only non-LLM operators (e.g., `GeneralFilter`, `KBCChunkGenerator`, `FileOrURLToMarkdownConverterFlash`)

<!-- @if profile==webui -->
**WebUI deployment context**:

When the pipeline is intended for WebUI execution (not local `python pipeline.py`), the serving must also be registered in the WebUI Serving Manager. After generating code, remind the user:
- Use the WebUI Serving Manager to create a serving with the correct `api_url`, `model_name`, and `api_key`
- In the WebUI pipeline editor, assign the serving to **ALL** LLM-dependent operators (not just the first one)
- Common failure: only the first operator gets a serving assigned; the rest remain empty, causing `Failed to process parameter: llm_serving` errors at execution time
- In WebUI/MCP mode, older prompts may still say `list_servings`; the backend now provides both `list_serving` and a backward-compatible alias, but prefer `list_serving` in new prompts/skills
<!-- @endif -->

<!-- @if mcp==yes -->
### MCP `create_pipeline` config structure (MANDATORY — MCP mode)

When building a pipeline via the MCP `create_pipeline` / `update_pipeline` tools (NOT
local codegen), the operator params JSON has two buckets — `init` and `run` — and
**where each value goes is determined by the operator's real signature**, which you
MUST fetch first with `get_operator_detail_by_name`. Getting this wrong makes the
pipeline crash at execution time, not at create time.

Hard rules:

1. **LLM serving goes in `init.llm_serving`, as the serving *id* (not the name).**
   Call `list_serving`, find the serving the user named, and use its `id`
   (e.g. `"0510fc816d385c6f"`). NEVER put serving in a `run` param such as
   `serving_name` — the operator's `run()` does not accept it and
   `run(**run_params)` will raise "unexpected keyword argument".

2. **`system_prompt` / `user_prompt` / `json_schema` / `prompt_template` are `init`
   params, not `run` params** for generator operators (e.g. `PromptedGenerator`,
   `ReasoningAnswerGenerator`). Only put a value in `run` if it appears in the
   operator's `run()` signature returned by `get_operator_detail_by_name`.

3. **Never send `None` (or empty string) for an optional text param that has a
   non-empty default.** If you don't have a real `user_prompt`, OMIT the param
   entirely so the operator uses its own default (e.g. `user_prompt=""`). Sending
   `None` can cause `None + str` TypeErrors inside the operator.

4. **`prompt_template` must be a plain allowed class name string** (e.g.
   `"MathAnswerGeneratorPrompt"`), taken from the operator detail's
   `allowed_prompts`. Do NOT send the `<class '...'>` repr — the validator rejects it.

5. **Always call `validate_pipeline_config` before `create_pipeline`** and fix any
   reported error (not just warnings) before creating.

Minimal correct example (PromptedGenerator, one LLM op):

```
operators: [{
  name: "PromptedGenerator",
  params: {
    init: [
      { name: "llm_serving",   value: "<serving_id_from_list_serving>" },
      { name: "system_prompt", value: "You are a helpful assistant." }
      // user_prompt omitted -> operator default "" is used
    ],
    run: [
      { name: "input_key",  value: "instruction" },
      { name: "output_key", value: "generated_answer" }
    ]
  }
}]
```
<!-- @endif -->

## Standard Code Generation Rule (MANDATORY)

**All generated Python code must follow the standard pipeline organization shown in the `examples/` folder of this skill package.**

**Input Data Format**:

- `first_entry_file_name` MUST be set to the **user-provided file path** (the JSONL sample file)
- File extension must be `.jsonl` (one JSON object per line, NOT an array)
- **DO NOT create new file paths** - use the exact path the user provided

**Required structure**: `__init__` (storage + llm_serving + operators) → `forward` (sequential `operator.run(storage=self.storage.step(), ...)`) → `if __name__ == "__main__"` entry point.

**DO NOT**: generate custom runtime executors, `forward(plan)` style frameworks, or dynamic dispatch engines.

## Operator Parameter Signature Rule (MANDATORY)

Use repository-valid constructor/run signatures only. Never invent parameter names.

### Base Components

**`FileStorage`**

```python
FileStorage(
  first_entry_file_name="...jsonl",
  cache_path="./cache",
  file_name_prefix="dataflow_cache_step",
  cache_type="jsonl"
)
```

**`APILLMServing_request`**

```python
APILLMServing_request(
  api_url="...",                       # user's LLM API endpoint
  key_name_of_api_key="OPENAI_API_KEY",  # env var name holding the API key (reads os.environ at runtime)
  model_name="gpt-4o",
  max_workers=10
)
```

**Note**: `key_name_of_api_key` is the **name of the environment variable** (not the key itself).
The class default is `"DF_API_KEY"`, but most deployments use `"OPENAI_API_KEY"`.
Always match the env var the user has set. The `api_url` should be the user's actual
API endpoint, not a placeholder.

**API Key handling (MANDATORY):**

The key is **read from the environment at run time and never written anywhere**.
`key_name_of_api_key` takes the *name* of an environment variable, not a key.

- **Never** put a key in generated code, not even as a placeholder to be replaced —
  that includes `os.environ["DF_API_KEY"] = "sk-..."`, a literal `api_key=`
  argument, and committed `.env` files.
- **Never** ask the user to paste a key into the chat, and never echo one back.
- If the variable the user named is not set, tell them to export it themselves and
  stop. Do not work around it:

  ```bash
  export OPENAI_API_KEY=...        # the user runs this, in their own shell
  ```

- In generated code, reference the variable by name only:

  ```python
  APILLMServing_request(
      api_url="https://api.openai.com/v1/chat/completions",
      key_name_of_api_key="OPENAI_API_KEY",   # name of the env var, never the key
      model_name="gpt-4o",
  )
  ```
<!-- @if profile==webui -->
- **WebUI deployment**: The API key is managed by the WebUI Serving Manager — users input it
  in the `api_key` field when creating/editing a serving. The WebUI backend injects it into the
  environment at execution time. Do NOT include `api_key` or `key_name_of_api_key` in
  `operators.json` — the engine handles this via the serving config.
<!-- @endif -->

### Six Core Operators: Signatures + Key Requirements

**1) `PromptedGenerator`**

- Constructor: `PromptedGenerator(llm_serving, system_prompt="You are a helpful agent.", user_prompt="", json_schema=None)`
- **`json_schema` rule**: If using `json_schema`, every `"type": "object"` in the schema MUST include `"additionalProperties": False`. Omitting it causes API 500 errors and infinite retries.
- Run: `run(storage=self.storage.step(), input_key="raw_content", output_key="generated_content")`
- `input_key` column must exist. Generated rows written to `output_key`.

**2) `FormatStrPromptedGenerator`**

- Constructor: `FormatStrPromptedGenerator(llm_serving, system_prompt="You are a helpful agent.", prompt_template=FormatStrPrompt(...), json_schema=None)`
- **`json_schema` rule**: If using `json_schema`, every `"type": "object"` in the schema MUST include `"additionalProperties": False`. Omitting it causes API 500 errors and infinite retries.
- Run: `run(storage=self.storage.step(), output_key="generated_content", **input_keys)`
- `**input_keys`: each kwarg maps a **template variable name** (key) to a **dataframe column name** (value). Internally does `row[input_keys[key]]` per row, then `prompt_template.build_prompt(need_fields, **key_dict)`.
- Kwarg keys must match `{placeholder}` names in `FormatStrPrompt.f_str_template`. Kwarg values must be existing dataframe columns.
- `prompt_template` cannot be `None` (raises `ValueError`). Must pass an instantiated `FormatStrPrompt(f_str_template="...")`.
- Import: `from dataflow.prompts.core_text import FormatStrPrompt`

**3) `Text2MultiHopQAGenerator`**

- Constructor: `Text2MultiHopQAGenerator(llm_serving=self.llm_serving, seed=0, lang="en", prompt_template=None, num_q=5)`
  - `llm_serving` — LLM serving instance (required)
  - `seed` (int, default `0`) — random seed for reproducibility
  - `lang` (str, default `"en"`) — language for generation prompt; controls sentence splitting (`"."` for `"en"`, `"。"` for `"zh"`)
  - `prompt_template` — custom `DIYPromptABC` instance; pass `None` to use default `Text2MultiHopQAGeneratorPrompt`
  - `num_q` (int, default `5`) — **maximum** number of QA pairs to **keep** per input row (truncates the generated list; actual generation count depends on sentence triples in the text)
- Run: `run(storage, input_key="cleaned_chunk", output_key="QA_pairs", output_meta_key="QA_metadata")`
  - `input_key` must exist (cleaned text chunk column)
  - `output_key` — column containing a **nested list** of QA dicts per row. Each dict has keys: `question` (str), `reasoning_steps` (list of `{step: str}`), `answer` (str), `supporting_facts` (list of str), `type` (str)
  - `output_meta_key` — column containing metadata dict per row with keys: `source`, `timestamp`, `complexity`
  - Output column named by `output_key` / `output_meta_key` must NOT pre-exist.
- Each input row produces **one row** with a nested list in the `output_key` column. The list items are dicts — `question`, `answer`, etc. are **NOT** separate dataframe columns. Downstream operators like `FormatStrPromptedGenerator` cannot directly reference `question` or `answer` as column names. To use individual QA pairs downstream, you must **post-process** (explode the list into separate rows) outside the operator chain.
- **Input text constraints** (texts failing these checks produce empty `qa_pairs: []`):
  - Length: 100–200,000 characters
  - Must contain at least 2 sentences (2+ `.` or 2+ `。`)
  - Special character ratio must be ≤ 30%

**4) `PromptedFilter`**

- Constructor: `PromptedFilter(llm_serving, system_prompt="...", min_score=1, max_score=5)`
- Run: `run(storage=self.storage.step(), input_key="raw_content", output_key="eval")`
- `input_key` must exist. `output_key` is numeric score column; rows outside `[min_score, max_score]` are filtered out.
- **IMPORTANT**: Rows where `input_key` is empty, null, or falsy are **silently dropped before scoring** — they will not appear in the output at all. Ensure the upstream operator produces non-empty values for every row, or expect row count to decrease.
- `system_prompt` controls the evaluation rubric. Default: `"Please evaluate the quality of this data on a scale from 1 to 5."`. Set a custom prompt for better scoring accuracy (e.g., specify evaluation criteria).

**5) `GeneralFilter`**

- Constructor: `GeneralFilter([lambda df: df["score"] >= 4, ...])`
- Run: `run(storage=self.storage.step())`
- Each rule must return boolean `pd.Series`. Referenced fields must already exist.

**6) KBC Trio (always used in this order)**

**Step 1 — `FileOrURLToMarkdownConverterFlash`**

- Constructor: `FileOrURLToMarkdownConverterFlash(intermediate_dir="../example_data/KBCleaningPipeline/flash/", mineru_model_path="opendatalab/MinerU2.5-2509-1.2B", batch_size=4, replicas=1, num_gpus_per_replica=1.0, engine_gpu_util_rate_to_ray_cap=0.9)`
- **Does NOT take `llm_serving`** — this operator has no LLM dependency.
- `mineru_model_path` is **required** — passing `None` raises `ValueError`. Use a HuggingFace model ID or local path.
- Run: `run(storage=self.storage.step(), input_key="source", output_key="text_path")`
- Input must be a file path or URL (`.pdf`, `.png`, `.jpg`, `.jpeg`, `.webp`, `.gif`, `.html`, `.xml`, `.txt`, `.md`).

**Step 2 — `KBCChunkGenerator`**

- Constructor: `KBCChunkGenerator(chunk_size=512, chunk_overlap=50, split_method="token", min_tokens_per_chunk=100, tokenizer_name="bert-base-uncased")`
- Run: `run(storage=self.storage.step(), input_key="text_path", output_key="raw_chunk")`
- `split_method` options: `"token"`, `"sentence"`, `"semantic"`, `"recursive"`.

**Step 3 — `KBCTextCleaner`**

- Constructor: `KBCTextCleaner(llm_serving, lang="en")`
- Run: `run(storage=self.storage.step(), input_key="raw_chunk", output_key="cleaned_chunk")`
- LLM-cleans each chunk; output is ready for downstream QA generation.

### Correct Import Paths (MANDATORY)

```python
# Base components
from dataflow.utils.storage import FileStorage
from dataflow.serving import APILLMServing_request

# Operators
from dataflow.operators.core_text import PromptedGenerator, FormatStrPromptedGenerator, Text2MultiHopQAGenerator, PromptedFilter, GeneralFilter
from dataflow.operators.knowledge_cleaning import FileOrURLToMarkdownConverterFlash, KBCChunkGenerator, KBCTextCleaner
```

## Extended operator reference

Use the sibling `../core_text/` package only when the six core primitives above
do not cover the task, or when you need an operator's edge cases, exact return
semantics, or documented failure modes. Do not load it preemptively.

<!-- @if mcp==yes -->
1. Call `get_operator_detail_by_name`; it reflects the installed version and is
   authoritative.
<!-- @else -->
1. Read `../core_text/SKILL.md` to choose a category, then read the selected
   `../core_text/<category>/<operator>/SKILL.md` and its documented failure
   example. This bundled reference is the offline source for this profile.
<!-- @endif -->
2. Treat the six core primitives in this file as the fast path. The `core_text`
   index owns the full operator list, so do not duplicate it here.

## Input File Content Analysis Rule (MANDATORY)

Analyze sample data content to determine task nature:

**File path fields** (e.g., `pdf_path`, `image_path`, `doc_path`):

- → KBC trio in order: `FileOrURLToMarkdownConverterFlash` → `KBCChunkGenerator` → `KBCTextCleaner` (supports `.pdf`, `.png`, `.jpg`, `.jpeg`, `.webp`, `.gif`, `.html`, `.xml`, `.txt`, `.md`)
- → Document/file processing workflow

**Plain text fields** (e.g., `text`, `content`, `review_text`):

- → Use `PromptedGenerator`, `PromptedFilter`, `Text2MultiHopQAGenerator`, `FormatStrPromptedGenerator`, `GeneralFilter`
- → Do NOT use KBC

**Multiple semantic fields** (e.g., `instruction`, `output`, `question`, `answer`):

- → Use `FormatStrPromptedGenerator` for combining fields
- → Use `GeneralFilter` for field-based rules

## Examples

See `examples/` folder for complete workflows:

1. **`examples/basic_generate_and_filter.md`** — `PromptedGenerator` + `PromptedFilter` (simplest pattern)
2. **`examples/multifield_scoring.md`** — `FormatStrPromptedGenerator` with multi-field scoring
3. **`examples/multi_stage_pipeline.md`** — Multiple `PromptedGenerator` stages + `GeneralFilter`
4. **`examples/kbc_pdf_to_qa.md`** — KBC trio (`FileOrURLToMarkdownConverterFlash` + `KBCChunkGenerator` + `KBCTextCleaner`) + `Text2MultiHopQAGenerator` + `PromptedFilter` (scores nested QA_pairs column per chunk)

These are strategy guidance, not templates to copy blindly. Generated code must follow standard pipeline structure.