企业官网建设流程全解析

• 人工智能
• RAG
• 大模型

本文围绕 LlamaIndex 的llama_index.program.evaporate模块（对应 API 参考页 evaporate.md）展开，系统讲解DFEvaporateProgram这一核心程序类：它如何借鉴 Evaporate 的两阶段思想，先在训练文本上由 LLM"拟合"出 Python 解析函数，再在推理文本上执行函数批量抽取数据，最终产出结构化 DataFrame。读完本文，你将掌握DFEvaporateProgram的完整 API、底层EvaporateExtractor的沙箱执行原理，并能在自己的 RAG / 文档处理管线中用它完成零标注的字段抽取任务。

一、设计理念：先拟合函数，再执行推理

DFEvaporateProgram的核心思想来自斯坦福 HazyResearch 团队的 Evaporate 项目（源码注释中明确标注了这一出处，见 extractor.py）。与常见的"让 LLM 直接回答"不同，Evaporate 类方案将任务拆成两个阶段：

1. 拟合（fit）阶段：给定若干训练文本节点和待抽取字段名，程序调用 LLM 根据文本样例合成一段 Python 函数，该函数被设计为能稳定地从文本中抽取出指定字段值。
2. 推理（inference）阶段：把拟合好的函数应用到大批量推理文本上，逐节点执行，把每个节点对应的字段值组装成行，最终汇总为 DataFrame。

这种"用 LLM 生成规则、用规则做批量抽取"的方式，可以在不手工编写正则、不做大量人工标注的情况下，把非结构化文本转换成结构化数据，同时避免对每个节点都调用 LLM 的高成本——LLM 只在拟合阶段被调用，推理阶段是纯本地 Python 执行。

在类层次上，DFEvaporateProgram继承自BaseEvaporateProgram（泛型参数为DataFrameRowsOnly），而BaseEvaporateProgram又实现了核心接口BasePydanticProgram（定义于 types.py，要求提供output_cls与__call__）。测试用例 test_program_evaporate.py 正是通过__mro__断言了DFEvaporateProgram是BasePydanticProgram的子类，验证了这条继承链。

二、安装与依赖

Evaporate 程序以独立集成包的形式发布，其依赖声明见 pyproject.toml：

pip install llama-index-program-evaporate

该包当前版本为0.7.0，运行环境要求Python >= 3.10, < 4.0，硬性依赖为pandas与llama-index-core>=0.13.0,<0.15（pandas 用于最终 DataFrame 的组装，llama-index-core 提供 LLM、Node、Settings 等基础组件）。此外，实际运行还需要一个 LLM 后端，例如：

pip install llama-index-llms-openai

完整运行示例可参考官方演示 Notebook：evaporate_program.ipynb。

三、快速上手：用DFEvaporateProgram抽取城市人口

3.1 准备数据节点

假设我们要从维基百科文本中抽取城市人口，先用SimpleDirectoryReader读取文档并切分成节点：

from llama_index.core import SimpleDirectoryReader, Settings from llama_index.llms.openai import OpenAI Settings.llm = OpenAI(temperature=0, model="gpt-3.5-turbo") Settings.chunk_size = 512 city_docs = {} for wiki_title in ["Toronto", "Seattle", "Chicago", "Boston", "Houston"]: city_docs[wiki_title] = SimpleDirectoryReader( input_files=[f"data/{wiki_title}.txt"] ).load_data() city_nodes = {} for wiki_title in city_docs: docs = city_docs[wiki_title] city_nodes[wiki_title] = Settings.node_parser.get_nodes_from_documents(docs)

3.2 创建程序并拟合函数

from llama_index.program.evaporate import DFEvaporateProgram program = DFEvaporateProgram.from_defaults( fields_to_extract=["population"], )

from_defaults是官方推荐的工厂方法（实现见 base.py），其完整参数如下：

随后用训练数据拟合字段解析函数：

program.fit_fields(city_nodes["Toronto"][:1])

fit_fields会为每个字段调用一次拟合（base.py），把字段名映射到生成的函数源码字符串。此时可以查看生成的函数：

print(program.get_function_str("population"))

上面示例会输出类似下面的 LLM 生成代码（注意get_function_str实现于 base.py）：

def get_population_field(text: str): """ Function to extract population. """ # Use regex to find the population field pattern = r'(?<=population of )(\d+,?\d*)' population_field = re.search(pattern, text).group(1) # Return the population field as a single value return int(population_field.replace(',', ''))

3.3 执行推理

