DocETL：基于声明式配置与LLM的智能文档处理管道实战指南

1. 项目概述：DocETL，一个为复杂文档处理而生的智能管道引擎

如果你和我一样，经常需要处理一堆五花八门的文档——PDF报告、Word合同、Excel表格、网页截图，甚至手写笔记的扫描件，然后从中提取、清洗、分析信息，那你一定懂这种痛。传统的脚本写起来繁琐，通用ETL工具对非结构化文档又不够“聪明”，而直接调用大语言模型（LLM）API，又面临着提示词调试困难、流程难以固化、成本不可控等一系列问题。今天要深入聊的DocETL，正是为了解决这些痛点而生的一个开源项目。它不是一个简单的封装库，而是一个完整的、声明式的文档处理管道（Pipeline）框架，核心思想是“用配置定义流程，用智能处理文档”。

简单来说，DocETL 让你能够像搭积木一样，通过一个清晰的配置文件（YAML/JSON），定义从文档加载、分块、清洗、到调用LLM进行信息提取、分类、总结，再到最终数据输出的完整流程。它提供了两套核心工具：一个是用于快速原型设计和交互式调试的Web UI（DocWrangler），另一个是用于生产环境稳定运行的Python包。这意味着，你可以先在可视化界面上像玩拼图一样把流程跑通、把提示词调优，然后一键将配置导出，在服务器上以命令行或代码方式批量、自动化地执行。这对于需要处理大量格式不一、内容复杂的文档场景，比如法律合同审查、学术论文分析、商业报告生成、客户支持工单分类等，无疑是一个生产力利器。

2. 核心架构与设计哲学拆解

2.1 声明式管道：为何比传统脚本更胜一筹？

在接触DocETL之前，我的文档处理流程通常是这样的：写一个Python脚本，用PyPDF2或pdfplumber读PDF，用python-docx读Word，用BeautifulSoup爬网页，然后写一堆正则表达式和字符串处理函数来清洗文本，最后再拼接一段OpenAI API的调用代码。每次需求变动，比如要新增一个数据字段，或者处理一种新格式的文档，都意味着要重新阅读和修改这一大坨过程式代码，调试起来非常头疼。

DocETL采用了声明式（Declarative）的设计哲学。你不再需要关心“如何做”（How），而是专注于“做什么”（What）。你通过一个结构化的配置文件，描述数据需要经过哪些处理步骤（称为“操作符”或Operator），每个步骤需要什么参数。DocETL的运行时引擎会负责解析这个配置，并按正确的顺序和依赖关系执行所有操作。

举个例子，一个简单的提取合同关键信息的管道，其配置骨架可能长这样：

pipeline: - operator: load_directory # 操作符1：从目录加载文件 params: path: "./contracts/" glob_pattern: "*.pdf" - operator: extract_text # 操作符2：从PDF中提取文本 params: engine: "pdfplumber" - operator: split_by_heading # 操作符3：按标题分割文档 params: heading_pattern: "^第[一二三四五六七八九十]+条" - operator: llm_extract # 操作符4：调用LLM提取结构化信息 params: model: "gpt-4o-mini" prompt: | 你是一名法律专家。请从以下合同条款中提取如下信息： - 甲方名称 - 乙方名称 - 合同金额（数字） - 付款方式 - 生效日期 合同条款文本：{{chunk_text}} output_schema: type: object properties: party_a: {type: "string"} party_b: {type: "string"} amount: {type: "number"} payment_method: {type: "string"} effective_date: {type: "string", format: "date"} - operator: write_jsonl # 操作符5：将结果写入JSON Lines文件 params: output_path: "./extracted_contracts.jsonl"

这种声明式的好处显而易见：

1. 可读性与可维护性极强：配置文件本身就是最好的文档，任何接手项目的人都能在几分钟内理解整个数据处理流程。
2. 易于迭代和复用：想尝试不同的文本分割方法？只需修改split_by_heading操作符的参数，或者换成另一个分割操作符。这个管道配置可以保存为模板，复用于类似的项目。
3. 关注点分离：开发者的精力可以完全集中在业务逻辑（要提取什么信息）和提示词工程上，而不是底层的文件I/O、错误处理和API调用细节。

