跳到内容

重试

使用 Pydantic 的好处之一是我们可以轻松定义验证器。我们在许多文章中介绍了这个主题，例如“重新验证”以及我们的博客文章“好的 LLM 验证就是好的验证”。

本文主要介绍如何使用简单和更复杂的重试机制及逻辑。

验证器示例

在开始之前，我们将使用一个简单的验证器示例。一个检查姓名是否为大写字母的验证器。虽然我们可以明确提示要求姓名使用大写字母，但这展示了如何在不改变提示的情况下构建额外的逻辑。

from typing import Annotated
from pydantic import AfterValidator, BaseModel


def uppercase_validator(v):
    if v.islower():
        raise ValueError("Name must be ALL CAPS")
    return v


class UserDetail(BaseModel):
    name: Annotated[str, AfterValidator(uppercase_validator)]
    age: int


try:
    UserDetail(name="jason", age=12)
except Exception as e:
    print(e)
    """
    1 validation error for UserDetail
    name
      Value error, Name must be ALL CAPS [type=value_error, input_value='jason', input_type=str]
        For further information visit https://errors.pydantic.dev/2.9/v/value_error
    """

简单：最大重试次数

设置重试最简单的方法是将一个整数值赋给 max_retries。

import openai
import instructor
from pydantic import BaseModel


class UserDetail(BaseModel):
    name: str
    age: int


client = instructor.from_openai(openai.OpenAI(), mode=instructor.Mode.TOOLS)

response = client.chat.completions.create(
    model="gpt-4-turbo-preview",
    response_model=UserDetail,
    messages=[
        {"role": "user", "content": "Extract `jason is 12`"},
    ],
    max_retries=3,  # (1)!
)
print(response.model_dump_json(indent=2))
"""
{
  "name": "jason",
  "age": 12
}
"""
# (2)!
  1. 我们将最大重试次数设置为 3。这意味着如果模型返回错误，我们将最多重新请求模型 3 次。
  2. 我们断言姓名是全大写字母。

捕获重试异常

如果想捕获重试异常，可以这样做，并访问 last_completion、n_attempts 和 messages 属性。

from pydantic import BaseModel, field_validator
import openai
import instructor
from instructor.exceptions import InstructorRetryException
from tenacity import Retrying, retry_if_not_exception_type, stop_after_attempt

# Patch the OpenAI client to enable response_model
client = instructor.from_openai(openai.OpenAI())


# Define a Pydantic model for the user details
class UserDetail(BaseModel):
    name: str
    age: int

    @field_validator("age")
    def validate_age(cls, v: int):
        raise ValueError(f"You will never succeed with {str(v)}")


retries = Retrying(
    retry=retry_if_not_exception_type(ZeroDivisionError), stop=stop_after_attempt(3)
)
# Use the client to create a user detail
try:
    user = client.chat.completions.create(
        model="gpt-3.5-turbo",
        response_model=UserDetail,
        messages=[{"role": "user", "content": "Extract Jason is 25 years old"}],
        max_retries=retries,
    )
except InstructorRetryException as e:
    print(e.messages[-1]["content"])  # type: ignore
    """
    Validation Error found:
    1 validation error for UserDetail
    age
      Value error, You will never succeed with 25 [type=value_error, input_value=25, input_type=int]
        For further information visit https://errors.pydantic.dev/2.9/v/value_error
    Recall the function correctly, fix the errors
    """

    print(e.n_attempts)
    #> 3

    print(e.last_completion)
    """
    ChatCompletion(
        id='chatcmpl-B7YgHmfrWA8FxsSxvzUUvdSe2lo9h',
        choices=[
            Choice(
                finish_reason='stop',
                index=0,
                logprobs=None,
                message=ChatCompletionMessage(
                    content=None,
                    refusal=None,
                    role='assistant',
                    audio=None,
                    function_call=None,
                    tool_calls=[
                        ChatCompletionMessageToolCall(
                            id='call_zvTyhnBKPIhDXrOCxNzlsgzN',
                            function=Function(
                                arguments='{"name":"Jason","age":25}', name='UserDetail'
                            ),
                            type='function',
                        )
                    ],
                ),
            )
        ],
        created=1741141309,
        model='gpt-3.5-turbo-0125',
        object='chat.completion',
        service_tier='default',
        system_fingerprint=None,
        usage=CompletionUsage(
            completion_tokens=30,
            prompt_tokens=522,
            total_tokens=552,
            completion_tokens_details=CompletionTokensDetails(
                audio_tokens=0, reasoning_tokens=0
            ),
            prompt_tokens_details=PromptTokensDetails(audio_tokens=0, cached_tokens=0),
        ),
    )
    """

高级：重试逻辑

