尧图网站建设 
基于 Instructor 与 asyncio 的 LLM 并发处理实战asyncio.gather 与 asyncio.as_completed【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor导读本指南围绕 Instructor 仓库中docs/blog/posts/learn-async.md一文展开系统讲解如何利用 Pythonasyncio的asyncio.gather与asyncio.as_completed对 LLM 结构化抽取任务进行高效并发处理。文中将给出完整可运行的异步客户端配置、四种核心处理模式顺序基线、并发收集、流式完成、信号量限流的代码与性能对比并结合仓库中的examples/learn-async/run.py与examples/asyncio-benchmarks/run.py源码补充错误处理、超时控制、进度跟踪、分批处理等生产级进阶模式。读完后你将掌握在大批量 LLM 抽取场景下正确选择并发策略、实施限流与监控的方法。一、理解 asyncio.gather 与 asyncio.as_completedPython 的asyncio库为并发执行提供了两种强大的手段它们的核心差异决定了适用场景asyncio.gather并发执行所有任务并按输入顺序返回结果。适合需要保持结果顺序、且要求全部任务成功完成的场景。asyncio.as_completed按完成先后返回结果不保证输入顺序。适合希望尽快开始处理已完成任务、面向大数据集或流式输出的场景。两种方式都远优于串行顺序处理。需要特别注意的是asyncio.gather会在某个任务抛出异常时立即向上传播并中断整体等待而asyncio.as_completed要求你在循环内逐个await task并自行处理可能的异常——这一点在后续“生产级进阶模式”中会结合源码进一步展开。二、完整环境搭建创建 Instructor 异步客户端Instructor 支持同步与异步两种客户端。现代推荐写法是使用统一的from_provider接口通过async_clientTrue直接获得异步客户端参见 from_provider 指南import instructor from pydantic import BaseModel # 创建异步客户端 client instructor.from_provider(openai/gpt-4o-mini, async_clientTrue) class Person(BaseModel): name: str age: int occupation: str async def extract_person(text: str) - Person: 使用 LLM 从文本中抽取人物信息。 return await client.create( modelgpt-4o-mini, response_modelPerson, messages[{role: user, content: fExtract person info: {text}}], )async_client参数是from_provider的核心开关。从源码结构看instructor/v2/auto_client.py中的各个 provider 工厂函数约第 27–101 行都接收async_client: bool并在其内部根据该布尔值选择返回同步 SDK 客户端还是Async客户端最终统一由client.create(...)提供一致的接口。仓库中的历史示例 examples/learn-async/run.py 采用的是更早期的apatch风格instructor.apatch(AsyncOpenAI())调用点为client.chat.completions.create(...)import time import asyncio import instructor from pydantic import BaseModel from openai import AsyncOpenAI client instructor.apatch(AsyncOpenAI()) class Person(BaseModel): name: str age: int async def extract_person(text: str) - Person: return await client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: text}], response_modelPerson, )两种写法底层都通过 patch 机制为 SDK 客户端注入了response_model结构化抽取能力async_clientTrue是当前推荐方式而apatch仍是向后兼容的可用路径。在进入并发模式之前先准备一份待处理的样本数据集下文所有示例共用dataset [ John Smith is a 30-year-old software engineer, Sarah Johnson is a 25-year-old data scientist, Mike Davis is a 35-year-old product manager, Lisa Wilson is a 28-year-old UX designer, Tom Brown is a 32-year-old DevOps engineer, Emma Garcia is a 27-year-old frontend developer, David Lee is a 33-year-old backend developer, ]三、方法一顺序处理基线顺序处理是最简单的写法逐个await每次等待前一次完成后再发起下一次请求。它是衡量并发收益的基线。async def sequential_processing() - List[Person]: 逐条处理——最慢的方式。 start_time time.time() persons [] for text in dataset: person await extract_person(text) persons.append(person) print(fProcessed: {person.name}) end_time time.time() print(fSequential processing took: {end_time - start_time:.2f} seconds) return persons # 运行方式persons await sequential_processing()在示例输出中examples/learn-async/run.py 的运行注释7 条数据顺序处理耗时约6.17 秒。每次 HTTP 往返的延迟被完全串行叠加当数据量增大时耗时线性增长。四、方法二asyncio.gather 并发处理asyncio.gather把所有协程包装成任务后一次性并发调度await等待全部完成后返回结果列表结果顺序与输入顺序严格一致async def gather_processing() - List[Person]: 并发处理全部条目按输入顺序返回。 start_time time.time() # 为所有条目创建任务 tasks [extract_person(text) for text in dataset] # 并发执行所有任务 persons await asyncio.gather(*tasks) end_time time.time() print(fasyncio.gather took: {end_time - start_time:.2f} seconds) # 结果保持原始顺序 for person in persons: print(fProcessed: {person.name}) return persons # 运行方式persons await gather_processing()要点解析tasks [extract_person(text) for text in dataset]只是创建协程对象列表真正的调度发生在asyncio.gather(*tasks)通过*解包一次性传入所有任务时返回的persons列表下标与dataset一一对应方便按序落库或对齐下游处理由于所有请求几乎同时发出总耗时约等于单次请求的最长耗时示例中约0.85 秒而非各请求耗时之和。五、方法三asyncio.as_completed 流式处理asyncio.as_completed接受一个可等待对象列表返回一个按完成顺序产出任务的迭代器。每完成一个请求就可以立刻await并处理其结果async def as_completed_processing() - List[Person]: 并发处理条目按完成顺序消费结果。 start_time time.time() persons [] # 为所有条目创建任务 tasks [extract_person(text) for text in dataset] # 按完成顺序处理结果 for task in asyncio.as_completed(tasks): person await task persons.append(person) print(fCompleted: {person.name}) end_time time.time() print(fasyncio.as_completed took: {end_time - start_time:.2f} seconds) return persons # 运行方式persons await as_completed_processing()与gather相比as_completed的核心价值在于边完成边消费在示例输出中它耗时约0.95 秒但在这段时间内每个结果一旦就绪就会被立即打印和追加无需等待最慢的那个请求。对于结果需要逐条写入数据库、推送消息队列或渲染进度的应用这能显著降低端到端的首条结果延迟。六、方法四基于信号量的限流处理并发度拉满虽然快但会给上游 API 带来压力也容易触发速率限制rate limit。asyncio.Semaphore可以在保持并发框架的同时把同时进行的请求数量约束在指定上限async def rate_limited_extract_person( text: str, semaphore: asyncio.Semaphore ) - Person: 带限流的人物信息抽取。 async with semaphore: return await extract_person(text) async def rate_limited_gather(concurrency_limit: int 3) - List[Person]: 使用 asyncio.gather 配合受控并发度处理。 start_time time.time() # 创建信号量限制并发请求数 semaphore asyncio.Semaphore(concurrency_limit) # 创建带限流的任务 tasks [rate_limited_extract_person(text, semaphore) for text in dataset] # 在限流约束下执行 persons await asyncio.gather(*tasks) end_time time.time() print( fRate-limited gather (limit{concurrency_limit}) took: {end_time - start_time:.2f} seconds ) return persons async def rate_limited_as_completed(concurrency_limit: int 3) - List[Person]: 使用 asyncio.as_completed 配合受控并发度处理。 start_time time.time() persons [] # 创建信号量限制并发请求数 semaphore asyncio.Semaphore(concurrency_limit) # 创建带限流的任务 tasks [rate_limited_extract_person(text, semaphore) for text in dataset] # 按完成顺序处理结果 for task in asyncio.as_completed(tasks): person await task persons.append(person) print(fRate-limited completed: {person.name}) end_time time.time() print( fRate-limited as_completed (limit{concurrency_limit}) took: {end_time - start_time:.2f} seconds ) return persons # 运行方式 # persons await rate_limited_gather(concurrency_limit2) # persons await rate_limited_as_completed(concurrency_limit2)信号量的工作方式async with semaphore:进入时若已并发请求数达到上限例如 2后续协程会在此处挂起等待直到某个正在执行的任务退出信号量上下文。示例输出显示concurrency_limit2时两种限流方法耗时约3.04 秒 / 3.26 秒——比全并发慢但远快于串行且对 API 更友好。七、性能对比与选型指南以下为 7 条数据在示例运行中观察到的典型耗时来自 examples/learn-async/run.py 的注释输出实际数值会随模型、网络和 API 状态波动仅作量级参考方法执行时间并发度适用场景Sequential顺序6.17 秒1基线对照asyncio.gather0.85 秒7追求速度、需要有序结果asyncio.as_completed0.95 秒7流式消费结果Rate-limited gather限流 gather3.04 秒2对 API 友好Rate-limited as_completed限流流式3.26 秒2流式 限流何时使用 asyncio.gather结果必须与输入保持相同顺序要求所有任务都成功完成任一失败即整体失败可接受追求最快的整体执行时间内存占用不是首要考量需要同时持有全部结果。何时使用 asyncio.as_completed希望结果一到达就立刻处理流式写入、逐条上报结果顺序无关紧要面向超大数据集希望尽早开始消费以控制内存与等待时间需要实时的进度反馈。何时必须限流面对 API 速率限制如429错误需要对外部服务保持礼貌、避免封禁控制资源消耗与成本构建生产级应用时几乎总是需要配合重试策略。八、生产级进阶模式仓库中 examples/asyncio-benchmarks/run.py 在前述四种基础模式之上进一步提供了面向生产环境的五种进阶模式下面逐一说明。8.1 错误容错gather 的 return_exceptionsasyncio.gather默认任一任务抛异常会立刻中断整体。传入return_exceptionsTrue后异常会被当作普通返回值收进结果列表由你在循环内逐项甄别async def robust_gather_processing() - tuple[list[Person], float]: 带错误处理并发抽取。 start_time time.time() tasks [extract_person(text) for text in dataset] # 异常不再直接中断而是作为返回值 results await asyncio.gather(*tasks, return_exceptionsTrue) persons [] for i, result in enumerate(results): if isinstance(result, Exception): print(fError processing item {i}: {result}) else: persons.append(result) end_time time.time() print(fRobust gather processing took: {end_time - start_time:.2f} seconds) return persons, end_time - start_time这样单条抽取失败不会拖垮整批任务失败项可以被单独记录或后续重试。8.2 整体超时asyncio.wait_for给整批并发操作设置硬性截止时间超时后取消任务async def timeout_gather_processing( timeout_seconds: float 30.0, ) - tuple[list[Person], float]: 带超时的并发处理。 start_time time.time() tasks [extract_person(text) for text in dataset] try: persons await asyncio.wait_for( asyncio.gather(*tasks), timeouttimeout_seconds ) end_time time.time() print(fTimeout gather processing took: {end_time - start_time:.2f} seconds) return persons, end_time - start_time except asyncio.TimeoutError: end_time time.time() print( fProcessing timed out after {timeout_seconds} seconds (took {end_time - start_time:.2f}s) ) return [], end_time - start_time关于超时与重试的更完整语义如timeout同时转发给 SDK 并作为验证重试的 elapsed 停止条件、asyncio.wait_for只保证取消等待而不保证远端停止计算可参考仓库中的 重试与 Tenacity 指南。8.3 进度跟踪配合as_completed可以精确统计已完成条数输出实时百分比async def progress_tracking_processing() - tuple[list[Person], float]: 带进度跟踪的并发处理。 start_time time.time() persons [] total_items len(dataset) completed 0 tasks [extract_person(text) for text in dataset] for task in asyncio.as_completed(tasks): person await task persons.append(person) completed 1 print(fProgress: {completed}/{total_items} ({completed / total_items * 100:.1f}%)) end_time time.time() print(fProgress tracking processing took: {end_time - start_time:.2f} seconds) return persons, end_time - start_time这正是as_completed相对gather的差异化优势场景gather要等全部完成才能拿到任何结果而as_completed天然支持增量反馈。8.4 分批处理Chunking当数据集极大、内存或限流要求严格时可以把数据切成若干 chunk每批内部用gather并发批与批之间串行兼顾并发度与资源控制async def chunked_processing(chunk_size: int 3) - tuple[list[Person], float]: 分批并发处理控制内存与速率限制。 start_time time.time() all_persons [] for i in range(0, len(dataset), chunk_size): chunk dataset[i : i chunk_size] print(fProcessing chunk {i // chunk_size 1}) tasks [extract_person(text) for text in chunk] chunk_results await asyncio.gather(*tasks) all_persons.extend(chunk_results) end_time time.time() duration end_time - start_time print(fChunked processing took: {duration:.2f} seconds) return all_persons, duration8.5 完整基准测试编排examples/asyncio-benchmarks/run.py中的benchmark_all_methods()会把上述所有方法统一注册进一个methods列表依次运行并在结尾打印汇总表与相对串行的加速倍数speedup。它是现成的性能验证工具设置好OPENAI_API_KEY后直接运行即可对比各策略的实际耗时便于针对自己的数据规模确定并发度与是否限流。九、与错误处理、重试机制的配合并发只是第一步生产环境还要求对失败进行兜底。结合仓库文档建议如下配合区分重试层次SDK 层max_retries负责传输层重试Instructor 的client.create(max_retries...)负责抽取与校验失败后的重试二者是不同预算不要相乘估算实际调用次数见 重试与 Tenacity 指南在并发中接入 Tenacity可使用AsyncRetrying配合retry_if_exception_type((RateLimitError, APIConnectionError, ValidationError))、wait_exponential指数退避把重试策略直接传给client.create(max_retriesRetrying(...))捕获结构化异常Instructor 的异常层次都以InstructorError为基类常用InstructorRetryException携带n_attempts、failed_attempts、total_usage与ValidationError在并发循环内逐条捕获即可实现优雅降级详见 错误处理指南。十、关键结论asyncio.gather是追求有序结果时的最快路径但需注意其异常传播语义asyncio.as_completed更适合流式消费与超大数据集可显著降低首条结果延迟信号量限流是生产应用的必需品可把并发度压到 API 允许的范围错误处理必不可少return_exceptionsTrue、asyncio.wait_for、Instructor 异常层次与 Tenacity 重试应组合使用用基准测试验证而非猜测examples/asyncio-benchmarks/run.py提供了现成的对比框架助你针对真实数据规模确定最优策略。延伸阅读异步客户端配置from_provider错误处理模式重试与 Tenacity 限流策略并行工具调用Parallel Tools完整示例learn-async/run.py基准测试asyncio-benchmarks/run.py【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 基于 Instructor 与 asyncio 的 LLM 并发处理实战：asyncio.gather 与 asyncio.as_completed
基于 Instructor 与 asyncio 的 LLM 并发处理实战asyncio.gather 与 asyncio.as_completed【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor导读本指南围绕 Instructor 仓库中docs/blog/posts/learn-async.md一文展开系统讲解如何利用 Pythonasyncio的asyncio.gather与asyncio.as_completed对 LLM 结构化抽取任务进行高效并发处理。文中将给出完整可运行的异步客户端配置、四种核心处理模式顺序基线、并发收集、流式完成、信号量限流的代码与性能对比并结合仓库中的examples/learn-async/run.py与examples/asyncio-benchmarks/run.py源码补充错误处理、超时控制、进度跟踪、分批处理等生产级进阶模式。读完后你将掌握在大批量 LLM 抽取场景下正确选择并发策略、实施限流与监控的方法。一、理解 asyncio.gather 与 asyncio.as_completedPython 的asyncio库为并发执行提供了两种强大的手段它们的核心差异决定了适用场景asyncio.gather并发执行所有任务并按输入顺序返回结果。适合需要保持结果顺序、且要求全部任务成功完成的场景。asyncio.as_completed按完成先后返回结果不保证输入顺序。适合希望尽快开始处理已完成任务、面向大数据集或流式输出的场景。两种方式都远优于串行顺序处理。需要特别注意的是asyncio.gather会在某个任务抛出异常时立即向上传播并中断整体等待而asyncio.as_completed要求你在循环内逐个await task并自行处理可能的异常——这一点在后续“生产级进阶模式”中会结合源码进一步展开。二、完整环境搭建创建 Instructor 异步客户端Instructor 支持同步与异步两种客户端。现代推荐写法是使用统一的from_provider接口通过async_clientTrue直接获得异步客户端参见 from_provider 指南import instructor from pydantic import BaseModel # 创建异步客户端 client instructor.from_provider(openai/gpt-4o-mini, async_clientTrue) class Person(BaseModel): name: str age: int occupation: str async def extract_person(text: str) - Person: 使用 LLM 从文本中抽取人物信息。 return await client.create( modelgpt-4o-mini, response_modelPerson, messages[{role: user, content: fExtract person info: {text}}], )async_client参数是from_provider的核心开关。从源码结构看instructor/v2/auto_client.py中的各个 provider 工厂函数约第 27–101 行都接收async_client: bool并在其内部根据该布尔值选择返回同步 SDK 客户端还是Async客户端最终统一由client.create(...)提供一致的接口。仓库中的历史示例 examples/learn-async/run.py 采用的是更早期的apatch风格instructor.apatch(AsyncOpenAI())调用点为client.chat.completions.create(...)import time import asyncio import instructor from pydantic import BaseModel from openai import AsyncOpenAI client instructor.apatch(AsyncOpenAI()) class Person(BaseModel): name: str age: int async def extract_person(text: str) - Person: return await client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: text}], response_modelPerson, )两种写法底层都通过 patch 机制为 SDK 客户端注入了response_model结构化抽取能力async_clientTrue是当前推荐方式而apatch仍是向后兼容的可用路径。在进入并发模式之前先准备一份待处理的样本数据集下文所有示例共用dataset [ John Smith is a 30-year-old software engineer, Sarah Johnson is a 25-year-old data scientist, Mike Davis is a 35-year-old product manager, Lisa Wilson is a 28-year-old UX designer, Tom Brown is a 32-year-old DevOps engineer, Emma Garcia is a 27-year-old frontend developer, David Lee is a 33-year-old backend developer, ]三、方法一顺序处理基线顺序处理是最简单的写法逐个await每次等待前一次完成后再发起下一次请求。它是衡量并发收益的基线。async def sequential_processing() - List[Person]: 逐条处理——最慢的方式。 start_time time.time() persons [] for text in dataset: person await extract_person(text) persons.append(person) print(fProcessed: {person.name}) end_time time.time() print(fSequential processing took: {end_time - start_time:.2f} seconds) return persons # 运行方式persons await sequential_processing()在示例输出中examples/learn-async/run.py 的运行注释7 条数据顺序处理耗时约6.17 秒。每次 HTTP 往返的延迟被完全串行叠加当数据量增大时耗时线性增长。四、方法二asyncio.gather 并发处理asyncio.gather把所有协程包装成任务后一次性并发调度await等待全部完成后返回结果列表结果顺序与输入顺序严格一致async def gather_processing() - List[Person]: 并发处理全部条目按输入顺序返回。 start_time time.time() # 为所有条目创建任务 tasks [extract_person(text) for text in dataset] # 并发执行所有任务 persons await asyncio.gather(*tasks) end_time time.time() print(fasyncio.gather took: {end_time - start_time:.2f} seconds) # 结果保持原始顺序 for person in persons: print(fProcessed: {person.name}) return persons # 运行方式persons await gather_processing()要点解析tasks [extract_person(text) for text in dataset]只是创建协程对象列表真正的调度发生在asyncio.gather(*tasks)通过*解包一次性传入所有任务时返回的persons列表下标与dataset一一对应方便按序落库或对齐下游处理由于所有请求几乎同时发出总耗时约等于单次请求的最长耗时示例中约0.85 秒而非各请求耗时之和。五、方法三asyncio.as_completed 流式处理asyncio.as_completed接受一个可等待对象列表返回一个按完成顺序产出任务的迭代器。每完成一个请求就可以立刻await并处理其结果async def as_completed_processing() - List[Person]: 并发处理条目按完成顺序消费结果。 start_time time.time() persons [] # 为所有条目创建任务 tasks [extract_person(text) for text in dataset] # 按完成顺序处理结果 for task in asyncio.as_completed(tasks): person await task persons.append(person) print(fCompleted: {person.name}) end_time time.time() print(fasyncio.as_completed took: {end_time - start_time:.2f} seconds) return persons # 运行方式persons await as_completed_processing()与gather相比as_completed的核心价值在于边完成边消费在示例输出中它耗时约0.95 秒但在这段时间内每个结果一旦就绪就会被立即打印和追加无需等待最慢的那个请求。对于结果需要逐条写入数据库、推送消息队列或渲染进度的应用这能显著降低端到端的首条结果延迟。六、方法四基于信号量的限流处理并发度拉满虽然快但会给上游 API 带来压力也容易触发速率限制rate limit。asyncio.Semaphore可以在保持并发框架的同时把同时进行的请求数量约束在指定上限async def rate_limited_extract_person( text: str, semaphore: asyncio.Semaphore ) - Person: 带限流的人物信息抽取。 async with semaphore: return await extract_person(text) async def rate_limited_gather(concurrency_limit: int 3) - List[Person]: 使用 asyncio.gather 配合受控并发度处理。 start_time time.time() # 创建信号量限制并发请求数 semaphore asyncio.Semaphore(concurrency_limit) # 创建带限流的任务 tasks [rate_limited_extract_person(text, semaphore) for text in dataset] # 在限流约束下执行 persons await asyncio.gather(*tasks) end_time time.time() print( fRate-limited gather (limit{concurrency_limit}) took: {end_time - start_time:.2f} seconds ) return persons async def rate_limited_as_completed(concurrency_limit: int 3) - List[Person]: 使用 asyncio.as_completed 配合受控并发度处理。 start_time time.time() persons [] # 创建信号量限制并发请求数 semaphore asyncio.Semaphore(concurrency_limit) # 创建带限流的任务 tasks [rate_limited_extract_person(text, semaphore) for text in dataset] # 按完成顺序处理结果 for task in asyncio.as_completed(tasks): person await task persons.append(person) print(fRate-limited completed: {person.name}) end_time time.time() print( fRate-limited as_completed (limit{concurrency_limit}) took: {end_time - start_time:.2f} seconds ) return persons # 运行方式 # persons await rate_limited_gather(concurrency_limit2) # persons await rate_limited_as_completed(concurrency_limit2)信号量的工作方式async with semaphore:进入时若已并发请求数达到上限例如 2后续协程会在此处挂起等待直到某个正在执行的任务退出信号量上下文。示例输出显示concurrency_limit2时两种限流方法耗时约3.04 秒 / 3.26 秒——比全并发慢但远快于串行且对 API 更友好。七、性能对比与选型指南以下为 7 条数据在示例运行中观察到的典型耗时来自 examples/learn-async/run.py 的注释输出实际数值会随模型、网络和 API 状态波动仅作量级参考方法执行时间并发度适用场景Sequential顺序6.17 秒1基线对照asyncio.gather0.85 秒7追求速度、需要有序结果asyncio.as_completed0.95 秒7流式消费结果Rate-limited gather限流 gather3.04 秒2对 API 友好Rate-limited as_completed限流流式3.26 秒2流式 限流何时使用 asyncio.gather结果必须与输入保持相同顺序要求所有任务都成功完成任一失败即整体失败可接受追求最快的整体执行时间内存占用不是首要考量需要同时持有全部结果。何时使用 asyncio.as_completed希望结果一到达就立刻处理流式写入、逐条上报结果顺序无关紧要面向超大数据集希望尽早开始消费以控制内存与等待时间需要实时的进度反馈。何时必须限流面对 API 速率限制如429错误需要对外部服务保持礼貌、避免封禁控制资源消耗与成本构建生产级应用时几乎总是需要配合重试策略。八、生产级进阶模式仓库中 examples/asyncio-benchmarks/run.py 在前述四种基础模式之上进一步提供了面向生产环境的五种进阶模式下面逐一说明。8.1 错误容错gather 的 return_exceptionsasyncio.gather默认任一任务抛异常会立刻中断整体。传入return_exceptionsTrue后异常会被当作普通返回值收进结果列表由你在循环内逐项甄别async def robust_gather_processing() - tuple[list[Person], float]: 带错误处理并发抽取。 start_time time.time() tasks [extract_person(text) for text in dataset] # 异常不再直接中断而是作为返回值 results await asyncio.gather(*tasks, return_exceptionsTrue) persons [] for i, result in enumerate(results): if isinstance(result, Exception): print(fError processing item {i}: {result}) else: persons.append(result) end_time time.time() print(fRobust gather processing took: {end_time - start_time:.2f} seconds) return persons, end_time - start_time这样单条抽取失败不会拖垮整批任务失败项可以被单独记录或后续重试。8.2 整体超时asyncio.wait_for给整批并发操作设置硬性截止时间超时后取消任务async def timeout_gather_processing( timeout_seconds: float 30.0, ) - tuple[list[Person], float]: 带超时的并发处理。 start_time time.time() tasks [extract_person(text) for text in dataset] try: persons await asyncio.wait_for( asyncio.gather(*tasks), timeouttimeout_seconds ) end_time time.time() print(fTimeout gather processing took: {end_time - start_time:.2f} seconds) return persons, end_time - start_time except asyncio.TimeoutError: end_time time.time() print( fProcessing timed out after {timeout_seconds} seconds (took {end_time - start_time:.2f}s) ) return [], end_time - start_time关于超时与重试的更完整语义如timeout同时转发给 SDK 并作为验证重试的 elapsed 停止条件、asyncio.wait_for只保证取消等待而不保证远端停止计算可参考仓库中的 重试与 Tenacity 指南。8.3 进度跟踪配合as_completed可以精确统计已完成条数输出实时百分比async def progress_tracking_processing() - tuple[list[Person], float]: 带进度跟踪的并发处理。 start_time time.time() persons [] total_items len(dataset) completed 0 tasks [extract_person(text) for text in dataset] for task in asyncio.as_completed(tasks): person await task persons.append(person) completed 1 print(fProgress: {completed}/{total_items} ({completed / total_items * 100:.1f}%)) end_time time.time() print(fProgress tracking processing took: {end_time - start_time:.2f} seconds) return persons, end_time - start_time这正是as_completed相对gather的差异化优势场景gather要等全部完成才能拿到任何结果而as_completed天然支持增量反馈。8.4 分批处理Chunking当数据集极大、内存或限流要求严格时可以把数据切成若干 chunk每批内部用gather并发批与批之间串行兼顾并发度与资源控制async def chunked_processing(chunk_size: int 3) - tuple[list[Person], float]: 分批并发处理控制内存与速率限制。 start_time time.time() all_persons [] for i in range(0, len(dataset), chunk_size): chunk dataset[i : i chunk_size] print(fProcessing chunk {i // chunk_size 1}) tasks [extract_person(text) for text in chunk] chunk_results await asyncio.gather(*tasks) all_persons.extend(chunk_results) end_time time.time() duration end_time - start_time print(fChunked processing took: {duration:.2f} seconds) return all_persons, duration8.5 完整基准测试编排examples/asyncio-benchmarks/run.py中的benchmark_all_methods()会把上述所有方法统一注册进一个methods列表依次运行并在结尾打印汇总表与相对串行的加速倍数speedup。它是现成的性能验证工具设置好OPENAI_API_KEY后直接运行即可对比各策略的实际耗时便于针对自己的数据规模确定并发度与是否限流。九、与错误处理、重试机制的配合并发只是第一步生产环境还要求对失败进行兜底。结合仓库文档建议如下配合区分重试层次SDK 层max_retries负责传输层重试Instructor 的client.create(max_retries...)负责抽取与校验失败后的重试二者是不同预算不要相乘估算实际调用次数见 重试与 Tenacity 指南在并发中接入 Tenacity可使用AsyncRetrying配合retry_if_exception_type((RateLimitError, APIConnectionError, ValidationError))、wait_exponential指数退避把重试策略直接传给client.create(max_retriesRetrying(...))捕获结构化异常Instructor 的异常层次都以InstructorError为基类常用InstructorRetryException携带n_attempts、failed_attempts、total_usage与ValidationError在并发循环内逐条捕获即可实现优雅降级详见 错误处理指南。十、关键结论asyncio.gather是追求有序结果时的最快路径但需注意其异常传播语义asyncio.as_completed更适合流式消费与超大数据集可显著降低首条结果延迟信号量限流是生产应用的必需品可把并发度压到 API 允许的范围错误处理必不可少return_exceptionsTrue、asyncio.wait_for、Instructor 异常层次与 Tenacity 重试应组合使用用基准测试验证而非猜测examples/asyncio-benchmarks/run.py提供了现成的对比框架助你针对真实数据规模确定最优策略。延伸阅读异步客户端配置from_provider错误处理模式重试与 Tenacity 限流策略并行工具调用Parallel Tools完整示例learn-async/run.py基准测试asyncio-benchmarks/run.py【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考