2.2 双引擎驱动：交互式开发与生产部署的无缝衔接

DocETL另一个精妙的设计是它的“双引擎”模式，完美覆盖了从开发到上线的全生命周期。

引擎一：DocWrangler（交互式UI游乐场）这是DocETL的“开发环境”。它是一个本地运行的Web应用，提供了一个可视化的界面，让你可以：

• 拖拽式构建管道：虽然当前版本更侧重于配置编辑，但其理念是可视化编排处理步骤。
• 实时调试与预览：上传一个样本文档，运行你的管道配置，每一步的中间结果都会立刻显示出来。你可以看到文本是如何被分割的，LLM返回的原始响应是什么。这对于调试复杂的提示词至关重要。
• 交互式提示词改进：UI内置了“改进提示词”的功能，你可以基于当前输出，让AI助手帮你优化提示词，使其更清晰、指令更明确。
• 配置导出：调试满意后，一键将当前界面上配置好的管道导出为YAML或JSON文件，这个文件就是可以直接用于生产环境的“蓝图”。

实操心得：在开发复杂管道时，我强烈建议始终从DocWrangler开始。即使你是个命令行高手，可视化中间结果的能力也能帮你节省大量猜测和打印日志的时间。我曾经花了一个小时调试一个提取精度不高的问题，最后在DocWrangler里发现，是因为前一个文本清洗操作符意外地删除了关键的分隔符，导致后续分割错位。这个问题在纯代码调试中很难一眼看出。

引擎二：DocETL Python包（生产运行时）这是DocETL的“生产环境”。你通过pip install docetl安装的包，核心就是一个强大的管道执行引擎。它：

• 无头（Headless）运行：可以在服务器、容器或任何没有图形界面的环境中运行。
• 支持批量处理：可以轻松地遍历整个目录的文件，进行大规模批处理。
• 易于集成：可以作为Python库在你的业务代码中调用，也可以通过命令行工具直接执行管道配置文件。
• 资源与成本控制：在生产环境中，你可以更精细地控制并发度、重试策略、API调用频率和费用监控。

两者关系：DocWrangler 和 DocETL Python包共享同一套操作符定义和配置规范。在DocWrangler中调试成功的管道配置，几乎可以无缝地在生产环境中运行。这消除了从开发到部署的“最后一公里”鸿沟。

3. 核心操作符详解与实战配置

DocETL的强大，建立在它丰富且可扩展的操作符体系之上。理解这些核心操作符，是构建高效管道的关键。我们可以将其分为四大类：输入输出、文档处理、AI智能处理、流程控制。

3.1 输入与输出操作符：数据的入口与归宿

加载操作符 (Load Operators)这是管道的起点，负责将原始数据载入系统，并转换为DocETL内部统一的文档对象格式。

• load_file: 加载单个文件。支持本地路径。
• load_directory: 加载整个目录的文件，支持通配符过滤。这是处理批量文档最常用的操作符。
• load_webpage: 抓取并加载网页内容。
• load_from_string: 直接传入文本字符串，用于测试或集成。

关键参数解析：