seattle_df = program(nodes=city_nodes["Seattle"][:1]) seattle_df # DataFrameRowsOnly(rows=[DataFrameRow(row_values=[749256])])

__call__的入参有两种形式（base.py）：

• nodes=...：直接传入List[BaseNode]；
• texts=...：传入字符串列表，程序内部会用TextNode(text=t)自动包装；
• 两者都不提供时抛出ValueError。

程序会为每个字段在全部节点上执行对应函数，按字段组装成列，最终返回DataFrameRowsOnly对象——其rows字段是DataFrameRow列表，每个DataFrameRow.row_values对应一条节点记录。若要转成 pandas，可使用 df.py 中定义的to_df()：

seattle_df.to_df()

DataFrameRowsOnly.to_df(existing_df=None)在传入已存在 DataFrame 时，会把新行拼接在其后（pd.concat，ignore_index=True），适合增量抽取场景。

四、多值场景：MultiValueEvaporateProgram

DFEvaporateProgram的假设是二维表格结构、每个节点恰好一行。当一段文本需要抽出同字段的多个值时（例如从奥运奖牌表中抽取所有国家名单），应改用MultiValueEvaporateProgram。它同样定义于 base.py，与DFEvaporateProgram的三点关键差异（源码 docstring 中已明确）：

1. 输出按列组织：每个DataFrameRow对应一个字段列而非一行记录；
2. 每列长度可变，不保证每个节点恰好一个值；
3. 内部使用FN_GENERATION_LIST_PROMPT提示词，要求 LLM 生成返回列表的函数（from_defaults中自动切换，见 base.py）。

Notebook 示例中的用法（抽奖牌数场景）：

from llama_index.core.program.predefined import MultiValueEvaporateProgram program = MultiValueEvaporateProgram.from_defaults( fields_to_extract=["countries", "medal_count"], ) program.fit_fields(train_nodes[:1]) result = program(nodes=infer_nodes[:1]) print(result.columns[0].row_values) # 某字段的全部取值

拟合出的函数会形如countries_field = re.findall(r'<a href=".*">(.*)</a>', text)，即一次抽出所有匹配项，返回类型为DataFrameValuesPerColumn（定义见 df.py）。

五、底层原理：EvaporateExtractor的三段式流水线

DFEvaporateProgram并不直接接触 LLM，所有脏活都由EvaporateExtractor承担（extractor.py）。它对外暴露三个核心能力：

5.1identify_fields：字段自动发现

当你不确定该抽哪些字段时，可以只给主题词，让提取器从文本中自动归纳候选字段：

extractor = program.extractor # 通过 program.extractor 属性拿到底层提取器 existing_fields = extractor.identify_fields( city_pop_nodes, topic="population", fields_top_k=4 ) # 示例输出：["seattle metropolitan area's population"]

其实现（extractor.py）流程为：对每个节点调用SchemaIDPrompt提示 LLM 列出文本中与主题相关的属性，再用extract_field_dicts解析 "字段: 值" 列表并做归一化去重，最后按出现频次排序取 Top-K。extract_field_dicts的字段归一化规则（extractor.py）包括：去除空行、去-/_/空格变体、小写化，并要求字段词确实出现在原文中才保留。

5.2extract_fn_from_nodes：从节点合成函数

fit阶段真正调用的方法是extract_fn_from_nodes（extractor.py）。其内部机制：

1. 用get_function_field_from_attribute把字段名清洗成合法 Python 标识符（非字母数字替换为_，见 extractor.py）；
2. 若调用方提供了expected_output，则把期望输出示例拼进提示词，约束函数输出格式；
3. 以FnGeneratePrompt作为 QA 模板，使用ResponseMode.TREE_SUMMARIZE的响应合成器（来自llama_index.core.response_synthesizers）对多个训练节点做树状汇总，得到函数体；
4. 将函数体包装为def get_{function_field}_field(text: str): ...，并做后处理：只保留含return的语句（取第一个 return 之前）、剔除print行、只保留以空格/制表符/def开头的缩进行。

5.3run_fn_on_nodes：沙箱执行生成的代码

LLM 生成的代码是"不可信输入"，因此执行阶段被严格沙箱化（extractor.py）：

