拓冰网站定制 
txtai 工作流数据导出指南ExportTask 将任务输出写入 CSV 与 Excel【免费下载链接】txtai All-in-one AI framework for semantic search, LLM orchestration and language model workflows项目地址: https://gitcode.com/GitHub_Trending/tx/txtaiExportTask 是 txtai 工作流体系中的一个专用任务Task负责把工作流中前序任务产生的结果批量导出为 CSV 或 Excelxlsx文件。它适用于语义检索、RAG、摘要等流水线结束后需要落盘存档、人工审阅或交付给下游系统的场景也可以作为中间环节在导出的同时把结果继续传递给后续任务。读完本文你将掌握 ExportTask 的 Python 与配置YAML两种创建方式、output与timestamp两个关键参数的语义、底层基于 Pandas 的写入实现原理以及如何通过workflow安装选项正确启用该功能。ExportTask 是什么ExportTask 是 txtai 内置的任务类型之一其定位非常单一将任务输出导出为 CSV 或 Excel。它本身不做复杂的数据变换而是承接工作流中前一个任务例如摘要、实体抽取、相似度计算的产出逐元素整理后写入本地文件。从源码看ExportTask 继承了通用的Task基类见 src/python/txtai/workflow/task/base.py基类已经承担了输入过滤select、解包unpack、并发执行concurrency、多 action 结果合并merge等通用职责ExportTask 只需通过register方法补充自己的专属参数output、timestamp并重写__call__完成写盘动作见 src/python/txtai/workflow/task/export.py。ExportTask 对输入数据的结构没有严格要求元素可以是普通字符串、元组或字典。当元素为字典时字典的键会成为导出表格的列名这一点在后面的测试用例中可以得到印证当元素为列表或元组时Pandas 会自动生成数字列索引。快速上手Python 方式创建最简单的用法是直接把 ExportTask 放入Workflow需要显式指定输出文件路径from txtai.workflow import ExportTask, Workflow # 指定输出路径工作流执行时会将任务结果写入该文件 workflow Workflow([ExportTask(outputexport.csv)]) workflow([Input 1, Input 2])执行后当前目录下会生成export.csv每一行对应一个输入元素。需要注意原文档示例中的导入语句写的是from txtai.workflow import FileTask, Workflow但实际使用的是ExportTask请按上文修正为导入ExportTask两者都由 src/python/txtai/workflow/task/init.py 统一导出可直接from txtai.workflow import ExportTask。ExportTask 返回的是原始任务输出outputs因此它完全可以作为工作流的中间任务使用——先落盘一份副本再把同样的数据交给下一个任务继续处理实现边导出、边流转。配置驱动YAML 方式创建与 txtai 其他组件一致ExportTask 也可以通过工作流配置文件声明式创建。任务类型名export由 src/python/txtai/workflow/task/factory.py 中的TaskFactory.get解析本地任务名会被自动拼接为xxxTask类名export→ExportTask再通过配置字典实例化workflow: name: tasks: - task: export output: export.csv timestamp: true其中task指定任务类型为exportoutput与timestamp会作为关键字参数传入ExportTask。这种方式非常适合把导出逻辑固化到工作流配置中与 API 部署见 docs/api/index.md配合实现配置文件即工作流。参数详解ExportTask 通过register方法接收两个专属参数见 src/python/txtai/workflow/task/export.py参数类型默认值说明outputstrNone输出文件路径。扩展名为.xlsx时使用 Pandas 写入 Excel其余扩展名一律按 CSV 处理timestampboolNone为True时在文件名扩展名之前插入当前 UTC 时间戳避免多次运行互相覆盖register方法还承担了依赖校验职责如果环境中未安装 Pandas会直接抛出ImportError提示信息为ExportTask is not available - install workflow extra to enable。此外ExportTask 作为Task子类同样继承基类的通用参数包括action、select、unpack、column、merge、initialize、finalize、concurrency、onetomany等完整签名见 src/python/txtai/workflow/task/base.py可用于精细化控制参与导出的数据范围与执行方式。底层实现原理ExportTask 的执行逻辑非常直观核心代码位于 src/python/txtai/workflow/task/export.py可拆解为四个步骤第一步执行任务本体。__call__首先调用基类的__call__得到处理后的outputs这一步保证了与基类任务调度机制的兼容。第二步解析输出扩展名。通过os.path.splitext拆分output取扩展名并转为小写用于决定写入格式parts list(os.path.splitext(output)) extension parts[-1].lower()第三步按需附加时间戳。当timestampTrue时使用 UTC 时间生成格式为%Y%m%dT%H%M%SZ的字符串例如20260914T034705Z并插入到扩展名之前timestamp datetime.datetime.now(datetime.timezone.utc).strftime(%Y%m%dT%H%M%SZ) parts[-1] timestamp parts[-1] output ..join(parts)例如export.csv会变成export.20260914T034705Z.csv。使用 UTC 时间而非本地时间保证了跨时区部署时文件命名的一致性。第四步按扩展名写入文件。扩展名为.xlsx时调用pd.DataFrame(outputs).to_excel(output, indexFalse)否则调用pd.DataFrame(outputs).to_csv(output, indexFalse)。indexFalse意味着 Pandas 的默认行索引不会被写入文件导出的表格只有数据列。值得注意的是CSV 分支是兜底逻辑——除了.xlsx之外的一切扩展名.csv、.tsv等都会走to_csv。最后方法返回原始outputs从而保持任务链的输入输出一致性。依赖与安装前提ExportTask 依赖 Pandas写入 Excel.xlsx还需要openpyxl引擎。这两者都位于 txtai 的workflow可选依赖组中见 setup.pyextras[workflow] [ apache-libcloud3.3.1, croniter1.2.0, openpyxl3.0.9, pandas1.1.0, pillow7.1.2, requests2.26.0, xmltodict0.12.0, ]因此使用前需要安装pip install txtai[workflow]如果只安装基础版 txtai 而未安装workflow扩展实例化 ExportTask 时会因pandas导入失败而抛出上文提到的ImportError见 src/python/txtai/workflow/task/export.py 的顶层条件导入逻辑。该检查在register阶段完成也就是在Task.__init__调用register(**kwargs)时触发见 src/python/txtai/workflow/task/base.py。测试用例验证仓库自带的单元测试 test/python/testworkflow.pytestExportWorkflow完整覆盖了 ExportTask 的三种典型场景可直接作为使用范本Excel 导出ExportTask(outputpath)处理字典列表[{id: 1, text: Sentence 1}, {id: 2, text: Sentence 2}]后断言生成的.xlsx文件非空CSV 导出同样的输入与outputexport.csv断言生成的 CSV 文件非空带时间戳的 CSV 导出ExportTask(outputpath, timestampTrue)执行后通过glob匹配export-timestamp*.csv找到实际生成的时间戳文件并断言其非空。测试同时印证了两个事实其一字典元素会按 key 展开为表格列id、text其二时间戳机制确实会把文件名改写为带YYYYMMDDTHHMMSSZ后缀的形式。实战组合导出语义检索结果ExportTask 的典型价值在于把工作流终点落盘。例如将文档数据经过分词、向量化后把检索结果导出为 CSV 存档from txtai.embeddings import Embeddings from txtai.workflow import ExportTask, Task, Workflow embeddings Embeddings(pathsentence-transformers/all-MiniLM-L6-v2) # 第一个任务执行语义检索第二个任务导出结果 workflow Workflow([ Task(lambda x: [embeddings.search(q)[0] for q in x]), ExportTask(outputresults.xlsx, timestampTrue), ]) list(workflow([query 1, query 2]))结合timestampTrue每次运行都会生成独立命名的文件方便留痕比对而 ExportTask 透传输出的特性也让后续可以继续追加StorageTask云存储等任务将导出文件进一步归档完整任务列表见 docs/workflow/task/index.md。小结ExportTask 是 txtai 工作流中最轻量也最实用的收尾任务之一两个参数、一段基于 Pandas 的写盘逻辑即可把任意工作流产出固化为 CSV 或 Excel。理解其output扩展名驱动的格式选择、timestamp的 UTC 命名规则以及workflow扩展的依赖前提你就可以在语义搜索、RAG、批处理等各类流水线中安全、可复现地完成数据导出。【免费下载链接】txtai All-in-one AI framework for semantic search, LLM orchestration and language model workflows项目地址: https://gitcode.com/GitHub_Trending/tx/txtai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 txtai 工作流数据导出指南：ExportTask 将任务输出写入 CSV 与 Excel
txtai 工作流数据导出指南ExportTask 将任务输出写入 CSV 与 Excel【免费下载链接】txtai All-in-one AI framework for semantic search, LLM orchestration and language model workflows项目地址: https://gitcode.com/GitHub_Trending/tx/txtaiExportTask 是 txtai 工作流体系中的一个专用任务Task负责把工作流中前序任务产生的结果批量导出为 CSV 或 Excelxlsx文件。它适用于语义检索、RAG、摘要等流水线结束后需要落盘存档、人工审阅或交付给下游系统的场景也可以作为中间环节在导出的同时把结果继续传递给后续任务。读完本文你将掌握 ExportTask 的 Python 与配置YAML两种创建方式、output与timestamp两个关键参数的语义、底层基于 Pandas 的写入实现原理以及如何通过workflow安装选项正确启用该功能。ExportTask 是什么ExportTask 是 txtai 内置的任务类型之一其定位非常单一将任务输出导出为 CSV 或 Excel。它本身不做复杂的数据变换而是承接工作流中前一个任务例如摘要、实体抽取、相似度计算的产出逐元素整理后写入本地文件。从源码看ExportTask 继承了通用的Task基类见 src/python/txtai/workflow/task/base.py基类已经承担了输入过滤select、解包unpack、并发执行concurrency、多 action 结果合并merge等通用职责ExportTask 只需通过register方法补充自己的专属参数output、timestamp并重写__call__完成写盘动作见 src/python/txtai/workflow/task/export.py。ExportTask 对输入数据的结构没有严格要求元素可以是普通字符串、元组或字典。当元素为字典时字典的键会成为导出表格的列名这一点在后面的测试用例中可以得到印证当元素为列表或元组时Pandas 会自动生成数字列索引。快速上手Python 方式创建最简单的用法是直接把 ExportTask 放入Workflow需要显式指定输出文件路径from txtai.workflow import ExportTask, Workflow # 指定输出路径工作流执行时会将任务结果写入该文件 workflow Workflow([ExportTask(outputexport.csv)]) workflow([Input 1, Input 2])执行后当前目录下会生成export.csv每一行对应一个输入元素。需要注意原文档示例中的导入语句写的是from txtai.workflow import FileTask, Workflow但实际使用的是ExportTask请按上文修正为导入ExportTask两者都由 src/python/txtai/workflow/task/init.py 统一导出可直接from txtai.workflow import ExportTask。ExportTask 返回的是原始任务输出outputs因此它完全可以作为工作流的中间任务使用——先落盘一份副本再把同样的数据交给下一个任务继续处理实现边导出、边流转。配置驱动YAML 方式创建与 txtai 其他组件一致ExportTask 也可以通过工作流配置文件声明式创建。任务类型名export由 src/python/txtai/workflow/task/factory.py 中的TaskFactory.get解析本地任务名会被自动拼接为xxxTask类名export→ExportTask再通过配置字典实例化workflow: name: tasks: - task: export output: export.csv timestamp: true其中task指定任务类型为exportoutput与timestamp会作为关键字参数传入ExportTask。这种方式非常适合把导出逻辑固化到工作流配置中与 API 部署见 docs/api/index.md配合实现配置文件即工作流。参数详解ExportTask 通过register方法接收两个专属参数见 src/python/txtai/workflow/task/export.py参数类型默认值说明outputstrNone输出文件路径。扩展名为.xlsx时使用 Pandas 写入 Excel其余扩展名一律按 CSV 处理timestampboolNone为True时在文件名扩展名之前插入当前 UTC 时间戳避免多次运行互相覆盖register方法还承担了依赖校验职责如果环境中未安装 Pandas会直接抛出ImportError提示信息为ExportTask is not available - install workflow extra to enable。此外ExportTask 作为Task子类同样继承基类的通用参数包括action、select、unpack、column、merge、initialize、finalize、concurrency、onetomany等完整签名见 src/python/txtai/workflow/task/base.py可用于精细化控制参与导出的数据范围与执行方式。底层实现原理ExportTask 的执行逻辑非常直观核心代码位于 src/python/txtai/workflow/task/export.py可拆解为四个步骤第一步执行任务本体。__call__首先调用基类的__call__得到处理后的outputs这一步保证了与基类任务调度机制的兼容。第二步解析输出扩展名。通过os.path.splitext拆分output取扩展名并转为小写用于决定写入格式parts list(os.path.splitext(output)) extension parts[-1].lower()第三步按需附加时间戳。当timestampTrue时使用 UTC 时间生成格式为%Y%m%dT%H%M%SZ的字符串例如20260914T034705Z并插入到扩展名之前timestamp datetime.datetime.now(datetime.timezone.utc).strftime(%Y%m%dT%H%M%SZ) parts[-1] timestamp parts[-1] output ..join(parts)例如export.csv会变成export.20260914T034705Z.csv。使用 UTC 时间而非本地时间保证了跨时区部署时文件命名的一致性。第四步按扩展名写入文件。扩展名为.xlsx时调用pd.DataFrame(outputs).to_excel(output, indexFalse)否则调用pd.DataFrame(outputs).to_csv(output, indexFalse)。indexFalse意味着 Pandas 的默认行索引不会被写入文件导出的表格只有数据列。值得注意的是CSV 分支是兜底逻辑——除了.xlsx之外的一切扩展名.csv、.tsv等都会走to_csv。最后方法返回原始outputs从而保持任务链的输入输出一致性。依赖与安装前提ExportTask 依赖 Pandas写入 Excel.xlsx还需要openpyxl引擎。这两者都位于 txtai 的workflow可选依赖组中见 setup.pyextras[workflow] [ apache-libcloud3.3.1, croniter1.2.0, openpyxl3.0.9, pandas1.1.0, pillow7.1.2, requests2.26.0, xmltodict0.12.0, ]因此使用前需要安装pip install txtai[workflow]如果只安装基础版 txtai 而未安装workflow扩展实例化 ExportTask 时会因pandas导入失败而抛出上文提到的ImportError见 src/python/txtai/workflow/task/export.py 的顶层条件导入逻辑。该检查在register阶段完成也就是在Task.__init__调用register(**kwargs)时触发见 src/python/txtai/workflow/task/base.py。测试用例验证仓库自带的单元测试 test/python/testworkflow.pytestExportWorkflow完整覆盖了 ExportTask 的三种典型场景可直接作为使用范本Excel 导出ExportTask(outputpath)处理字典列表[{id: 1, text: Sentence 1}, {id: 2, text: Sentence 2}]后断言生成的.xlsx文件非空CSV 导出同样的输入与outputexport.csv断言生成的 CSV 文件非空带时间戳的 CSV 导出ExportTask(outputpath, timestampTrue)执行后通过glob匹配export-timestamp*.csv找到实际生成的时间戳文件并断言其非空。测试同时印证了两个事实其一字典元素会按 key 展开为表格列id、text其二时间戳机制确实会把文件名改写为带YYYYMMDDTHHMMSSZ后缀的形式。实战组合导出语义检索结果ExportTask 的典型价值在于把工作流终点落盘。例如将文档数据经过分词、向量化后把检索结果导出为 CSV 存档from txtai.embeddings import Embeddings from txtai.workflow import ExportTask, Task, Workflow embeddings Embeddings(pathsentence-transformers/all-MiniLM-L6-v2) # 第一个任务执行语义检索第二个任务导出结果 workflow Workflow([ Task(lambda x: [embeddings.search(q)[0] for q in x]), ExportTask(outputresults.xlsx, timestampTrue), ]) list(workflow([query 1, query 2]))结合timestampTrue每次运行都会生成独立命名的文件方便留痕比对而 ExportTask 透传输出的特性也让后续可以继续追加StorageTask云存储等任务将导出文件进一步归档完整任务列表见 docs/workflow/task/index.md。小结ExportTask 是 txtai 工作流中最轻量也最实用的收尾任务之一两个参数、一段基于 Pandas 的写盘逻辑即可把任意工作流产出固化为 CSV 或 Excel。理解其output扩展名驱动的格式选择、timestamp的 UTC 命名规则以及workflow扩展的依赖前提你就可以在语义搜索、RAG、批处理等各类流水线中安全、可复现地完成数据导出。【免费下载链接】txtai All-in-one AI framework for semantic search, LLM orchestration and language model workflows项目地址: https://gitcode.com/GitHub_Trending/tx/txtai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考