メインコンテンツへスキップ
見出し画像

The Complete Agentic AI Engineering Course (2025): Async Python と OpenAI Agents SDK (第2/6週・1/5日目)

    • 非同期Pythonの核心(async/await・イベントループ・asyncio.gather)でI/O待ちを効率化し、典型的な落とし穴(トップレベルでawait不可・awaitableは使い切り・旧式yield fromは非推奨)を押さえる。

    • OpenAI Agents SDK(旧Swarm)の最小概念(Agent/Handoffs/Guardrails)と「Agent作成→trace(...)で記録→await Runner.run(...)で実行」の三手順を、トレース活用やagents命名衝突への注意とともに実演。

    • “バイブコーディング”の5習慣(最新API指定・複数LLMでの検証・小さく分割し独立テスト・別モデルでのレビュー・3案比較)を示し、以降の学習(CrewAI→LangGraph→AutoGen→MCP)への橋渡しを明確化。


    コース最終回のゆっくり丁寧なウォークスルー:いまどきのエージェント実行をゼロから体得する

    ここまでの旅の締めくくりに、非同期(async)Pythonを腹落ちさせてから、OpenAI Agents SDKで最小構成の “作る→記録する→走らせる” をやってみます。
    この SDK は軽量・柔軟で、Week 6 の MCP 回でも再登場。だから今ここで基礎を固めておくと、後半の伸びが気持ちよくなります。

    進行はゆっくり、噛みくだいて、つまずきやすい所で必ず減速します。途中に生存テクも挟むので、エラーと仲良くなる必要はありません。


    なぜ「非同期(async)Python」が“Agents Land”の心臓なのか

    このコースで触れた OpenAI Agents SDK / CrewAI / LangGraph / AutoGen / MCP、さらには “ノーフレームワーク” 構成に至るまで、舞台裏では非同期 Python(asyncio)が脈打っています。理由はシンプルで、LLM アプリは I/O 待ちがほとんどだから。モデル呼び出し、ウェブ検索、各種ツール API──みんな“待ち”です。重たいスレッドや別プロセスを増やさずに待ち時間を上手に切り替えるのが async の役目です。

    まずは「ショートバージョン」(暗記してOKな最小ルール)

    • 併走させたい処理は async def で定義する。

    • それを呼ぶときは 必ず await で待つ。

    async def do_some_processing() -> str:
        # ここでは I/O 的な処理をしたと想定
        return "done!"
    
    result = await do_some_processing()
    

    「async def で作って await で走らせる」──このワンセットだけで、最初の壁は越えられます。

    つづいて「ほんとうの話」(ここが腑に落ちると一気に楽)

    • asyncio はスレッド/プロセスの軽量代替。 OS スレッドではなく イベントループが“ひとつのコルーチンを順番に回す”。コルーチンが I/O で待機した瞬間、別のコルーチンへスイッチ。

    • async def で定義したものは「コルーチン」。
      呼んでもすぐ実行されない。まずは“コルーチンオブジェクト”が返るだけ。

    • await で実行がスケジュールされ、イベントループ上で進む。I/O 待ちで止まれば、他のコルーチンにバトンが渡る。

    脳内イメージ:イベントループは有能な交通整理員。青信号(CPU が空いてる)なら車(コルーチン)を進め、赤になって待ち(I/O)に入ったら次の車を出す──それだけ。

    今週すぐ使う並行パターン(これだけで十分戦えます)

    # 1) 複数のコルーチンをまとめて投げ、ぜんぶ揃うのを待つ
    results = await asyncio.gather(
        fetch_customer(),
        fetch_product_catalog(),
        fetch_competitors()
    )
    
    # 2) 先にタスク化しておいて、あとで回収する
    task = asyncio.create_task(fetch_customer())  # Task は await 可能
    # ...ここで他の処理をしても良い...
    customer = await task
    

    つまずきやすいポイント(今のうちに回避)

    • await は async def の内側でしか使えない。 スクリプト直書きなら asyncio.run(main()) で包む。

    • awaitable は “一回こっきり”。 完了したものをもう一度 await できない(イテレータを使い切るのと同じ)。

    • モダンスタイルは async/await。 旧式の “ジェネレータベースのコルーチン”(@asyncio.coroutine や yield from)は非推奨。今から書く新規コードでは混ぜない。

    • async は I/O 向け。 CPU が詰む処理はスレッド/プロセスか、外部サービスに逃がす。


    OpenAI Agents SDK を紹介(旧称 “Swarm”)

    このツールキットの良さは 軽量・柔軟・出しゃばらないところ。日常的に面倒な“ツールの JSON 配線”などはいい感じに肩代わりしてくれます。思想の押しつけが弱く、設計の自由度が高いのも特徴。MCP とも相性がよく、Week 6 で再登場します。

    覚える用語は最小限:

    • Agents: 役割(指示)とツールを持つ LLM の表現

    • Handoffs: エージェント間のやりとり

    • Guardrails: 入力・出力の制御(逸脱防止の柵)

    モデル固定ではありません。引数に素のモデル名を渡すと既定で OpenAI 扱いですが、モデルは切り替え可能。SDK 自体は“非・押し付け”です。


    はじめての実行:Create → Trace → Run(そして await) の三手詰め

    以下は最小の “こんにちは” 例。ひっかかりがちな 2 箇所に注意書きを入れておきます。

    # 0) インポートと環境変数の読み込み
    import asyncio
    from dotenv import load_dotenv
    from agents import Agent, Runner, trace  # SDK のパッケージ名は 'agents'
    
    load_dotenv(override=True)
    
    # 1) エージェント定義:役割(instructions)とモデル
    agent = Agent(
        name="Jokester",
        instructions="あなたはジョークを語るアシスタントです。",
        model="gpt-4o-mini"  # 素の名前なら既定は OpenAI
    )
    
    # 2) 実行を trace で包んで、モニタリングに流す
    async def main():
        with trace("ジョークを語る"):
            # 3) エージェントを走らせる — Runner.run(...) は「非同期」。必ず await。
            result = await Runner.run(agent, "自律型AIエージェントに関するジョークを1つ。短く。")
            print(result.final_output)
    
    asyncio.run(main())
    

    ありがちな落とし穴(誰もが 1 回は見るやつ)

    result = Runner.run(agent, "ジョークを…")
    print(result.final_output)
    # AttributeError: 'coroutine' object has no attribute 'final_output'
    

    これは「await を忘れてるよ」という Python からの優しい叱責。async 関数の中で await Runner.run(...) しましょう。

    なぜ trace(...) で包むのか?
    ブロック内のやりとりが一つのタイムラインとしてモニタリングに記録されるから。単発の LLM 呼び出しでは地味でも、複数エージェント+ツールが絡み出した瞬間に、差が“可視化力”として出ます。

    名前衝突の注意:SDK のパッケージ名が agents と汎用的。自作 agents.py があるとインポートが影になります。モジュール名の整理か、絶対インポートで回避を。


    今日アップデートされるメンタルモデル(3 点)

    1. Async はエージェント工学の必修科目。
      イベントループは常に 1 本のコルーチンを進めるけれど、I/O 待ちで素早く切り替わるから“体感的には並行”。

      • 合言葉:async def で作り、await で走らせる。

      • 複数待ちは asyncio.gather(...)。

    2. OpenAI Agents SDK は「軽いのに助かる」。

      • Agent → Handoffs → Guardrails の素直な思考枠。

      • Create → Trace → Run(await) で最初の 1 歩が出しやすい。

      • ツール配線や JSON の雑務は肩代わり、出荷速度が上がる。

    3. Tracing は装飾ではなくシートベルト。

      • trace(...) の習慣化で、マルチエージェントやツール連携が時間順に可視化。原因追跡も説明責任も楽になる。


    「バイブコーディング」生存キット(速さと正確さの両立)

    新しいフレームワークは “LLM に下書きさせて手直し” が速い。けれど規律を混ぜないと泥濘にハマります。以下の 5 習慣で滑走路を平らに:

    1. Good vibes:再利用するプロンプトひな型を用意。

      • 簡潔なコードを要求。

      • “本日時点の最新 API” を使うと明記(学習由来の旧 API に戻りがちだから)。

    2. Vibe but verify:2 つの LLM に同じ質問を投げる(例:ChatGPT と Claude)。

      • 答えの差分が学び。片方の過剰さ・見落としに気づける。

    3. Step up the vibe:コード生成の前に、問題を小さな独立テスト可能ステップに分解させる。

      • 1 ステップ 10 行前後で実装→テスト。

      • 200 行の“合体物”ではなく、10 行 × 20 個の“合格パーツ”を組む。

    4. Vibe and validate:モデル A が書いたら、モデル B にレビューを依頼。

      • バグ・冗長・構造のまずさを他の目で摘出。

      • 手動の Evaluator–Optimizer パターンは効く。

    5. Vibe with variety:同じ小課題で実装を 3 通り出してもらい、トレードオフで選ぶ。

      • なぜその形にしたか短い根拠も添えてもらうと理解が進む。

    まとめ:薄切りで進める&各薄切りをテストする。 これだけで“LLM 産 200 行キメラ”問題は 8 割消えます。


    この先のやさしい予告(全体ロードマップの文脈づけ)

    今日は async を腹で理解し、OpenAI Agents SDK で初の実行を体験しました。Week 2 の後半では SDR(インサイドセールス)エージェント、会話・ガードレール、Deep Research ミニアプリへ。
    そこから CrewAI(Week 3)→ LangGraph(Week 4)→ AutoGen(Week 5) と上り詰め、Week 6 で Agents SDK × MCP に戻って複数のローカル/リモート MCP サーバを束ね、AI Equity Traders まで踏み込みます。味は違えど、鼓動は同じ async。


    机の横に貼るチェックリスト

    • I/O なら async def & await。

    • asyncio.gather(...) で多待ちを束ねる。

    • 旧式 yield from 系と async/await を混在させない。

    • Agent を作る → trace(...) で包む → await Runner.run(...)。

    • バイブは検証・査読・多様化で薄切り実装。


    最後に:コピペしてすぐ走る最小スクリプト

    # run_agent.py
    import asyncio
    from dotenv import load_dotenv
    from agents import Agent, Runner, trace
    
    load_dotenv(override=True)
    
    agent = Agent(
        name="Sidekick",
        instructions="あなたは開発者の頼れる相棒。回答は簡潔に。",
        model="gpt-4o-mini"
    )
    
    async def main():
        with trace("初めてのエージェント実行"):
            msg = "なぜ asyncio が I/O バウンドなエージェントアプリに向いているのか、要点だけまとめて。"
            res = await Runner.run(agent, msg)
            print("\n=== Assistant ===\n", res.final_output)
    
    if __name__ == "__main__":
        asyncio.run(main())
    

    一度走らせて、モニタリング画面のトレースを覗いてみてください。リクエストとレスポンスが時系列で一望できます。ここからツールを足し、エージェントを増やし、やがてこの数行が本格オーケストレーションの出発点になります。


    ここまで到達できたあなたは、async の思考回路を持ち、数行でエージェントを立ち上げる腕を持ち、厳しめのバイブ運用術まで身につけました。
    残るのは反復と好奇心。この二つは良いエンジニアの燃料で、在庫切れになりません。次に進むたび、今日の三手詰め(Create → Trace → Run)を合図に始めれば、迷いません。


    (補足:読みやすさのためのプチ解説)

    • awaitable(アウェイタブル):await できるものの総称。ネイティブコルーチン、asyncio.Task / Future、__await__ を実装したカスタム型など。

    • CPU バウンドの扱い:モデル応答整形や長い前処理など CPU を食うものは、asyncio.to_thread(...) や concurrent.futures、もしくは外部ワーカーに逃すと健全。

    • パッケージ名の衝突:自作の agents.py と SDK の agents は同名。衝突したらモジュールのリネームや名前空間の明示で回避。


    「超温和なパイソン」へ

    あなたへのおすすめ