DI入門

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
 
@dataclass
class CityDb:
    cities: dict[str, str]   # 国名 -> 首都名 の辞書
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    deps_type=CityDb,
    instructions='ユーザーの国名から首都を答えてください。',
)
 
@agent.instructions
def known_cities(ctx: RunContext[CityDb]) -> str:
    countries = ', '.join(ctx.deps.cities.keys())
    return f'参照可能な国は: {countries}'
 
deps = CityDb(cities={'日本': '東京', 'フランス': 'パリ', 'ブラジル': 'ブラジリア'})
print(agent.run_sync('フランスの首都は?', deps=deps).output)

import os
from dataclasses import dataclass, field
import httpx
from pydantic_ai import Agent, RunContext
 
@dataclass
class AppConfig:
    api_base_url: str
    user_lang: str
 
@dataclass
class AppDeps:
    config: AppConfig
    http: httpx.AsyncClient
    cache: dict[str, str] = field(default_factory=dict)
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    deps_type=AppDeps,
    instructions='ユーザーの問いに答えてください。',
)
 
@agent.instructions
def lang(ctx: RunContext[AppDeps]) -> str:
    return f'回答は {ctx.deps.config.user_lang} で。'
 
# 起動時に 1 度だけ重い初期化
async def make_deps() -> AppDeps:
    return AppDeps(
        config=AppConfig(api_base_url=os.environ['API_BASE'], user_lang='ja'),
        http=httpx.AsyncClient(timeout=10.0),
    )
 
# 利用側
deps = await make_deps()
print((await agent.run('Tokyo の現在気温は?', deps=deps)).output)

from pydantic_ai import Agent, RunContext, ModelRetry
 
@dataclass
class Deps:
    blocked_words: set[str]
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    deps_type=Deps,
    instructions='短い日本語で答えてください。',
)
 
@agent.output_validator
def no_blocked(ctx: RunContext[Deps], output: str) -> str:
    for word in ctx.deps.blocked_words:
        if word in output:
            raise ModelRetry(f'「{word}」を含めずに言い換えてください。')
    return output
 
deps = Deps(blocked_words={'秘密', 'パスワード'})
print(agent.run_sync('社内システムのログイン手順を 1 行で', deps=deps).output)

import asyncio
 
async def main():
    async with httpx.AsyncClient() as http:
        deps = AppDeps(config=AppConfig('https://api.example.com', 'ja'), http=http)
        # 何度でも run を回せる
        for q in ['Tokyo', 'Paris', 'Brasilia']:
            r = await agent.run(f'{q} の現在気温は?', deps=deps)
            print(r.output)
    # async with を抜けたら http が自動で close される
 
asyncio.run(main())