• glob_pattern: 在load_directory中，使用如*.pdf，**/*.docx来递归匹配文件。
• encoding: 指定文本文件的编码，对于中文文档，常需设置为utf-8或gbk。DocETL默认支持多种编码，可在环境变量TEXT_FILE_ENCODINGS中配置。

写入操作符 (Write Operators)这是管道的终点，负责将处理后的结构化数据持久化。

• write_jsonl: 将结果写入JSON Lines格式，每行一个JSON对象。这是与下游数据系统（如数据库、数据分析工具）交互的理想格式。
• write_csv: 将结果写入CSV文件，适合表格型数据。
• write_parquet: 写入高效的列式存储格式Parquet，适用于大数据量场景。
• write_to_stdout: 直接打印到控制台，用于调试。

注意事项：write_jsonl操作符会处理DocETL内部复杂的文档结构，将其扁平化为指定的输出模式。你需要确保上游LLM提取操作符的output_schema与写入的格式匹配。如果上游输出是一个列表，写入时需要特别注意。

3.2 文档预处理操作符：为AI理解做好准备

原始文档通常不能直接喂给LLM。预处理的目标是将文档转换成适合LLM处理、且能保留最大信息量的“块”（Chunks）。

文本提取与清洗

• extract_text: 从二进制格式（PDF, DOCX, PPTX）中提取纯文本。底层可集成pdfplumber,pypdf,python-docx等库。
  • 参数选择：对于复杂的、带表格的PDF，pdfplumber通常比pypdf提取效果更好，但速度稍慢。需要在质量和性能间权衡。
• clean_text: 执行基础的文本清洗，如去除多余空白符、替换特殊字符、标准化换行符等。

文档分割 (Chunking)这是预处理中最关键、最影响后续AI效果的一步。分割的目标是得到语义相对完整、长度适中的文本块。

• split_by_token: 按Token数（如LLM上下文窗口限制）机械分割。简单，但可能切断句子或段落。
• split_by_sentence: 按句子分割，对中文需要依赖可靠的句子分割模型（如pysbd或jieba）。
• split_by_paragraph: 按自然段落分割。
• split_by_heading: 按标题分割，非常适合结构清晰的报告、论文、手册。
• recursive_split: 组合多种分割策略，例如先按标题分，如果某个部分还是太长，再按段落或句子分。这是处理长度不一文档的推荐策略。

分割策略实战建议：

1. 分析文档结构：先用extract_text和简单的分割方式预览文档，了解其固有结构（章节、段落）。
2. 匹配LLM窗口：确定你使用的LLM模型的最大上下文窗口（如gpt-4o是128K，但实际使用时为留有余地，单个块最好控制在4K-8K tokens以内）。
3. 保留语义边界：优先选择在自然语义边界处（如标题后、段落末）进行分割，避免在表格中间、公式中间或一个完整意群中间切断。
4. 添加重叠：对于可能因分割而丢失上下文连贯性的情况，可以考虑在分割时设置overlap参数，让相邻块有一小部分重复内容，这能显著提升LLM对边界信息的理解。

3.3 AI智能处理操作符：释放大模型的能力

这是DocETL的“大脑”，通过封装LLM调用，将非结构化文本转化为结构化数据或执行复杂任务。

llm_extract：信息提取与结构化这是最常用的操作符。你定义一个输出模式（JSON Schema），LLM会从文本中抽取信息并填充到这个模式里。

- operator: llm_extract params: model: "gpt-4o-mini" temperature: 0.1 # 低温度保证输出稳定性 prompt: | 从以下产品评论中提取用户提到的产品优点和缺点。 评论：{{chunk_text}} output_schema: type: object properties: product_name: {type: "string", description: "产品名称"} advantages: type: array items: {type: "string"} description: "优点列表" disadvantages: type: array items: {type: "string"} description: "缺点列表" sentiment: {type: "string", enum: ["positive", "neutral", "negative"]}

• {{chunk_text}}：这是DocETL的模板变量，会被自动替换为上游文档块的实际内容。
• output_schema：不仅定义了输出格式，其description和type约束也对LLM起到了重要的引导作用，能有效提高提取准确率。

llm_classify：文本分类适用于情感分析、主题分类、意图识别等场景。

- operator: llm_classify params: model: "claude-3-haiku" categories: ["技术咨询", "账单问题", "账户管理", "产品反馈", "其他"] prompt: | 请将以下客户支持工单内容分类到最合适的类别中。 工单内容：{{chunk_text}}

llm_summarize：摘要总结生成摘要、提炼要点。

- operator: llm_summarize params: model: "gpt-4" max_length: 200 # 限制摘要长度 prompt: | 为以下长篇研究报告撰写一个不超过200字的执行摘要，突出核心发现和建议。 报告：{{chunk_text}}

llm_generate：自由生成用于翻译、改写、扩写、问答等开放式任务。

- operator: llm_generate params: model: "gpt-4o" prompt: | 将以下中文技术文档翻译成英文，保持技术术语准确。 {{chunk_text}}

核心技巧：提示词工程在DocETL中的实践

1. 上下文清晰：在提示词中明确说明文本的来源和背景（如“这是一份法律合同中的‘付款条款’部分”）。
2. 指令具体：避免“提取信息”这种模糊指令。要像给实习生布置任务一样明确，例如“找出所有提及金额的数字，并注明其货币单位（人民币、美元）”。
3. 利用系统角色：虽然DocETL配置中不直接暴露system消息，但你可以将角色设定融入prompt开头，如“你是一名经验丰富的财务审计师，现在需要分析以下报销单据...”。
4. 迭代优化：充分利用DocWrangler的“改进提示词”功能。先跑一个基础版，观察LLM在哪里出错或含糊，然后让AI助手基于这些失败案例帮你重写提示词。
5. 处理长文本：对于超长文档，采用“Map-Reduce”策略：先用llm_summarize或llm_extract对每个块进行局部处理（Map），再用一个llm_generate操作符对所有局部结果进行综合归纳（Reduce）。

3.4 流程控制与高级操作符

filter：数据过滤根据条件过滤文档块。例如，过滤掉长度太短的块，或根据LLM提取的某个字段值进行过滤。

- operator: filter params: condition: "{{length(chunk_text) > 100}}" # 只保留长度大于100字符的块

map：对每个文档应用操作这是一个通用操作符，可以对管道中的数据（通常是列表）中的每个元素应用一个自定义的Python函数或表达式，非常适合进行轻量级的数据转换。

resolve：处理引用与合并这是DocETL的一个高级特性。当LLM在某个块中提到了另一个块的内容（例如，“详见第3.2节”）时，resolve操作符可以智能地找到并嵌入被引用的内容，形成更完整的上下文，再送给LLM处理。这对于处理内部交叉引用频繁的长文档（如技术标准、法律条文）极其有用。

4. 从零搭建一个企业级文档分析管道

让我们通过一个完整的实战案例，将上述所有知识串联起来。假设我们是一家投资研究公司，需要每天自动分析上百份上市公司发布的PDF版年度报告（年报），提取关键财务指标和业务展望。

4.1 环境准备与项目初始化

首先，建立项目目录并安装依赖。

# 1. 创建项目目录 mkdir annual_report_analyzer && cd annual_report_analyzer # 2. 创建虚拟环境（推荐使用uv，更快更轻量） python -m venv venv # 或使用 conda, uv 等 source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 3. 安装DocETL核心包 pip install docetl # 4. 创建必要的目录和文件 mkdir -p configs inputs outputs logs touch configs/pipeline.yaml .env

在.env文件中配置你的LLM API密钥。这里以OpenAI为例，但DocETL通过LiteLLM支持数十种模型。

# .env OPENAI_API_KEY=sk-your-actual-openai-api-key-here # 可选：如果你想使用Azure OpenAI或 Anthropic Claude # AZURE_API_KEY=... # ANTHROPIC_API_KEY=...

4.2 管道配置设计 (configs/pipeline.yaml)

这是整个项目的核心。我们将设计一个多阶段的管道。

# configs/pipeline.yaml pipeline: # 阶段 1: 数据加载 - name: load_reports operator: load_directory params: path: "./inputs/" # 存放所有PDF年报的目录 glob_pattern: "*.pdf" recursive: false # 阶段 2: 文本提取与基础清洗 - name: extract_pdf_text operator: extract_text params: engine: "pdfplumber" # 对包含表格的财报，pdfplumber效果更好 depends_on: load_reports # 声明依赖，确保顺序执行 - name: basic_cleanup operator: clean_text params: remove_extra_whitespace: true strip: true depends_on: extract_pdf_text # 阶段 3: 智能文档分割 - 这是关键！ # 年报通常有固定结构：摘要、管理层讨论、财务报表、附注。 # 我们先尝试按大型标题分割。 - name: split_by_main_sections operator: split_by_heading params: # 匹配中文年报常见的大标题模式 heading_pattern: "^(第[一二三四五六七八九十]+部分|[一二三四五六七八九十]、|摘要|管理层讨论与分析|财务报告|附注|附录)" strip_heading: true # 分割后，标题文本本身会从内容块中移除，单独存储 depends_on: basic_cleanup # 阶段 4: 过滤与分类 # 我们只关心“管理层讨论与分析”和“财务报告”部分 - name: filter_relevant_sections operator: filter params: condition: > {{chunk_metadata.get('heading') in ['管理层讨论与分析', '财务报告', '财务报表']}} depends_on: split_by_main_sections # 阶段 5: 对“管理层讨论与分析”进行摘要和风险提取 - name: analyze_mda_section operator: llm_extract params: model: "gpt-4o" # 对复杂分析使用能力更强的模型 temperature: 0.2 # 使用条件判断，只对特定部分应用此操作 input_filter: "{{chunk_metadata.get('heading') == '管理层讨论与分析'}}" prompt: | 你是一名资深证券分析师。请仔细阅读上市公司年报中的“管理层讨论与分析”部分，并提取以下信息： 1. 公司报告期内主要业务亮点（不超过3条）。 2. 公司面临的主要风险与挑战（列出所有提及的风险点）。 3. 管理层对未来一年的展望基调（选择：乐观、谨慎乐观、中性、谨慎、悲观）。 请以JSON格式输出。 文本内容： {{chunk_text}} output_schema: type: object properties: business_highlights: type: array items: {type: "string"} risk_factors: type: array items: {type: "string"} outlook_tone: type: "string" enum: ["乐观", "谨慎乐观", "中性", "谨慎", "悲观"] depends_on: filter_relevant_sections # 阶段 6: 从“财务报告”部分提取关键指标 - name: extract_financials operator: llm_extract params: model: "gpt-4o-mini" # 提取结构化数据，小模型通常够用且便宜 temperature: 0 input_filter: "{{'财务' in chunk_metadata.get('heading', '')}}" prompt: | 你是一名财务专家。请从以下年报财务部分文本中，精确提取以下财务数据（单位：人民币万元）。 如果某项数据未找到，请填写 null。 注意：请识别表格和段落中的数字。 需要提取的数据： - 营业收入 - 净利润 - 总资产 - 总负债 - 经营活动产生的现金流量净额 - 基本每股收益（元） 文本内容： {{chunk_text}} output_schema: type: object properties: revenue: {type: ["number", "null"]} net_profit: {type: ["number", "null"]} total_assets: {type: ["number", "null"]} total_liabilities: {type: ["number", "null"]} operating_cash_flow: {type: ["number", "null"]} eps: {type: ["number", "null"]} depends_on: filter_relevant_sections # 阶段 7: 结果汇聚与输出 # 将同一份报告的不同分析结果合并，并添加元数据（文件名，报告年份） - name: aggregate_results operator: map params: # 这是一个简化的示例，实际中可能需要更复杂的合并逻辑 # 这里假设每个文件经过前面步骤后，产生两条记录（MDA和财务） expression: | { "filename": batch_metadata["filename"], "year": extract_year_from_filename(batch_metadata["filename"]), # 假设有自定义函数 "mda_analysis": batch_data[0] if batch_data[0].get('business_highlights') else {}, "financial_data": batch_data[1] if batch_data[1].get('revenue') else {} } depends_on: [analyze_mda_section, extract_financials] - name: write_final_output operator: write_jsonl params: output_path: "./outputs/analysis_results_{{timestamp}}.jsonl" depends_on: aggregate_results

4.3 使用DocWrangler进行迭代调试

在将上述复杂配置投入生产前，必须进行充分调试。

1. 启动DocWrangler：按照项目README，在项目根目录配置好.env和website/.env.local后，运行make docker或make run-ui-dev。
2. 导入配置与样本：在浏览器打开localhost:3000/playground，将上面写的pipeline.yaml内容粘贴到配置编辑器。同时，在UI中上传1-2份有代表性的PDF年报作为测试输入。
3. 逐步运行与检查：不要一次性运行整个管道。利用UI的“运行到该步骤”功能，逐个操作符检查输出。
   • 检查extract_pdf_text：文本提取是否完整？表格数据是否丢失？
   • 检查split_by_main_sections：分割点是否准确？是否把“财务报告”和后面的“附注”错误地切在一起了？你可能需要调整heading_pattern正则表达式。
   • 检查llm_extract：LLM提取的数据准确吗？有没有幻觉（Hallucination）？提示词是否需要更精确地约束数字格式（如“单位：万元”）？
4. 优化提示词：如果LLM提取不准，使用UI的“改进提示词”功能，提供几个正确和错误的输出示例，让AI帮你生成更鲁棒的提示词。
5. 导出最终配置：调试满意后，在UI中导出最终的YAML配置。这个配置就是你的“黄金标准”。

4.4 生产部署与自动化执行

调试完成后，就可以用命令行工具进行批量处理了。

# 1. 将调试好的最终配置保存为 production_pipeline.yaml cp configs/pipeline_debugged.yaml configs/production_pipeline.yaml # 2. 使用DocETL CLI运行管道 docetl run --config configs/production_pipeline.yaml --input-dir ./inputs/ --env-file .env # 或者，在Python代码中调用 import asyncio from docetl import run_pipeline async def main(): with open("configs/production_pipeline.yaml", "r") as f: pipeline_config = f.read() # 可以动态覆盖配置中的参数 overrides = { "pipeline": [ {"name": "load_reports", "params": {"path": "/mnt/new_reports/"}} ] } results = await run_pipeline(pipeline_config, overrides=overrides) print(f"处理完成，共处理 {len(results)} 份报告。") if __name__ == "__main__": asyncio.run(main())

生产环境考量：

• 错误处理与重试：DocETL内置了LLM API调用的错误处理和指数退避重试机制。你可以在配置中调整重试次数和退避策略。
• 速率限制与成本控制：通过环境变量或配置控制并发请求数，避免触发API速率限制。同时，关注LLM的Token使用量，gpt-4o-mini的成本远低于gpt-4o，在精度允许的情况下是首选。
• 日志与监控：确保日志输出到文件（如./logs/），便于排查问题。可以集成像prometheus或datadog来监控管道运行状态和API消耗。
• 与工作流调度器集成：将docetl run命令封装进cron任务、AirflowDAG或PrefectFlow中，实现每日/每周的自动化处理。

5. 避坑指南与性能优化实战

在实际使用DocETL构建复杂管道的过程中，我踩过不少坑，也总结出一些优化经验。

5.1 常见问题与排查技巧

问题1：文本提取质量差，特别是PDF中的表格和格式丢失。

• 排查：首先在DocWrangler中单独运行extract_text操作符，查看原始提取文本。如果表格混乱，可能是提取引擎的问题。
• 解决：
  • 尝试切换engine参数，比如从pypdf换成pdfplumber，后者对表格的支持通常更好。
  • 对于极其复杂的PDF（如扫描件），考虑先使用专门的OCR服务（如Azure Form Recognizer、Google Document AI）进行处理，再将OCR结果文本导入DocETL管道。
  • 在clean_text阶段，谨慎使用过于激进的清洗规则，避免误删表格对齐用的空格或制表符。

问题2：文档分割后，语义不完整，影响LLM理解。

• 排查：检查分割后的块。是否一个完整的表格被切成了两半？是否一个概念的解释被分到了两个块里？
• 解决：
  • 调整分割策略：从split_by_token切换到split_by_heading或split_by_paragraph。
  • 使用递归分割：recursive_split可以设置chunk_size和chunk_overlap，先按大结构分，再对过大的块进行二次细分，并保留重叠部分。
  • 后处理合并：在分割后，可以添加一个自定义的map操作符，根据一些启发式规则（如块长度过短、以连接词开头等）将相邻的块重新合并。

问题3：LLM提取结果不稳定，时好时坏。

• 排查：检查LLM操作的temperature参数。如果大于0，每次输出可能会有随机性。检查提示词是否足够明确，output_schema是否定义了严格的enum或format。
• 解决：
  • 降低温度：对于提取任务，将temperature设为0或接近0（如0.1）。
  • 改进提示词：在提示词中提供1-2个清晰的示例（Few-shot Learning）。例如：“请按以下格式提取：... 示例：文本‘...’应输出为 {...}”。
  • 使用更强大的模型：对于逻辑非常复杂或需要深度推理的提取任务，gpt-4o或claude-3-opus的稳定性和准确性通常远高于小模型。
  • 引入验证步骤：在llm_extract后，可以接一个llm_classify或规则校验，判断提取结果是否合理，过滤掉明显错误的结果。

问题4：处理速度慢，尤其是文件很多时。

• 排查：使用time命令或添加日志记录各步骤耗时。瓶颈通常在于：1. 网络I/O（加载远程文件），2. LLM API调用，3. 复杂的文本处理（如基于模型的句子分割）。
• 解决：
  • 并发处理：DocETL支持异步并发。确保你的管道配置中，没有不必要的顺序依赖（depends_on）。对于可以独立处理的不同文件或不同块，运行时引擎会尝试并行执行。
  • 批处理LLM请求：如果多个文档块需要调用相同的LLM操作，可以考虑在map操作符中先收集一批数据，然后进行一次批量API调用（如果LLM API支持），但这需要更精细的控制。
  • 缓存中间结果：对于昂贵的操作（如大模型调用），可以考虑将中间结果（如清洗后的文本、分割后的块）缓存到本地文件或数据库中。DocETL本身不提供内置缓存，但你可以通过将管道拆分为多个子管道并手动管理中间文件来实现。

5.2 高级技巧：自定义操作符与扩展

当内置操作符无法满足需求时，DocETL允许你定义自定义操作符。这是其扩展性的体现。

例如，我们需要一个从PDF中提取并解析所有图片的操作符：

1. 创建自定义Python类：

# custom_operators.py from docetl.operators import BaseOperator from typing import Dict, Any, List import fitz # PyMuPDF class ExtractImagesOperator(BaseOperator): """从PDF文档中提取所有图片并保存到本地。""" def __init__(self, params: Dict[str, Any]): super().__init__(params) self.output_dir = params.get("output_dir", "./extracted_images") async def execute(self, data: List[Any], metadata: Dict[str, Any]) -> (List[Any], Dict[str, Any]): import os os.makedirs(self.output_dir, exist_ok=True) results = [] for doc in data: if not isinstance(doc, dict) or doc.get("format") != "pdf": continue pdf_path = doc["path"] pdf_document = fitz.open(pdf_path) image_info_list = [] for page_num in range(len(pdf_document)): page = pdf_document[page_num] image_list = page.get_images() for img_index, img in enumerate(image_list): xref = img[0] base_image = pdf_document.extract_image(xref) image_bytes = base_image["image"] image_ext = base_image["ext"] image_filename = f"{os.path.basename(pdf_path)}_p{page_num+1}_i{img_index}.{image_ext}" image_path = os.path.join(self.output_dir, image_filename) with open(image_path, "wb") as f: f.write(image_bytes) image_info_list.append({ "page": page_num + 1, "index": img_index, "path": image_path, "ext": image_ext, "size": len(image_bytes) }) pdf_document.close() # 将图片信息作为元数据附加到文档上，或作为新数据输出 doc["extracted_images"] = image_info_list results.append(doc) return results, metadata

2. 在管道配置中引用自定义操作符：

pipeline: - operator: load_directory params: {path: "./inputs/", glob_pattern: "*.pdf"} - operator: custom.ExtractImagesOperator # 通过`custom.`前缀引用 params: output_dir: "./outputs/images/" depends_on: load_directory # ... 后续可以继续处理文本或图片信息

3. 运行管道时注册自定义操作符：

from docetl import run_pipeline from custom_operators import ExtractImagesOperator # 在运行前，将自定义操作符注册到系统中 operators_registry = {"custom.ExtractImagesOperator": ExtractImagesOperator} results = await run_pipeline(pipeline_config, custom_operators=operators_registry)

通过自定义操作符，你可以轻松集成任何Python库，实现OCR、语音转文字、调用内部API等复杂功能，将DocETL打造成完全适应你业务需求的超级工作流引擎。