Hand-off

import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
 
model = GoogleModel('gemini-3-flash-preview')
 
summarizer = Agent(
    model,
    instructions='与えられた英語の文章を 3 文以内の英語で要約してください。',
)
 
translator = Agent(
    model,
    instructions='与えられた英文を自然な日本語に翻訳してください。',
)
 
async def pipeline(en_text: str) -> str:
    # Hand-off 1: 要約
    summary_en = (await summarizer.run(en_text)).output
    # Hand-off 2: 翻訳 — summarizer の出力を translator の入力に
    summary_ja = (await translator.run(summary_en)).output
    return summary_ja
 
ja = asyncio.run(pipeline(
    'Pydantic AI is a Python agent framework that focuses on type safety and developer experience...'
))
print(ja)

from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
 
model = GoogleModel('gemini-3-flash-preview')
 
class RouteDecision(BaseModel):
    category: Literal['billing', 'technical', 'cancel', 'other']
    reason_ja: str = Field(description='判断理由 (30 字以内)')
 
router = Agent(
    model,
    output_type=RouteDecision,
    instructions='ユーザー問い合わせを billing / technical / cancel / other に分類してください。',
)
 
billing_agent = Agent(model, instructions='請求関連の問い合わせに丁寧に答える担当です。')
tech_agent = Agent(model, instructions='技術トラブルの一次対応をする担当です。')
cancel_agent = Agent(model, instructions='解約手続きを案内する担当です。')
fallback_agent = Agent(model, instructions='その他の問い合わせを丁寧に受け止める担当です。')
 
async def handle(user_msg: str) -> str:
    decision = (await router.run(user_msg)).output
    target = {
        'billing': billing_agent,
        'technical': tech_agent,
        'cancel': cancel_agent,
    }.get(decision.category, fallback_agent)
    final = (await target.run(user_msg)).output
    return f'[{decision.category}] {final}'

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
 
model = GoogleModel('gemini-3-flash-preview')
 
class ExtractedInvoice(BaseModel):
    vendor: str = Field(description='請求元')
    total_yen: int = Field(description='合計金額 (円)')
    items: list[str] = Field(description='品目リスト')
 
extractor = Agent(
    model,
    output_type=ExtractedInvoice,
    instructions='請求書テキストから vendor / total_yen / items を抽出してください。',
)
 
approver = Agent(
    model,
    instructions='構造化された請求書情報を読み、承認可否と理由を 2 文で答えてください。',
)
 
async def review(invoice_text: str) -> str:
    inv = (await extractor.run(invoice_text)).output  # ExtractedInvoice
    prompt = (
        f'次の請求書を承認するかどうか判定してください:\\n'
        f'- 請求元: {inv.vendor}\\n'
        f'- 金額: {inv.total_yen} 円\\n'
        f'- 品目: {", ".join(inv.items)}'
    )
    return (await approver.run(prompt)).output

r1 = await router.run(msg)
r2 = await target.run(msg)
total_in = r1.usage().input_tokens + r2.usage().input_tokens
total_out = r1.usage().output_tokens + r2.usage().output_tokens