Pydantic Evals 重试策略完全指南为任务与评估器配置 Tenacity 自动重试【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai导读基于 LLM 的评测系统在运行过程中常常会遇到限流Rate Limit、网络超时、临时 API 故障以及上下文长度超限等瞬态故障一次失败的评测并不代表任务本身有问题。Pydantic Evals 提供了一套基于 Tenacity 的重试机制可通过evaluate()/evaluate_sync()的retry_task与retry_evaluators参数分别为被评测的任务函数和评估器配置独立的自动重试逻辑。读完本文你将掌握重试配置的完整参数语义、任务重试与评估器重试的触发时机、指数退避的计算方式、失败后的EvaluatorFailure上报以及限流、超时、上下文超长三类实战场景下的最佳配置方案。为什么需要重试LLM 系统的瞬态故障LLM 系统在评测evaluation过程中面临的故障大致可以分为两类瞬态故障transient failures——限流rate limits、网络超时network timeouts、临时 API 宕机temporary API outages这类故障在稍后重试即可恢复确定性故障——如上下文长度超限context length errors、参数校验失败这类故障单纯重试无法解决需要调整调用方式或在代码内处理。Pydantic Evals 的重试机制只针对前者设计。在 Pydantic Evals 中重试配置覆盖两个独立环节任务执行Task execution——被评测的函数本身task评估器执行Evaluator execution——对任务输出打分的评估器如LLMJudge。两者各自独立配置、互不影响因为 LLM 评测链路中任务调用大模型和评估器调用大模型都可能分别触发限流。安装前提Tenacity 依赖重试能力基于 Tenacity 实现而 Tenacity 是 Pydantic AI 的可选依赖。从 pydantic_ai_slim/pyproject.toml 可以看到它挂在retries可选组下retries [tenacity8.2.3, httpx0.27]因此使用重试功能前需要先安装该可选组Pydantic Evals 依赖pydantic-ai-slim见 pydantic_evals/pyproject.tomlpip install pydantic-ai-slim[retries]如果未安装 Tenacity 就传入retry_task/retry_evaluators源码会在运行时从pydantic_ai.retries导入retry时抛出带有明确提示的ImportError见 dataset.py 中的注释说明。基础重试配置最简上手在Dataset上调用evaluate()异步或evaluate_sync()同步包装通过关键字参数传入 Tenacity 参数即可启用重试。下面的例子为任务配置了最多 3 次尝试为评估器配置了最多 2 次尝试from tenacity import stop_after_attempt from pydantic_evals import Case, Dataset def my_function(inputs: str) - str: return fResult: {inputs} dataset Dataset(namebasic_retry, cases[Case(inputstest)], evaluators[]) report dataset.evaluate_sync( taskmy_function, retry_task{stop: stop_after_attempt(3)}, retry_evaluators{stop: stop_after_attempt(2)}, )这两个参数在源码中的签名类型均为RetryConfig | None见 dataset.py 中evaluate()的完整定义evaluate_sync是evaluate的同步包装参数完全一致见 dataset.py#L417-L434。RetryConfig与 Pydantic AI 完全一致的配置模型retry_task/retry_evaluators接受的是pydantic_ai.retries.RetryConfig——一个TypedDict其字段与 Tenacityretry装饰器的全部参数一一对应定义在 pydantic_ai_slim/pydantic_ai/retries.py#L72-L134。也就是说Tenacity 装饰器上能写的所有参数都能写进这个配置字典from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset def my_function(inputs: str) - str: return fResult: {inputs} dataset Dataset(nameretry_config, cases[Case(inputstest)], evaluators[]) retry_config { stop: stop_after_attempt(3), # Stop after 3 attempts wait: wait_exponential(multiplier1, min1, max10), # Exponential backoff: 1s, 2s, 4s, 8s (capped at 10s) reraise: True, # Re-raise the original exception after exhausting retries } dataset.evaluate_sync( taskmy_function, retry_taskretry_config, )常用参数速查表RetryConfig的所有字段均为可选未提供时使用 Tenacity 装饰器的默认值。常用参数如下参数类型说明Tenacity 默认值stopStopBaseT停止策略如stop_after_attempt(3)、stop_after_delay(60)stop_never永不停止waitWaitBaseT等待策略如wait_exponential()、wait_fixed(2)wait_none不等待retryRetryBaseT重试条件如retry_if_exception_type(TimeoutError)retry_if_exception_type()不重试任何异常reraisebool重试耗尽后是否重新抛出原始异常为False时抛出RetryErrorFalsebefore_sleepCallable每次重试休眠前的回调常用于打日志Nonebefore/afterCallable每次尝试前后回调before_nothing/after_nothingsleepCallable休眠实现可注入自定义 sleeptenacity.nap.sleepretry_error_clstype[RetryError]重试耗尽且reraiseFalse时抛出的异常类tenacity.RetryErrorretry_error_callbackCallable重试耗尽且reraiseFalse时的回调None一个值得注意的默认值陷阱Tenacity 的stop默认是stop_never即如果不显式指定stop理论上会无限重试。因此在实际使用中stop几乎是必填项。同时Tenacity 默认的retry条件是不重试任何异常若不指定retry条件则所有异常都会触发重试因为retry_if_exception_type()匹配任意异常的子类判断在未传入类型时对所有异常返回 True这意味着需要借助retry_if_exception_type(TimeoutError)之类的条件精确控制哪些异常才值得重试。更多完整选项可查阅 Tenacity 官方文档。任务重试为被评测函数配置重试任务函数是评测链路的第一步。当任务函数抛出异常时重试机制会按照配置重新执行任务。下面是一个模拟可能触发限流或超时的 LLM 任务配置示例异步任务from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset async def call_llm(inputs: str) - str: return fLLM response to: {inputs} async def flaky_llm_task(inputs: str) - str: This might hit rate limits or timeout. response await call_llm(inputs) return response dataset Dataset(nametask_retry, cases[Case(inputstest)]) report dataset.evaluate_sync( taskflaky_llm_task, retry_task{ stop: stop_after_attempt(5), # Try up to 5 times wait: wait_exponential(multiplier1, min1, max30), # Exponential backoff, capped at 30s reraise: True, }, )任务重试的触发时机任务重试在任务抛出异常时触发。在任务函数内部只要异常未被捕获吞掉而是向上传播无论是直接raise还是raise重新抛出就会进入重试流程class RateLimitError(Exception): pass class ValidationError(Exception): pass async def call_api(inputs: str) - str: return fAPI response: {inputs} async def my_task(inputs: str) - str: try: return await call_api(inputs) except RateLimitError: # Will trigger retry raise except ValidationError: # Will also trigger retry raise底层实现任务是如何被重试的从源码看任务重试的实现非常直接在 dataset.py#L965-L995 的_run_task()中先定义单次执行的内部函数_run_once()其中用anyio.to_thread.run_sync执行同步任务、直接await异步任务并包裹logfire_span记录耗时随后if retry: from pydantic_ai.retries import retry as tenacity_retry _run_once tenacity_retry(**retry)(_run_once)即用 Tenacity 的retry装饰器原地包装单次执行函数装饰器参数直接来自retry_task字典。这意味着任务重试发生在一次任务执行粒度上——每次尝试都会重新调用任务函数并记录独立的 span。指数退避的计算方式wait_exponential()的延迟按指数增长具体公式为Attempt 1: immediate Attempt 2: ~1s delay (multiplier * 2^0) Attempt 3: ~2s delay (multiplier * 2^1) Attempt 4: ~4s delay (multiplier * 2^2) Attempt 5: ~8s delay (multiplier * 2^3, capped at max)实际延迟取决于传给wait_exponential()的三个参数multiplier基础乘数控制每次退避的步长min最小等待时间秒max最大等待时间秒防止退避时间无限膨胀。例如wait_exponential(multiplier2, min2, max60)的序列约为 2s、4s、8s、16s、32s、60s封顶。评估器重试为打分环节配置重试评估器尤其是LLMJudge这类内部也要调用 LLM 的评估器同样可能触发限流。评估器重试通过retry_evaluators独立配置from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import LLMJudge def my_task(inputs: str) - str: return fResult: {inputs} dataset Dataset( nameevaluator_retry, cases[Case(inputstest)], evaluators[ # LLMJudge might hit rate limits LLMJudge(rubricResponse is accurate), ], ) report dataset.evaluate_sync( taskmy_task, retry_evaluators{ stop: stop_after_attempt(3), wait: wait_exponential(multiplier1, min0.5, max10), reraise: True, }, )注意retry_evaluators对数据集级评估器Dataset(evaluators[...])与用例级评估器Case(evaluators[...])统一生效。从 dataset.py#L1152-L1163 可以看到评测时会把case.evaluators dataset_evaluators合并再对每个评估器并行调用run_evaluator(ev, scoring_context, retry_evaluators)。评估器重试的触发时机评估器重试在评估器抛出异常时触发。自定义评估器只要在evaluate方法中抛出异常就会被重试逻辑捕获from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext async def external_api_call(output: str) - bool: return len(output) 0 dataclass class APIEvaluator(Evaluator): async def evaluate(self, ctx: EvaluatorContext) - bool: # If this raises an exception, retry logic will trigger result await external_api_call(ctx.output) return result底层实现评估器是如何被重试的评估器重试的实现与任务重试对称位于 pydantic_evals/pydantic_evals/evaluators/_run_evaluator.py#L35-L107 的run_evaluator()中evaluate evaluator.evaluate_async if retry is not None: from pydantic_ai.retries import retry as tenacity_retry evaluate tenacity_retry(**retry)(evaluate)重试包装的是evaluate_async方法本身随后在try/except中捕获所有异常并转为EvaluatorFailure返回。也就是说评估器重试耗尽后异常不会向上抛出而是被转换为EvaluatorFailure记录到报告中评测流程继续执行其他用例。评估器失败的记录与查看如果评估器在所有重试尝试后仍然失败该失败会被记录为EvaluatorFailure定义见 pydantic_evals/pydantic_evals/evaluators/evaluator.py#L103-L118其中包含name评估器名称、error_message异常类型: 异常信息、error_stacktrace完整堆栈、source评估器序列化 spec以及error_type异常类名同时作为error.type属性出现在 OTel 事件上。通过report.cases[i].evaluator_failures可以逐条检查from tenacity import stop_after_attempt from pydantic_evals import Case, Dataset def task(inputs: str) - str: return fResult: {inputs} dataset Dataset(nameevaluator_failures, cases[Case(inputstest)], evaluators[]) report dataset.evaluate_sync(task, retry_evaluators{stop: stop_after_attempt(3)}) # Check for evaluator failures for case in report.cases: if case.evaluator_failures: for failure in case.evaluator_failures: print(fEvaluator {failure.name} failed: {failure.error_message}) # (No output - no evaluator failures in this case)也可以在打印报告时直接通过include_evaluator_failuresTrue让失败信息一并展示from pydantic_evals import Case, Dataset def task(inputs: str) - str: return fResult: {inputs} dataset Dataset(namefailure_report, cases[Case(inputstest)], evaluators[]) report dataset.evaluate_sync(task) report.print(include_evaluator_failuresTrue) Evaluation Summary: task ┏━━━━━━━━━━┳━━━━━━━━━━┓ ┃ Case ID ┃ Duration ┃ ┡━━━━━━━━━━╇━━━━━━━━━━┩ │ Case 1 │ 10ms │ ├──────────┼──────────┤ │ Averages │ 10ms │ └──────────┴──────────┘ # # ✅ case_0 ━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% (0/0)组合配置任务与评估器独立重试任务和评估器是两条独立的重试链路可以同时配置、互不干扰。典型做法是任务链路允许更多尝试次数任务失败代价高评估器链路适度重试from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset def flaky_task(inputs: str) - str: return fResult: {inputs} dataset Dataset(namecombined_retry, cases[Case(inputstest)], evaluators[]) report dataset.evaluate_sync( taskflaky_task, retry_task{ stop: stop_after_attempt(5), # Retry task up to 5 times wait: wait_exponential(multiplier1, min1, max30), reraise: True, }, retry_evaluators{ stop: stop_after_attempt(3), # Retry evaluators up to 3 times wait: wait_exponential(multiplier1, min0.5, max10), reraise: True, }, )实战场景一限流处理Rate Limit Handling限流是 LLM 评测中最常见的故障。限流后 API 服务端通常会要求等待一段时间因此限流场景适合慷慨的尝试次数 较长且增长较快的指数退避from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import LLMJudge async def expensive_llm_call(inputs: str) - str: return fLLM response: {inputs} async def llm_task(inputs: str) - str: Task that might hit rate limits. return await expensive_llm_call(inputs) dataset Dataset( namerate_limit_retry, cases[Case(inputstest)], evaluators[ LLMJudge(rubricQuality check), # Also might hit rate limits ], ) # Generous retries for rate limits report dataset.evaluate_sync( taskllm_task, retry_task{ stop: stop_after_attempt(10), # Rate limits can take multiple retries wait: wait_exponential(multiplier2, min2, max60), # Start at 2s, exponential up to 60s reraise: True, }, retry_evaluators{ stop: stop_after_attempt(5), wait: wait_exponential(multiplier2, min2, max30), reraise: True, }, )如果服务端返回了Retry-After响应头Pydantic AI 还提供了wait_retry_after等待策略见 pydantic_ai_slim/pydantic_ai/retries.py#L514-L591它会优先读取响应头中的等待时长支持整数秒与 HTTP 日期两种格式无响应头时回退到指数退避可配合max_wait参数封顶最大等待时间——这是处理限流时比纯指数退避更精确的选择。实战场景二网络超时处理Network Timeout Handling网络超时通常是短时抖动适合少量快速重试避免拖慢整个评测流程import httpx from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset async def api_task(inputs: str) - str: Task that calls external API which might timeout. async with httpx.AsyncClient(timeout10.0) as client: response await client.post(https://api.example.com, json{input: inputs}) return response.text dataset Dataset(nametimeout_retry, cases[Case(inputstest)], evaluators[]) # Quick retries for network issues report dataset.evaluate_sync( taskapi_task, retry_task{ stop: stop_after_attempt(4), # A few quick retries wait: wait_exponential(multiplier0.5, min0.5, max5), # Fast retry, capped at 5s reraise: True, }, )这里用multiplier0.5、min0.5让首次退避仅有 0.5 秒快速重试网络抖动max5防止异常情况下等待时间失控。实战场景三上下文长度处理Context Length Handling上下文长度超限context length errors属于确定性错误单纯重试无法解决。正确的做法是在任务内部自行处理捕获ContextLengthError后截断输入、降低max_tokens重新调用同时在重试配置中把stop设为stop_after_attempt(1)禁用重试避免无意义的重复调用from tenacity import stop_after_attempt from pydantic_evals import Case, Dataset class ContextLengthError(Exception): pass async def llm_call(inputs: str, max_tokens: int 8000) - str: return fLLM response: {inputs[:100]} async def smart_llm_task(inputs: str) - str: Task that might exceed context length. try: return await llm_call(inputs, max_tokens8000) except ContextLengthError: # Retry with shorter context truncated_inputs inputs[:4000] return await llm_call(truncated_inputs, max_tokens4000) dataset Dataset(namecontext_length, cases[Case(inputstest)], evaluators[]) # Dont retry context length errors (handle in task) report dataset.evaluate_sync( tasksmart_llm_task, retry_task{stop: stop_after_attempt(1)}, # No retries, we handle it )重试 vs 错误处理边界划分并不是所有异常都值得重试正确的分层策略是适合交给重试机制处理瞬态故障限流、超时网络问题临时服务宕机可恢复错误recoverable errors适合在代码中用错误处理解决参数校验错误validation errors逻辑错误logic errors永久性失败permanent failures预期内会发生的错误条件expected error conditions推荐的统一写法是在任务内部处理确定性错误把瞬态错误原样抛出交给重试机制class RateLimitError(Exception): pass async def llm_call(inputs: str) - str: return fLLM response: {inputs} def is_valid(result: str) - bool: return len(result) 0 async def smart_task(inputs: str) - str: Handle expected errors, let retries handle transient failures. try: result await llm_call(inputs) # Validate output (dont retry validation errors) if not is_valid(result): return ERROR: Invalid output format return result except RateLimitError: # Let retry logic handle this raise except ValueError as e: # Dont retry - this is a permanent error return fERROR: {e}关键点在于raise出去的异常交给重试return的错误结果直接进入评估。raise的异常还可以借助retry_if_exception_type(...)进一步限定只有特定异常类型如TimeoutError、RateLimitError才触发重试其余异常立即失败。故障排查TroubleshootingStill failing after retries重试后仍然失败增加尝试次数或检查错误是否真的可重试。调试时可以利用 Tenacity 的日志能力——Tenacity 本身会记录每次重试尝试的日志开启DEBUG级别日志即可看到每次失败的原因import logging from tenacity import stop_after_attempt from pydantic_evals import Case, Dataset def task(inputs: str) - str: return fResult: {inputs} # Add logging to see whats failing logging.basicConfig(levellogging.DEBUG) dataset Dataset(nametroubleshooting, cases[Case(inputstest)], evaluators[]) # Tenacity logs retry attempts report dataset.evaluate_sync(task, retry_task{stop: stop_after_attempt(5)})也可以使用before_sleep回调例如tenacity.before_sleep_log在每次休眠前打印带有重试次数和异常信息的日志定位是哪一类异常在反复失败。Evaluations taking too long评测耗时过长重试会成倍增加总耗时。评估总时长的粗略上界约为任务单次耗时 × 任务尝试次数 评估器单次耗时 × 评估器尝试次数因此缩短耗时的直接手段是减少尝试次数、压缩等待时间from tenacity import stop_after_attempt, wait_exponential # Faster retries retry_config { stop: stop_after_attempt(3), # Fewer attempts wait: wait_exponential(multiplier0.1, min0.1, max2), # Quick retries, capped at 2s reraise: True, }此外评测本身支持max_concurrency参数见 dataset.py#L286默认为None即所有用例并发执行合理设置并发度能显著改善大批量数据集的总吞吐详见 并发与性能指南。Hitting rate limits despite retries重试仍然触发限流这说明重试节奏太激进——重试次数足够但等待时间不够或者并发请求过多。两个方向加大退避时间以及降低并发度max_concurrency源码在 dataset.py#L337 中通过anyio.Semaphore实现max_concurrency2表示同一时刻最多 2 个任务在跑from tenacity import stop_after_attempt, wait_exponential from pydantic_evals import Case, Dataset def task(inputs: str) - str: return fResult: {inputs} dataset Dataset(namerate_limit_config, cases[Case(inputstest)], evaluators[]) # Longer delays retry_config { stop: stop_after_attempt(5), wait: wait_exponential(multiplier5, min5, max60), # Start at 5s, exponential up to 60s reraise: True, } # Also reduce concurrency report dataset.evaluate_sync( tasktask, retry_taskretry_config, max_concurrency2, # Only 2 concurrent tasks )源码验证重试逻辑的测试保障仓库中的测试用例 tests/evals/test_dataset.py#L446-L519 直接验证了任务与评估器的重试行为测试构造了一个前 3 次调用都抛出RuntimeError的异步任务以及一个前 3 次都失败的自定义评估器然后以retry_taskRetryConfig(stopstop_after_attempt(3))、retry_evaluatorsRetryConfig(stopstop_after_attempt(3))运行评测最终断言任务与评估器各被调用了恰好 3 次且报告中的evaluator_failures为空——这印证了两个关键行为重试次数语义stop_after_attempt(3)表示最多尝试 3 次第 3 次成功即结束不会多试重试耗尽后评估器异常会被转换为EvaluatorFailure而非中断整个评测该测试中第 3 次即成功因此无失败记录。小结与延伸阅读Pydantic Evals 的重试机制可以总结为三个要点统一模型任务与评估器共用pydantic_ai.retries.RetryConfig即 Tenacityretry装饰器参数通过retry_task/retry_evaluators独立注入源码层面均以tenacity_retry(**retry)包装执行函数失败隔离任务重试耗尽后异常向外传播成为ReportCaseFailure评估器重试耗尽后转为EvaluatorFailure记录在报告中评测继续策略分层瞬态错误交给重试配指数退避确定性错误在任务内部处理配stop_after_attempt(1)或retry_if_exception_type精确限定。如果评测涉及大规模并发推荐继续阅读 并发与性能指南如果想在 Logfire 中观测每次重试的 span 与失败记录可以参考 Logfire 集成指南。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考