同期 vs 非同期

一行サマリ

run_sync / run / run_stream の 3 つの API は同じ Agent に対する 3 つの呼び出し方 で、用途と返ってくる型が違う。

ヒーロー: 3 つの API の関係

PydanticAI の Agent はどの API を使っても 同じ Agent オブジェクト・同じ instructions・同じツール を共有します。違うのは「結果の受け取り方」だけです。

概念: なぜ 3 つも API があるのか

LLM 呼び出しは ネットワーク I/O で、応答が返ってくるまで数秒かかります。この「待ち時間」をどう扱うかで API が分かれます。

• run_sync — 結果が返るまでスレッドをブロックする。Colab セルや CLI スクリプトのように 「1 件ずつ順番に動かせばよい」 場面で最も簡単
• run — await で待つ。同じプロセスで 複数の Agent 呼び出しを並列に走らせたい 場合や、FastAPI / Starlette のような async な Web フレームワーク から呼ぶときに使う
• run_stream — async for で 部分的な応答を逐次受け取る。Web UI で「タイピング中」を表現したり、長い応答を体感速度よく見せたりするのに使う

混乱しがちなのは「run_sync だけが同期で、他の 2 つは非同期」「run_stream は run の上位互換ではなく、用途が違う」の 2 点です。

コード: 3 つを並べて比較

run_sync — もっとも簡単

from pydantic_ai import Agent
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    instructions='日本語で簡潔に答えてください。',
)
 
# Cell 全体がブロックされ、応答が返るまで次のセルに進めない
result = agent.run_sync('PydanticAIを一言で')
print(result.output)

ポイント:

• 戻り値は AgentRunResult。.output で文字列を取り出す
• 内部では asyncio.run(agent.run(...)) 相当の処理を行っている
• Colab のように すでに動いているイベントループ がある環境では run_sync は使える (内部で nest_asyncio 相当の処理がある)

run — async から呼ぶ

import asyncio
from pydantic_ai import Agent
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    instructions='日本語で簡潔に答えてください。',
)
 
async def main():
    # 単発呼び出し
    result = await agent.run('PydanticAIを一言で')
    print(result.output)
 
    # 複数を並列に走らせる (これが run の本領)
    queries = ['Agentとは?', 'Toolとは?', 'output_typeとは?']
    results = await asyncio.gather(*(agent.run(q) for q in queries))
    for r in results:
        print('-', r.output[:60], '...')
 
asyncio.run(main())

ポイント:

• 戻り値は同じく AgentRunResult。run_sync との違いは 呼び出し側が await するか否か だけ
• asyncio.gather で N 個のクエリを並列に投げる と、合計時間がほぼ最長 1 件分になる
• FastAPI のエンドポイントは関数を async def で書けば、その中で await agent.run(...) をそのまま使える

run_stream — 部分応答を逐次

import asyncio
from pydantic_ai import Agent
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    instructions='日本語で簡潔に答えてください。',
)
 
async def main():
    async with agent.run_stream('PydanticAIの強みを3つ教えて') as response:
        async for chunk in response.stream_text():
            # chunk はそれまでに得られたテキスト全体 (デフォルト)
            # 末尾の差分だけ欲しい場合は delta=True を渡す
            print(chunk, end='\r', flush=True)
        print()  # 改行
        # ループを抜けたら最終結果が確定している
        final = await response.get_output()
        print('FINAL:', final)
 
asyncio.run(main())

ポイント:

• async with で StreamedRunResult を開く必要がある
• stream_text() が 部分テキストの async iterator を返す
• 入力途中のテキストを画面に流したいだけなら delta=True で末尾差分だけ受け取るとシンプル
• 最終応答全体は get_output() で確定させる

いつ何を選ぶか — 早見表

まとめ

• run_sync / run / run_stream は 同じ Agent への 3 つの呼び出し方。中身は同じ
• Colab・CLI なら run_sync、async Web フレームワークなら run、逐次表示なら run_stream
• 並列に投げるときは asyncio.gather + run。これが体感速度を一番伸ばす

次レッスンでは、これまで使ってきた 'google-gla:' プレフィックス (Google AI Studio) と、企業向けの 'google-vertex:' プレフィックス (Vertex AI) の違いと使い分けを扱います。