ChunkedPromptedGenerator Operator Reference

ChunkedPromptedGenerator reads file paths from a dataframe column, loads each file from disk, recursively splits long content into chunks, calls the LLM on all chunks, joins the generated outputs with a separator, writes the joined result into a new text file, and stores that output file path back into the dataframe.

1. Import

from dataflow.operators.core_text import ChunkedPromptedGenerator

2. Constructor

ChunkedPromptedGenerator(
    llm_serving=llm,
    system_prompt="You are a helpful agent.",
    json_schema=None,
    max_chunk_len=128000,
    enc=tiktoken.get_encoding("cl100k_base"),
    separator="\n",
)

3. run() Signature

op.run(
    storage=self.storage.step(),
    input_path_key="file_path",
    output_path_key="output_path",
)
# returns: output_path_key

4. Actual Execution Logic

The current implementation behaves as follows:

1. Read the dataframe from storage.
2. For each row, read the file content from Path(row[input_path_key]).read_text(encoding="utf-8").
3. Count tokens with len(enc.encode(text)).
4. If the text exceeds max_chunk_len, recursively split it into two halves by character position, not by sentence or token boundary.
5. For each chunk, build one LLM input as:

system_prompt + "\n" + chunk

6. Flatten all chunks from all rows into a single global batch.
7. Call:

llm_serving.generate_from_input(all_llm_inputs)

or, when json_schema is provided:

llm_serving.generate_from_input(all_llm_inputs, json_schema=json_schema)

8. Regroup the responses back by original row.
9. For each row, join that row's chunk results with the configured separator.
10. Write the joined text into a new file whose path is derived automatically as:

row[input_path_key].split(".")[0] + "_llm_output.txt"

11. Store that generated output file path into output_path_key.
12. Persist the dataframe through storage.write(dataframe) and return output_path_key.

5. Important Rules

1. Values in input_path_key must be readable filesystem paths, not inline text.
2. input_path_key and output_path_key are required runtime arguments; the source code does not define defaults for them.
3. Chunk splitting is recursive bisection by character index, guided by token counts, not semantic splitting.
4. Output files are auto-named from the input path using the _llm_output.txt suffix. The value already present in output_path_key is not used to decide the write location.
5. If global LLM generation fails, the implementation writes empty content per row into the derived output files and still writes output_path_key back into the dataframe.
6. All chunk requests across all rows are sent in one global batch before results are regrouped by row.

6. Typical Usage

import tiktoken

from dataflow.operators.core_text import ChunkedPromptedGenerator

generator = ChunkedPromptedGenerator(
    llm_serving=self.llm_serving,
    system_prompt="Please summarize the following text.",
    max_chunk_len=4096,
    enc=tiktoken.get_encoding("cl100k_base"),
    separator="\n\n",
)

generator.run(
    storage=self.storage.step(),
    input_path_key="file_path",
    output_path_key="output_path",
)