如果您想更精细地控制如何定义重试，例如退避策略和额外的重试逻辑，可以使用一个名为 Tenacity 的库。要了解更多信息，请查看 Tenacity 网站上的文档。

除了使用 @retry 装饰器之外，我们可以使用 Retrying 和 AsyncRetrying 类来定义我们自己的重试逻辑。

import openai
import instructor
from pydantic import BaseModel
from tenacity import Retrying, stop_after_attempt, wait_fixed

client = instructor.from_openai(openai.OpenAI(), mode=instructor.Mode.TOOLS)


class UserDetail(BaseModel):
    name: str
    age: int


response = client.chat.completions.create(
    model="gpt-4-turbo-preview",
    response_model=UserDetail,
    messages=[
        {"role": "user", "content": "Extract `jason is 12`"},
    ],
    max_retries=Retrying(
        stop=stop_after_attempt(2),  # (1)!
        wait=wait_fixed(1),  # (2)!
    ),  # (3)!
)
print(response.model_dump_json(indent=2))
"""
{
  "name": "jason",
  "age": 12
}
"""
  1. 我们在尝试 2 次后停止
  2. 我们在每次尝试之间等待 1 秒
  3. 现在我们可以定义自己的重试逻辑

异步重试

如果您正在使用异步代码，可以使用 AsyncRetrying 代替。

import openai
import instructor
from pydantic import BaseModel
from tenacity import AsyncRetrying, stop_after_attempt, wait_fixed

client = instructor.from_openai(openai.AsyncOpenAI(), mode=instructor.Mode.TOOLS)


class UserDetail(BaseModel):
    name: str
    age: int


task = client.chat.completions.create(
    model="gpt-4-turbo-preview",
    response_model=UserDetail,
    messages=[
        {"role": "user", "content": "Extract `jason is 12`"},
    ],
    max_retries=AsyncRetrying(
        stop=stop_after_attempt(2),
        wait=wait_fixed(1),
    ),
)

import asyncio

response = asyncio.run(task)
print(response.model_dump_json(indent=2))
"""
{
  "name": "jason",
  "age": 12
}
"""

Tenacity 的其他特性

Tenacity 提供了大量不同的重试功能。下面列出了其中一些功能。

  • Retrying(stop=stop_after_attempt(2)): 在尝试 2 次后停止
  • Retrying(stop=stop_after_delay(10)): 在 10 秒后停止
  • Retrying(wait=wait_fixed(1)): 在每次尝试之间等待 1 秒
  • Retrying(wait=wait_random(0, 1)): 在 0 到 1 秒之间随机等待一段时间
  • Retrying(wait=wait_exponential(multiplier=1, min=4, max=10)): 以指数级增长的方式在 4 到 10 秒之间等待
  • Retrying(wait=(stop_after_attempt(2) | stop_after_delay(10))): 在尝试 2 次或 10 秒后停止
  • Retrying(wait=(wait_fixed(1) + wait_random(0.2))): 至少等待 1 秒，并额外增加最多 0.2 秒的随机等待

请记住，对于异步客户端，您需要使用 AsyncRetrying 而不是 Retrying！

重试回调

您还可以定义在每次尝试之前和之后调用的回调函数。这对于日志记录或调试非常有用。

from pydantic import BaseModel, field_validator
import instructor
import tenacity
import openai

client = instructor.from_openai(openai.OpenAI())


class User(BaseModel):
    name: str
    age: int

    @field_validator("name")
    def name_is_uppercase(cls, v: str):
        assert v.isupper(), "Name must be uppercase"
        return v


resp = client.messages.create(
    model="gpt-3.5-turbo",
    max_tokens=1024,
    max_retries=tenacity.Retrying(
        stop=tenacity.stop_after_attempt(3),
        before=lambda _: print("before:", _),
"""
before:
<RetryCallState 13531729680: attempt #1; slept for 0.0; last result: none yet>
"""
        after=lambda _: print("after:", _),
    ),  # type: ignore
    messages=[
        {
            "role": "user",
            "content": "Extract John is 18 years old.",
        }
    ],
    response_model=User,
)

assert isinstance(resp, User)
assert resp.name == "JOHN"  # due to validation
assert resp.age == 18
print(resp)
#> name='JOHN' age=18
# Output: name='JOHN' age=18

# Sample output:
# before: <RetryCallState 4421908816: attempt #1; slept for 0.0; last result: none yet>
# after: <RetryCallState 4421908816: attempt #1; slept for 0.0; last result: failed (ValidationError 1 validation error for User
# name
#   Assertion failed, Name must be uppercase [type=assertion_error, input_value='John', input_type=str]
#     For further information visit https://errors.pydantic.dev/2.6/v/assertion_error)>
#
# before: <RetryCallState 4421908816: attempt #2; slept for 0.0; last result: none yet>
# name='JOHN' age=18