ModelRetry

from pydantic_ai import Agent, ModelRetry
 
USERS = {'tanaka': 1, 'sato': 2, 'suzuki': 3}
 
agent = Agent('google-gla:gemini-3-flash-preview', instructions='短く日本語で。')
 
@agent.tool_plain
def get_user_id(name: str) -> int:
    """ユーザー名から ID を取得します。name は社内システム上の英字ユーザー名。"""
    name_lower = name.lower()
    if name_lower not in USERS:
        raise ModelRetry(
            f'ユーザー {name!r} は存在しません。'
            f'利用可能なユーザー: {", ".join(USERS.keys())}'
        )
    return USERS[name_lower]
 
# LLM は最初 'Yamada' で呼ぶかもしれない → ModelRetry の理由を読んで存在するユーザーで再試行
print(agent.run_sync('山田さんの ID を教えて。なければ田中さんの ID で。').output)

agent_calc = Agent('google-gla:gemini-3-flash-preview', instructions='答えは整数で。')
 
@agent_calc.tool_plain
def divide(a: float, b: float) -> float:
    """a を b で割った商を返します。"""
    if b == 0:
        raise ModelRetry('0 で割ろうとしました。b に 0 以外の値を指定してください。')
    return a / b
 
print(agent_calc.run_sync('10 を 0 で割って').output)
# LLM は「不可能なので」と返すか、引数を 1 等に変えて再試行する

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext, ModelRetry
 
@dataclass
class WeatherDeps:
    http: httpx.AsyncClient
    api_base: str
 
agent_w = Agent(
    'google-gla:gemini-3-flash-preview',
    deps_type=WeatherDeps,
    instructions='短く日本語で。',
)
 
@agent_w.tool
async def get_weather(ctx: RunContext[WeatherDeps], city: str) -> str:
    """指定都市の天気を取得します。city は ISO 3166-1 / 主要都市の英字表記。"""
    response = await ctx.deps.http.get(f'{ctx.deps.api_base}/weather?city={city}')
    if response.status_code == 404:
        raise ModelRetry(
            f'{city!r} は API に登録されていません。'
            'Tokyo / Paris / New York のような英字の主要都市名で再試行してください。'
        )
    response.raise_for_status()
    return response.json()['summary']

@agent.tool_plain(retries=3)
def stricter_tool(x: str) -> str:
    if not x.startswith('OK-'):
        raise ModelRetry('値は "OK-" で始める必要があります')
    return x