Build and run LLM-powered data processing pipelines with DocETL. Use when users say "docetl", want to analyze unstructured data, process documents, extract info
复制下面这句话，粘贴给 Claude Code、Codex、Cursor 等 AI 编程工具，它会读取安装说明并在你确认后完成安装。
请阅读 https://ai.atlankj.com/install/asset/gh-docetl-745512a5f62c ，按照其中的说明把「docetl」安装到你（当前 AI 工具）中。执行前先告诉我将运行的命令和写入的位置，等我确认。
查看 AI 将读取的安装说明正在读取 GitHub 原文…
内容来自 GitHub 原始文件，由原作者维护。在 GitHub 查看
DocETL is a system for creating LLM-powered data processing pipelines. This skill helps you build end-to-end pipelines: from data preparation to execution and optimization.
Work like a data analyst: write → run → inspect → iterate. Never write all scripts at once and run them all at once. Each phase should be completed and validated before moving to the next.
sample: 10-20 for testingsample parameter and run full pipelineVisualization Aesthetics:
Report Structure:
Interactive Tables:
Source Document Links:
Key principle: The user should see results at every step. Don't proceed to the next phase until the current phase produces good results.
DocETL datasets must be JSON arrays or CSV files.
[
{"id": 1, "text": "First document content...", "metadata": "value"},
{"id": 2, "text": "Second document content...", "metadata": "value"}
]
id,text,metadata
1,"First document content...","value"
2,"Second document content...","value"
If user needs to collect data, write a Python script:
import json
# Collect/transform data
documents = []
for source in sources:
documents.append({
"id": source.id,
"text": source.content, # DO NOT truncate text
# Add relevant fields
})
# Save as DocETL dataset
with open("dataset.json", "w") as f:
json.dump(documents, f, indent=2)
Important: Never truncate document text in collection scripts. DocETL operations like split handle long documents properly. Truncation loses information.
Always run the collection script and inspect results before proceeding. Show the user:
import json
data = json.load(open("dataset.json"))
print(f"Total documents: {len(data)}")
print(f"Keys: {list(data[0].keys())}")
print(f"Avg length: {sum(len(str(d)) for d in data) // len(data)} chars")
# Show sample
print("\nSample document:")
print(json.dumps(data[0], indent=2)[:500])
Only proceed to pipeline development once the data looks correct.
CRITICAL: Before writing any prompts, READ the actual input data to understand:
import json
with open("dataset.json") as f:
data = json.load(f)
# Examine several examples
for doc in data[:5]:
print(doc)
This understanding is essential for writing specific, effective prompts.
DocETL supports two equivalent approaches. Use whichever the user prefers:
Create a YAML file with this structure:
default_model: gpt-5-nano
system_prompt:
dataset_description: <describe the data based on what you observed>
persona: <role for the LLM to adopt>
datasets:
input_data:
type: file
path: "dataset.json" # or dataset.csv
operations:
- name: <operation_name>
type: <operation_type>
prompt: |
<Detailed, specific prompt based on the actual data>
output:
schema:
<field_name>: <type>
pipeline:
steps:
- name: process
input: input_data
operations:
- <operation_name>
output:
type: file
path: "output.json"
intermediate_dir: "intermediates" # ALWAYS set this for debugging
The same pipeline as chainable Python calls:
import docetl
docetl.default_model = "gpt-5-nano"
docetl.intermediate_dir = "intermediates"
results = (
docetl.read_json("dataset.json") # or read_csv(), from_list()
.map(
prompt="...",
output={"schema": {"field": "type"}},
)
.reduce(
reduce_key="category",
prompt="...",
output={"schema": {"summary": "string"}},
)
.collect() # returns list[dict]; .to_pandas() for a DataFrame
)
# Cost and token tracking
print(f"Cost: ${pipeline.total_cost:.4f}")
Key Frame API methods:
docetl.read_json(), docetl.read_csv(), docetl.read_parquet(), docetl.from_list().map(), .filter(), .reduce(), .resolve(), .equijoin(), .split(), .gather(), .unnest(), .code_map(), .code_filter(), .code_reduce().show() (run on sample, print results), .collect() (list of dicts), .to_pandas() (DataFrame), .write_json(), .write_csv(), .write_parquet().schema() (output schema), .count() (input doc count on bare datasets), .to_yaml() (export pipeline as YAML), .to_python()docetl.default_model, docetl.max_threads, docetl.bypass_cache, docetl.rate_limits, docetl.intermediate_dir, docetl.agent_model, docetl.fallback_modelsAll operation parameters are the same between YAML and Python — just pass them as keyword arguments (e.g., validate=["len(output['items']) >= 1"], fold_prompt="...", fold_batch_size=100).
Use agent=docetl.Agent(...) on .map(), .filter(), or .reduce() when an operation needs tools before returning structured output:
@docetl.tool
def lookup_sla(customer_tier: str) -> dict[str, str | int]:
"""Return support entitlements for a customer tier."""
return {
"enterprise": {"response_hours": 1, "escalation": "page-on-call"},
"growth": {"response_hours": 4, "escalation": "queue-lead"},
"free": {"response_hours": 48, "escalation": "self-serve"},
}.get(customer_tier.lower(), {"response_hours": 24, "escalation": "standard"})
agent = docetl.Agent(tools=[lookup_sla], max_turns=5, max_tool_calls=3)
rows = (
docetl.from_list([{"ticket": "Production API latency is above SLO", "customer_tier": "enterprise"}])
.map(
prompt="Use lookup_sla to classify this ticket: {{ input.ticket }} / {{ input.customer_tier }}",
output={"schema": {"priority": "str", "next_action": "str"}},
model="azure/gpt-4o-mini",
agent=agent,
)
.collect()
)
Key points:
agent configs in YAML and do not export them with .to_yaml() / .to_python().model= still selects the model. Python tools wrapped with @docetl.tool are the most provider-portable path through LiteLLM-compatible models.WebSearchTool, hosted ShellTool, docetl.tools.Sandbox.create(...)) require an OpenAI hosted-tool path. docetl.tools.Sandbox.create(...) creates a persistent OpenAI hosted container; sandbox.bash() returns a shell tool bound to that container.specialist.as_tool(name=..., description=...).docs/api-reference/python.md#tool-equipped-mapfilterreducedocs/operators/map.md#tool-equipped-map-agentsdocs/operators/filter.md#tool-equipped-filter-agentsdocs/operators/reduce.md#tool-equipped-reduce-agentsdocs/examples/tool-equipped-research-agents.mdgpt-5-nano or gpt-5-mini for extraction/map operations| Operation Type | Recommended Model | Rationale |
|---|---|---|
| Map (extraction) | gpt-5-nano or gpt-5-mini | High volume, simple per-doc tasks |
| Filter | gpt-5-nano | Simple yes/no decisions |
| Reduce (summarization) | gpt-4.1 or gpt-5.1 | Complex synthesis across many docs |
| Resolve (deduplication) | gpt-5-nano or gpt-5-mini | Simple pairwise comparisons |
Use cheaper models for high-volume extraction, and more capable models for synthesis/summarization where quality matters most.
Prompts must be specific to the data, not generic. After reading the input data:
prompt: |
Extract key information from this document.
{{ input.text }}
prompt: |
You are analyzing a medical transcript from a doctor-patient visit.
The transcript follows this format:
- Doctor statements are prefixed with "DR:"
- Patient statements are prefixed with "PT:"
- Timestamps appear in brackets like [00:05:23]
From the following transcript, extract:
1. All medications mentioned (brand names or generic)
2. Dosages if specified
3. Patient-reported side effects or concerns
Transcript:
{{ input.transcript }}
Be thorough - patients often mention medication names informally.
If a medication is unclear, include it with a note.
Many tasks only need a single map operation. Use good judgement:
| Task | Recommended Approach |
|---|---|
| Extract info from each doc | Single map |
| Multiple extractions | Multiple map operations chained |
| Extract then summarize | map → reduce |
| Filter then process | filter → map |
| Split long docs | split → map → reduce |
| Deduplicate entities | map → unnest → resolve |
Applies an LLM transformation to each document independently.
- name: extract_info
type: map
prompt: |
Analyze this document:
{{ input.text }}
Extract the main topic and 3 key points.
output:
schema:
topic: string
key_points: list[string]
model: gpt-5-nano # optional, uses default_model if not set
skip_on_error: true # recommended for large-scale runs
validate: # optional
- len(output["key_points"]) == 3
num_retries_on_validate_failure: 2 # optional
Key parameters:
prompt: Jinja2 template, use {{ input.field }} to reference fieldsoutput.schema: Define output structureskip_on_error: Set true to continue on LLM errors (recommended at scale)validate: Python expressions to validate outputsample: Process only N documents (for testing)limit: Stop after producing N outputsKeeps or removes documents based on LLM criteria. Output schema must have exactly one boolean field.
- name: filter_relevant
type: filter
skip_on_error: true
prompt: |
Document: {{ input.text }}
Is this document relevant to climate change?
Respond true or false.
output:
schema:
is_relevant: boolean
Aggregates documents by a key using an LLM.
Always include fold_prompt and fold_batch_size for reduce operations. This handles cases where the group is too large to fit in context.
- name: summarize_by_category
type: reduce
reduce_key: category # use "_all" to aggregate everything
skip_on_error: true
prompt: |
Summarize these {{ inputs | length }} items for category "{{ inputs[0].category }}":
{% for item in inputs %}
- {{ item.title }}: {{ item.description }}
{% endfor %}
Provide a 2-3 sentence summary of the key themes.
fold_prompt: |
You have a summary based on previous items, and new items to incorporate.
Previous summary (based on {{ output.item_count }} items):
{{ output.summary }}
New items ({{ inputs | length }} more):
{% for item in inputs %}
- {{ item.title }}: {{ item.description }}
{% endfor %}
Write a NEW summary that covers ALL items (previous + new).
IMPORTANT: Output a clean, standalone summary as if describing the entire dataset.
Do NOT mention "updated", "added", "new items", or reference the incremental process.
fold_batch_size: 100
output:
schema:
summary: string
item_count: int
validate:
- len(output["summary"].strip()) > 0
num_retries_on_validate_failure: 2
Critical: Writing Good Fold Prompts
The fold_prompt is called repeatedly as batches are processed. Its output must:
prompt outputBad fold_prompt output: "Added 50 new projects. The updated summary now includes..." Good fold_prompt output: "Developers are building privacy-focused tools and local-first apps..."
Estimating fold_batch_size:
Key parameters:
reduce_key: Field to group by (or list of fields, or _all)fold_prompt: Template for incrementally adding items to existing output (required)fold_batch_size: Number of items per fold iteration (required, use 100+)associative: Set to false if order mattersDivides long text into smaller chunks. No LLM call.
- name: split_document
type: split
split_key: content
method: token_count # or "delimiter"
method_kwargs:
num_tokens: 500
model: gpt-5-nano
Output adds:
{split_key}_chunk: The chunk content{op_name}_id: Original document ID{op_name}_chunk_num: Chunk numberFlattens list fields into separate rows. No LLM call.
- name: unnest_items
type: unnest
unnest_key: items # field containing the list
keep_empty: false # optional
Example: If a document has items: ["a", "b", "c"], unnest creates 3 documents, each with items: "a", items: "b", items: "c".
Deduplicates and canonicalizes entities. Uses pairwise comparison.
- name: dedupe_names
type: resolve
optimize: true # let optimizer find blocking rules
skip_on_error: true
comparison_prompt: |
Are these the same person?
Person 1: {{ input1.name }} ({{ input1.email }})
Person 2: {{ input2.name }} ({{ input2.email }})
Respond true or false.
resolution_prompt: |
Standardize this person's name:
{% for entry in inputs %}
- {{ entry.name }}
{% endfor %}
Return the canonical name.
output:
schema:
name: string
Important: Set optimize: true and run docetl build to generate efficient blocking rules. Without blocking, this is O(n²).
Deterministic Python transformations without LLM calls.
code_map:
- name: compute_stats
type: code_map
code: |
def transform(doc) -> dict:
return {
"word_count": len(doc["text"].split()),
"char_count": len(doc["text"])
}
code_reduce:
- name: aggregate
type: code_reduce
reduce_key: category
code: |
def transform(items) -> dict:
total = sum(item["value"] for item in items)
return {"total": total, "count": len(items)}
code_filter:
- name: filter_long
type: code_filter
code: |
def transform(doc) -> bool:
return len(doc["text"]) > 100
Augment LLM operations with retrieved context from a LanceDB index. Useful for:
Define a retriever:
retrievers:
facts_index:
type: lancedb
dataset: extracted_facts # dataset to index
index_dir: workloads/wiki/lance_index
build_index: if_missing # if_missing | always | never
index_types: ["fts", "embedding"] # or "hybrid"
fts:
index_phrase: "{{ input.fact }}: {{ input.source }}"
query_phrase: "{{ input.fact }}"
embedding:
model: openai/text-embedding-3-small
index_phrase: "{{ input.fact }}"
query_phrase: "{{ input.fact }}"
query:
mode: hybrid
top_k: 5
Use in operations:
- name: find_conflicts
type: map
retriever: facts_index
prompt: |
Check if this fact conflicts with any retrieved facts:
Current fact: {{ input.fact }} (from {{ input.source }})
Related facts from other articles:
{{ retrieval_context }}
Return whether there's a genuine conflict.
output:
schema:
has_conflict: boolean
Python API — create a docetl.Retriever object and pass it to operations:
retriever = docetl.Retriever(
dataset="facts", # the frame's input dataset (basename of facts.json)
index_dir="workloads/wiki/lance_index",
index_types=["fts", "embedding"],
fts={"index_phrase": "{{ input.fact }}", "query_phrase": "{{ input.fact }}"},
embedding={
"model": "openai/text-embedding-3-small",
"index_phrase": "{{ input.fact }}",
"query_phrase": "{{ input.fact }}",
},
query={"mode": "hybrid", "top_k": 5},
)
results = (
docetl.read_json("facts.json")
.map(
prompt="Check conflicts: {{ input.fact }}\n{{ retrieval_context }}",
output={"schema": {"has_conflict": "boolean"}},
retriever=retriever,
)
.collect()
)
The retriever parameter is available on .map(), .filter(), .reduce(), and .extract().
In the Python API, pass the data to index directly with data= (a file path or list of dicts — use this for external knowledge bases), or reference an existing pipeline dataset with dataset=: the frame's own input (file basename, or from_list's name=, default "data") or a previous step's output (step_<operation_name>).
Key points:
{{ retrieval_context }} is injected into prompts automaticallybuild_index: if_missing)fts), vector (embedding), or hybrid searchsave_retriever_output: true to debug what was retrievedFor detailed parameters, advanced features, and more examples, read the docs:
docs/operators/ folder (map.md, reduce.md, filter.md, etc.)docs/concepts/ folder (pipelines.md, operators.md, schemas.md)docs/examples/ folderdocs/optimization/ folderBefore running, verify API keys exist:
# Check for .env file
cat .env
Required keys depend on the model:
OPENAI_API_KEYANTHROPIC_API_KEYGEMINI_API_KEYIf missing, prompt user to create .env:
OPENAI_API_KEY=sk-...
Always test on a sample first, then run full pipeline.
Add sample: 10-20 to your first operation, then run:
YAML:
docetl run pipeline.yaml
Python:
# Add sample=10 to the first operation for testing
results = (
docetl.read_json("dataset.json")
.map(prompt="...", output={"schema": {"field": "type"}}, sample=10)
.collect()
)
Inspect the test results before proceeding:
import json
from collections import Counter
# Load intermediate results
data = json.load(open("intermediates/step_name/operation_name.json"))
print(f"Processed: {len(data)} docs")
# Check distributions
if "domain" in data[0]:
print("Domain distribution:")
for k, v in Counter(d["domain"] for d in data).most_common():
print(f" {k}: {v}")
# Show sample outputs
print("\nSample output:")
print(json.dumps(data[0], indent=2))
Once test results look good:
sample parameter from the pipelineOptions:
--max_threads N - Control parallelismCheck intermediate results in the intermediate_dir folder to debug each step.
Use MOAR optimizer to find the Pareto frontier of cost vs. accuracy tradeoffs. MOAR experiments with different pipeline rewrites and models to find optimal configurations.
YAML approach — add optimizer_config to the pipeline:
optimizer_config:
type: moar
save_dir: ./optimization_results
available_models:
- gpt-5-nano
- gpt-4o-mini
- gpt-4o
evaluation_file: evaluate.py # User must provide
metric_key: score
max_iterations: 20
model: gpt-5-nano
docetl build pipeline.yaml --optimizer moar
Python approach — call .optimize() on the Frame:
@docetl.register_eval
def evaluate(results):
correct = sum(1 for o in results if is_correct(o))
return {"score": correct / len(results)}
optimized = frame.optimize(
eval_fn=evaluate,
metric_key="score",
models=["gpt-5-nano", "gpt-4o-mini", "gpt-4o"],
max_iterations=20,
save_dir="./optimization_results",
)
rows = optimized.collect()
print(optimized.search_results.to_df()) # Pareto frontier
MOAR will produce multiple pipeline variants on the Pareto frontier - user can choose based on their cost/accuracy preferences.
Keep schemas minimal and simple unless the user explicitly requests more fields. Default to 1-3 output fields per operation. Only add more fields if the user specifically asks for them.
Nesting limit: Maximum 2 levels deep (e.g., list[{field: str}] is allowed, but no deeper).
# Good - minimal, focused on the core task
output:
schema:
summary: string
# Good - a few fields when task requires it
output:
schema:
topic: string
keywords: list[string]
# Acceptable - 2 levels of nesting (list of objects)
output:
schema:
items: "list[{name: str, value: int}]"
# Bad - too many fields (unless user explicitly requested all of these)
output:
schema:
conflicts_found: bool
num_conflicts: int
conflicts: "list[{claim_a: str, source_a: str, claim_b: str, source_b: str}]"
analysis_summary: str
# Bad - more than 2 levels of nesting (not supported)
output:
schema:
data: "list[{nested: {too: {deep: str}}}]"
Guidelines:
Supported types: string, int, float, bool, list[type], enum
Always add validation to LLM-powered operations (map, reduce, filter, resolve). Validation catches malformed outputs and retries automatically.
- name: extract_keywords
type: map
prompt: |
Extract 3-5 keywords from: {{ input.text }}
output:
schema:
keywords: list[string]
validate:
- len(output["keywords"]) >= 3
- len(output["keywords"]) <= 5
num_retries_on_validate_failure: 2
Common validation patterns:
# List length constraints
- len(output["items"]) >= 1
- len(output["items"]) <= 10
# Enum/allowed values
- output["sentiment"] in ["positive", "negative", "neutral"]
# String not empty
- len(output["summary"].strip()) > 0
# Numeric ranges
- output["score"] >= 0
- output["score"] <= 100
For map operations, use input:
prompt: |
Document: {{ input.text }}
{% if input.metadata %}
Context: {{ input.metadata }}
{% endif %}
For reduce operations, use inputs (list):
prompt: |
Summarize these {{ inputs | length }} items:
{% for item in inputs %}
- {{ item.summary }}
{% endfor %}
.env has correct API keysvalidate rules with retriesgpt-5-nano or gpt-4o-minisample: 10 to test on subset firstLook in intermediate_dir folder to debug each step.
# Run pipeline
docetl run pipeline.yaml
# Run with more parallelism
docetl run pipeline.yaml --max_threads 16
# Optimize pipeline (cost/accuracy tradeoff)
docetl build pipeline.yaml --optimizer moar
# Clear LLM cache
docetl clear-cache
# Check version
docetl version