• 受限内建函数：_SANDBOX_BUILTINS只开放abs/all/any/bool/dict/enumerate/filter/float/int/len/list/map/max/min/range/str/sum/zip等安全内建，eval、exec、compile、open、__import__、getattr、globals、breakpoint等危险内建全部排除（见 extractor.py）；
• 受限 import：_SANDBOX_ALLOWED_IMPORTS仅允许re/math/datetime/json/string/typing/time/collections/itertools/functools/decimal/fractions/statistics/textwrap/unicodedata/operator（extractor.py），任何其他 import 都会触发ImportError；
• 静态校验：执行前先用ast.parse对生成代码做 AST 遍历（_validate_generated_code，extractor.py），拒绝访问任何双下划线开头的 dunder 名称/属性，拒绝非白名单 import；
• 超时保护：每个节点执行被限制在time_limit(1)即 1 秒内（extractor.py，基于SIGALRM信号实现，超时抛出TimeoutException）；
• 注入文本：每个节点的文本内容作为node_text变量注入沙箱 globals，然后执行__result__ = get_{field}_field(node_text)取结果。

此外，extractor.extract_datapoints_with_fn(nodes, topic, sample_k=5, fields_top_k=5)（extractor.py）提供一条"一键流水线"：随机采样sample_k个节点 →identify_fields发现字段 → 逐字段生成函数 → 全部节点执行 → 按节点组装成字典列表。

六、输出数据模型一览

DataFrameRowsOnly、DataFrameRow等数据模型集中定义在 df.py：

同文件还提供了两个基于 Function Calling 的通用解析程序：DFFullProgram让 LLM 同时输出列 schema 与数据行（DEFAULT_FULL_DF_PARSER_TMPL），DFRowsProgram在给定列 schema 时只抽取行（DEFAULT_ROWS_DF_PARSER_TMPL），二者都通过FunctionCallingProgram落地（df.py）。

七、提示词模板：决定抽取质量的关键

所有提示词模板集中定义于 prompts.py，其注释注明完整版权归属 Evaporate 仓库。理解它们有助于针对你的领域微调：

• SCHEMA_ID_PROMPT_TMPL：字段识别提示，内含加拿大信息框与用药记录两个 few-shot 示例，要求"列出文本中恰好提到的、与主题相关的所有属性"，输出- 属性: 值列表格式（prompts.py）；
• FN_GENERATION_PROMPT_TMPL：单值函数生成提示，明确要求"返回单个值（string/int/float）而非列表""必须包含 return 语句"，并以import re+def get_{function_field}_field(text: str):给出函数签名骨架（prompts.py）；
• FN_GENERATION_LIST_PROMPT_TMPL：多值版提示，要求"返回值为列表（单元素也要包成列表）"，MultiValueEvaporateProgram默认使用它（prompts.py）；
• DEFAULT_EXPECTED_OUTPUT_PREFIX_TMPL：期望输出引导语，约束生成的函数不得返回与期望不同的输出；
• DEFAULT_FIELD_EXTRACT_QUERY_TMPL：默认抽取查询，要求"抽取整个 {field} 字段但不要其他元数据，返回列表"。

这些模板均可通过from_defaults的schema_id_prompt、fn_generate_prompt、field_extract_query_tmpl参数替换，从而适配自定义领域。

八、工程化验证与使用限制

• 继承关系验证：tests/test_program_evaporate.py断言DFEvaporateProgram的 MRO 中包含BasePydanticProgram，保证其满足 LlamaIndex 程序接口约定。
• 调用方式：fit_fields是同步方法，拟合需在推理前完成；nodes_to_fit传入后会在构造时自动拟合。__call__必须提供nodes或texts。
• 适用限制：
  • DFEvaporateProgram假定"每个节点 = 一行"，文本拆分（chunk_size）会直接影响行粒度，示例中采用chunk_size=512；
  • 生成函数的可靠性取决于训练节点的代表性，建议训练与推理文本格式一致；
  • LLM 生成代码受白名单 import 与 1 秒超时约束，复杂的字段抽取可能需要expected_output提供示例引导；
  • 该程序依赖一个可用的 LLM 后端（通过Settings.llm或llm参数注入）。

九、小结

DFEvaporateProgram把"LLM 编写抽取规则"与"本地批量执行"解耦，为 LlamaIndex 的文档处理平台提供了一条低成本、可复用的非结构化→结构化抽取路径。配合MultiValueEvaporateProgram处理多值场景、EvaporateExtractor.identify_fields自动发现字段、以及严格的双层沙箱（受限内建 + 白名单 import + AST 静态校验 + 超时控制），你可以在不依赖人工正则或大规模标注的情况下，稳定地从海量文本节点中抽取结构化数据并组装成 DataFrame。深入阅读源码可重点参考 base.py、extractor.py、df.py 与 prompts.py，并对照 evaporate_program.ipynb 中的完整演示进行复现。

• 人工智能
• RAG
• 大模型