Instructor 批量处理实战四步跑通多 Provider 结构化抽取【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructorInstructor 是让 LLM 直接产出 Pydantic 结构化对象的 Python 库其批量处理模块把把几百上千条文本一次性抽成结构化数据的成本压到常规 API 的一半左右。本文面向刚接触该项目的开发者如果你正被三件小事卡住——Serverless 环境里不方便落盘临时文件、三家 Provider 的批量请求格式互不相通、批结果要人工逐行解析——那么instructor/batch/目录下的BatchProcessor统一抽象就是为此准备的。快速上手从响应模型到结果落袋的最小链路整条链路只需要四步定义响应模型、生成批请求、提交、轮询取回。默认参数为max_tokens1000、temperature0.1示例里按需覆盖import time from pydantic import BaseModel from instructor.batch import BatchProcessor, extract_results class User(BaseModel): name: str age: int processor BatchProcessor(openai/gpt-4o-mini, User) msgs [[{role: user, content: Hi, Im Alice, 28 years old.}]] processor.create_batch_from_messages(msgs, file_pathreqs.jsonl, max_tokens200) batch_id processor.submit_batch(reqs.jsonl) while processor.get_batch_status(batch_id).get(status) ! completed: time.sleep(10) print(extract_results(processor.get_results(batch_id)))custom_id由库统一生成为request-0、request-1……与messages_list的下标一一对应方便事后把结果对回原始请求。为什么一个处理器能路由到多家 ProviderBatchProcessor.__init__接收的模型字符串必须写成provider/model-name源码用model.split(/, 1)拆出前缀instructor/batch/processor.py格式不对会抛出ValueError: Model string must be in format provider/model-name (e.g. openai/gpt-4 or anthropic/claude-3-sonnet)拆出的前缀交给get_provider工厂函数instructor/batch/providers/init.py它只做一件事根据名字返回对应的 Provider 实例。这个工厂用importlib.util.find_spec做懒加载——只有当openai或anthropic包确实安装时才导入实现否则符号被置为None调用时报OpenAI is not installed/Anthropic is not installed而不是在import instructor时就炸掉。每家 Provider 的实现都要继承BatchProvider抽象基类instructor/batch/providers/base.py它约束了七个必须实现的方法submit_batch、get_status、retrieve_results、download_results、cancel_batch、delete_batch、list_batches。BatchProcessor自身不写任何 HTTP 细节全部委托给这个抽象——所以你的业务代码里只出现processor.xxx一种写法换 Provider 只改模型字符串的前缀。内存批处理怎么开启create_batch_from_messages的第二个参数file_path决定投递方式传字符串就逐行追加写入磁盘 JSONL若文件已存在会先删掉重建传file_pathNone则写入io.BytesIO缓冲区函数返回str | io.BytesIO联合类型。buffer processor.create_batch_from_messages( msgs, file_pathNone, max_tokens150 ) print(len(buffer.getvalue()), bytes) # 查看缓冲区大小 batch_id processor.submit_batch(buffer)file_pathNone分支在写完所有请求后会执行buffer.seek(0)重置读位置instructor/batch/processor.py 第 90 行附近。submit_batch内部直接读取这个缓冲区位置不在 0 就只会上传残片——如果你在提交前自己read()过缓冲区预览内容提交前记得再seek(0)一次。两种方式的取舍如下维度文件方案file_pathreqs.jsonl内存方案file_pathNone清理负担需手动os.remove异常路径容易漏无缓冲区随进程释放磁盘暴露面敏感提示词会短暂留在磁盘不落盘适合安全敏感环境适用场景大批量任务、需要调试与审计留痕ServerlessLambda、Cloud Functions调试便利可直接打开文件检查请求格式只能打印缓冲区内容预览结论Serverless 与安全敏感环境选内存方案需要留痕、复查请求格式的大批量任务选文件方案。仓库示例 examples/batch_api/in_memory_batch_example.py 的compare_file_vs_memory()函数把两条路径并排跑了一遍可对照阅读。三家 Provider 的差异清单统一门面之下三家在请求格式、环境变量与容错行为上并不相同。格式转换发生在BatchRequest.save_to_fileinstructor/batch/request.pyProvider请求格式转换环境变量Key 缺失时OpenAIto_openai_formatPOST /v1/chat/completionsresponse_format.type json_schema且strict: True并递归补additionalProperties: False满足严格模式OPENAI_API_KEY报错退出Anthropicto_anthropic_format把system角色消息抽取合并为顶层system参数Schema 包装成名为extract_data的tool_use工具固定tool_choice {type: tool, name: extract_data}ANTHROPIC_API_KEY报错退出Google测试脚本走内联提交submit_batch(..., use_inlineTrue)不经本地批文件GOOGLE_API_KEY打印警告并降级为模拟模式OpenAI批 API 限速与常规 API 相互独立处理时长通常几小时内完成、24 小时保证内文档建议单任务至少 25,000 条请求以摊薄开销。Anthropic走 Beta 端点client.beta.messages.batches大多数批次 1 小时内完成支持cancel与deleteOpenAI 不支持按 API 删除批次。Google完整实现依赖 GCS需GOOGLE_CLOUD_PROJECT、GCS_BUCKET、GOOGLE_APPLICATION_CREDENTIALS三个环境变量与roles/aiplatform.user、roles/storage.objectUser两个 IAM 角色有 24 小时执行上限且 GCS 桶必须与批任务同区域。命令行接管批任务不想写 Python 时instructor batch提供cancel、create、create-from-file、delete、download-file、list、results七个子命令docs/cli/batch.md。list支持--limit默认 10、--poll、--screen、--live实时刷新表格提供商默认--provider openai旧标志--use-anthropic已废弃官方建议改用--model。表格会展示创建/开始时间、耗时OpenAI 显示 Completed/Failed/TotalAnthropic 显示 Succeeded/Errored/Processing数据来自BatchJobInfo的归一化字段# 列出批任务 instructor batch list --provider openai --limit 10 --live instructor batch list --model anthropic/claude-3-5-sonnet-20241022 # 状态查询与结果拉取 instructor batch status --batch-id batch_123 --model openai/gpt-4o-mini instructor batch results --batch-id batch_123 --output-file results.jsonl \ --model openai/gpt-4o-mini # 两步创建消息文件 - 请求文件 - 提交 instructor batch create --messages-file messages.jsonl \ --model openai/gpt-4o-mini --response-model examples.User \ --output-file batch_requests.jsonl instructor batch create-from-file --file-path batch_requests.jsonl \ --model openai/gpt-4o-mini批结果怎么读联合类型与四个工具函数批结果从不返回裸对象而是统一联合类型instructor/batch/models.pyBatchResult: TypeAlias Union[BatchSuccess[Any], BatchError]BatchSuccess[T]custom_id 解析好的result: Tsuccess: TrueBatchErrorcustom_iderror_typeerror_messageraw_datasuccess: False。解析器parse_resultsinstructor/batch/processor.py按行读 JSONLOpenAI 从response.body.choices[0].message.content取 JSONAnthropic 优先取tool_use块的input、失败则回退解析text块模型校验失败或提供商返回错误都落为BatchError不会中断整批。配套四个工具函数instructor/batch/utils.py函数返回filter_successful(results)List[BatchSuccess[T]]filter_errors(results)List[BatchError]extract_results(results)仅成功项的List[T]get_results_by_custom_id(results){custom_id: BatchResult}字典轮询间隔如何设置仓库示例用 10 秒examples/batch_api/in_memory_batch_example.py测试脚本fetch --poll默认 30 秒、--max-wait默认 600 秒。一个完整循环import time while True: status processor.get_batch_status(batch_id).get(status) if status completed: results processor.get_results(batch_id) break if status in (failed, cancelled, expired): raise SystemExit(fbatch ended with status {status}) print(fstatus{status}, waiting...) time.sleep(10)状态方面各家的原始值会被归一化为BatchStatus六种枚举pending、processing、completed、failed、cancelled、expired。OpenAI 的validating归入pendingin_progress/finalizing归入processingAnthropic 的in_progress映射processing、ended映射completedinstructor/batch/models.py 的from_openai/from_anthropic。判断可下载结果时只认归一化后的completed原始值则保留在BatchJobInfo.raw_status供排查。踩坑与排错速查报错信息触发位置处理办法Error: OPENAI_API_KEY environment variable is not set测试脚本的check_api_keyexport OPENAI_API_KEYyour-keyAnthropic 同理Error: Model must be in format provider/model-name模型字符串缺少/写成openai/gpt-4o-mini这类格式Unsupported provider: xyzget_provider工厂仅识别openai、anthropic前缀Google 由测试脚本特殊处理Missing GCS_BUCKET (Google)Google 完整实现设置GCS_BUCKET等三个 GCS 相关环境变量Permission Denied (Google)IAM 角色不足授予aiplatform.user与storage.objectUser几条容易被忽略的边界提交前若自行读取过 BytesIO 缓冲区必须重新seek(0)再submit_batchOpenAI 批次不支持通过 API 删除delete子命令对 Anthropic 才生效Google 批任务 24 小时执行上限超期归为expiredGCS 桶与任务需同区域单任务条数太少时成本优势不明显非紧急负载才值得走批处理。相关文件批量处理概念文档docs/concepts/batch.mdCLI 批任务命令详解docs/cli/batch.md统一处理器实现instructor/batch/processor.py结果模型与状态枚举instructor/batch/models.py请求格式转换instructor/batch/request.py内存批处理示例examples/batch_api/in_memory_batch_example.py【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/16 18:19:16 · 阅读 1.2k Instructor 批量处理实战：四步跑通多 Provider 结构化抽取
Instructor 批量处理实战四步跑通多 Provider 结构化抽取【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructorInstructor 是让 LLM 直接产出 Pydantic 结构化对象的 Python 库其批量处理模块把把几百上千条文本一次性抽成结构化数据的成本压到常规 API 的一半左右。本文面向刚接触该项目的开发者如果你正被三件小事卡住——Serverless 环境里不方便落盘临时文件、三家 Provider 的批量请求格式互不相通、批结果要人工逐行解析——那么instructor/batch/目录下的BatchProcessor统一抽象就是为此准备的。快速上手从响应模型到结果落袋的最小链路整条链路只需要四步定义响应模型、生成批请求、提交、轮询取回。默认参数为max_tokens1000、temperature0.1示例里按需覆盖import time from pydantic import BaseModel from instructor.batch import BatchProcessor, extract_results class User(BaseModel): name: str age: int processor BatchProcessor(openai/gpt-4o-mini, User) msgs [[{role: user, content: Hi, Im Alice, 28 years old.}]] processor.create_batch_from_messages(msgs, file_pathreqs.jsonl, max_tokens200) batch_id processor.submit_batch(reqs.jsonl) while processor.get_batch_status(batch_id).get(status) ! completed: time.sleep(10) print(extract_results(processor.get_results(batch_id)))custom_id由库统一生成为request-0、request-1……与messages_list的下标一一对应方便事后把结果对回原始请求。为什么一个处理器能路由到多家 ProviderBatchProcessor.__init__接收的模型字符串必须写成provider/model-name源码用model.split(/, 1)拆出前缀instructor/batch/processor.py格式不对会抛出ValueError: Model string must be in format provider/model-name (e.g. openai/gpt-4 or anthropic/claude-3-sonnet)拆出的前缀交给get_provider工厂函数instructor/batch/providers/init.py它只做一件事根据名字返回对应的 Provider 实例。这个工厂用importlib.util.find_spec做懒加载——只有当openai或anthropic包确实安装时才导入实现否则符号被置为None调用时报OpenAI is not installed/Anthropic is not installed而不是在import instructor时就炸掉。每家 Provider 的实现都要继承BatchProvider抽象基类instructor/batch/providers/base.py它约束了七个必须实现的方法submit_batch、get_status、retrieve_results、download_results、cancel_batch、delete_batch、list_batches。BatchProcessor自身不写任何 HTTP 细节全部委托给这个抽象——所以你的业务代码里只出现processor.xxx一种写法换 Provider 只改模型字符串的前缀。内存批处理怎么开启create_batch_from_messages的第二个参数file_path决定投递方式传字符串就逐行追加写入磁盘 JSONL若文件已存在会先删掉重建传file_pathNone则写入io.BytesIO缓冲区函数返回str | io.BytesIO联合类型。buffer processor.create_batch_from_messages( msgs, file_pathNone, max_tokens150 ) print(len(buffer.getvalue()), bytes) # 查看缓冲区大小 batch_id processor.submit_batch(buffer)file_pathNone分支在写完所有请求后会执行buffer.seek(0)重置读位置instructor/batch/processor.py 第 90 行附近。submit_batch内部直接读取这个缓冲区位置不在 0 就只会上传残片——如果你在提交前自己read()过缓冲区预览内容提交前记得再seek(0)一次。两种方式的取舍如下维度文件方案file_pathreqs.jsonl内存方案file_pathNone清理负担需手动os.remove异常路径容易漏无缓冲区随进程释放磁盘暴露面敏感提示词会短暂留在磁盘不落盘适合安全敏感环境适用场景大批量任务、需要调试与审计留痕ServerlessLambda、Cloud Functions调试便利可直接打开文件检查请求格式只能打印缓冲区内容预览结论Serverless 与安全敏感环境选内存方案需要留痕、复查请求格式的大批量任务选文件方案。仓库示例 examples/batch_api/in_memory_batch_example.py 的compare_file_vs_memory()函数把两条路径并排跑了一遍可对照阅读。三家 Provider 的差异清单统一门面之下三家在请求格式、环境变量与容错行为上并不相同。格式转换发生在BatchRequest.save_to_fileinstructor/batch/request.pyProvider请求格式转换环境变量Key 缺失时OpenAIto_openai_formatPOST /v1/chat/completionsresponse_format.type json_schema且strict: True并递归补additionalProperties: False满足严格模式OPENAI_API_KEY报错退出Anthropicto_anthropic_format把system角色消息抽取合并为顶层system参数Schema 包装成名为extract_data的tool_use工具固定tool_choice {type: tool, name: extract_data}ANTHROPIC_API_KEY报错退出Google测试脚本走内联提交submit_batch(..., use_inlineTrue)不经本地批文件GOOGLE_API_KEY打印警告并降级为模拟模式OpenAI批 API 限速与常规 API 相互独立处理时长通常几小时内完成、24 小时保证内文档建议单任务至少 25,000 条请求以摊薄开销。Anthropic走 Beta 端点client.beta.messages.batches大多数批次 1 小时内完成支持cancel与deleteOpenAI 不支持按 API 删除批次。Google完整实现依赖 GCS需GOOGLE_CLOUD_PROJECT、GCS_BUCKET、GOOGLE_APPLICATION_CREDENTIALS三个环境变量与roles/aiplatform.user、roles/storage.objectUser两个 IAM 角色有 24 小时执行上限且 GCS 桶必须与批任务同区域。命令行接管批任务不想写 Python 时instructor batch提供cancel、create、create-from-file、delete、download-file、list、results七个子命令docs/cli/batch.md。list支持--limit默认 10、--poll、--screen、--live实时刷新表格提供商默认--provider openai旧标志--use-anthropic已废弃官方建议改用--model。表格会展示创建/开始时间、耗时OpenAI 显示 Completed/Failed/TotalAnthropic 显示 Succeeded/Errored/Processing数据来自BatchJobInfo的归一化字段# 列出批任务 instructor batch list --provider openai --limit 10 --live instructor batch list --model anthropic/claude-3-5-sonnet-20241022 # 状态查询与结果拉取 instructor batch status --batch-id batch_123 --model openai/gpt-4o-mini instructor batch results --batch-id batch_123 --output-file results.jsonl \ --model openai/gpt-4o-mini # 两步创建消息文件 - 请求文件 - 提交 instructor batch create --messages-file messages.jsonl \ --model openai/gpt-4o-mini --response-model examples.User \ --output-file batch_requests.jsonl instructor batch create-from-file --file-path batch_requests.jsonl \ --model openai/gpt-4o-mini批结果怎么读联合类型与四个工具函数批结果从不返回裸对象而是统一联合类型instructor/batch/models.pyBatchResult: TypeAlias Union[BatchSuccess[Any], BatchError]BatchSuccess[T]custom_id 解析好的result: Tsuccess: TrueBatchErrorcustom_iderror_typeerror_messageraw_datasuccess: False。解析器parse_resultsinstructor/batch/processor.py按行读 JSONLOpenAI 从response.body.choices[0].message.content取 JSONAnthropic 优先取tool_use块的input、失败则回退解析text块模型校验失败或提供商返回错误都落为BatchError不会中断整批。配套四个工具函数instructor/batch/utils.py函数返回filter_successful(results)List[BatchSuccess[T]]filter_errors(results)List[BatchError]extract_results(results)仅成功项的List[T]get_results_by_custom_id(results){custom_id: BatchResult}字典轮询间隔如何设置仓库示例用 10 秒examples/batch_api/in_memory_batch_example.py测试脚本fetch --poll默认 30 秒、--max-wait默认 600 秒。一个完整循环import time while True: status processor.get_batch_status(batch_id).get(status) if status completed: results processor.get_results(batch_id) break if status in (failed, cancelled, expired): raise SystemExit(fbatch ended with status {status}) print(fstatus{status}, waiting...) time.sleep(10)状态方面各家的原始值会被归一化为BatchStatus六种枚举pending、processing、completed、failed、cancelled、expired。OpenAI 的validating归入pendingin_progress/finalizing归入processingAnthropic 的in_progress映射processing、ended映射completedinstructor/batch/models.py 的from_openai/from_anthropic。判断可下载结果时只认归一化后的completed原始值则保留在BatchJobInfo.raw_status供排查。踩坑与排错速查报错信息触发位置处理办法Error: OPENAI_API_KEY environment variable is not set测试脚本的check_api_keyexport OPENAI_API_KEYyour-keyAnthropic 同理Error: Model must be in format provider/model-name模型字符串缺少/写成openai/gpt-4o-mini这类格式Unsupported provider: xyzget_provider工厂仅识别openai、anthropic前缀Google 由测试脚本特殊处理Missing GCS_BUCKET (Google)Google 完整实现设置GCS_BUCKET等三个 GCS 相关环境变量Permission Denied (Google)IAM 角色不足授予aiplatform.user与storage.objectUser几条容易被忽略的边界提交前若自行读取过 BytesIO 缓冲区必须重新seek(0)再submit_batchOpenAI 批次不支持通过 API 删除delete子命令对 Anthropic 才生效Google 批任务 24 小时执行上限超期归为expiredGCS 桶与任务需同区域单任务条数太少时成本优势不明显非紧急负载才值得走批处理。相关文件批量处理概念文档docs/concepts/batch.mdCLI 批任务命令详解docs/cli/batch.md统一处理器实现instructor/batch/processor.py结果模型与状态枚举instructor/batch/models.py请求格式转换instructor/batch/request.py内存批处理示例examples/batch_api/in_memory_batch_example.py【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考