# OpenAI Agent SDKの便利だけど知られていない機能 🌍 ネストしたエージェントツールの実行中、進捗をリアルタイムにストリーミングしたいと思いませんか？ `on_stream` コールバックを使えば、子エージェントの実行イベントを親側でリアルタイムに受け取れます。 📌 タイトル：ネストしたエージェント実行のストリーミング 🔗 URL： 🧩 概要 エージェントを `as_tool()` でツール化する際、`on_stream` パラメータにコールバック関数を設定できます。このコールバックは `AgentToolStreamEvent` を受け取り、子エージェントの実行中に発生するイベント（`raw_response_event`、`run_item_stream_event` など）をリアルタイムで処理できます。イベントの種類は通常のストリーミングと同じ形式なので、既存のストリーミングハンドラを再利用できます。 🛠 使い方 ```python from agents import Agent, Runner research_agent = Agent( name="researcher", instructions="トピックについて詳細に調査してください", ) # ストリーミングコールバック async def handle_stream(event): # raw_response_event: モデルのレスポンス # run_item_stream_event: ツール呼び出し等 if hasattr(event, 'data'): print(f"[研究中] {") parent = Agent( name="coordinator", tools=[ research_agent.as_tool( tool_name="research", tool_description="トピックを調査する", on_stream=handle_stream, ), ], ) # ストリーミング実行 async for event in "AIの最新動向を調査して"): print(event) ``` 🏗 本番システムへの組み込み方 ・子エージェントの処理進捗をUIにリアルタイム表示し、ユーザーに待機感を与えない ・`raw_response_event` で生成テキストを逐次表示する ・`run_item_stream_event` でツール実行状況をログに記録する ・既存のストリーミングUIコンポーネントをそのまま再利用できる 💡 ユースケース 🖥 マルチエージェントの処理進捗をリアルタイムUI表示 📝 子エージェントの応答をタイプライター風に逐次表示 🔍 調査エージェントの検索・分析プロセスの可視化 📊 長時間処理の途中経過をユーザーにフィードバック ⚠️ 注意点 `on_stream` コールバックは子エージェントの実行スレッド内で呼ばれるため、重い処理を行うとエージェント全体のパフォーマンスに影響します。イベントハンドラは軽量に保ち、必要に応じて非同期キューに渡すなどの工夫をしてください。また、すべてのイベント型に対応できるよう、未知のイベントを無視するガード処理を入れておくと安全です。 ✨ ストリーミングで、マルチエージェントの「中身が見える」体験をユーザーに届けましょう。 #OpenAIAgentSDK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 AIエージェントのワークフローを「グラフ」として設計できたら、複雑な処理フローも見通しよく管理できると思いませんか？ ADK 2.0のWorkflowクラスは、ノードとエッジでエージェントの実行パスを定義するグラフベースのワークフロー構築機能です。AIエージェント、カスタム関数、ツール、さらにはネストされたWorkflowをノードとして組み合わせ、明示的なルーティングロジックで制御できます。 📌 タイトル：Workflow クラスによるグラフベース実行 🔗 URL： 🧩 概要 Workflowクラスは、ADK 2.0におけるグラフベースのエージェントワークフローを構築するための主要な構成要素です。ノードには、AIエージェント・コード関数・ツール・ネストされたWorkflowを配置でき、エッジ（edges）パラメータで実行パスを定義します。逐次実行は `("START", node1, node2, node3)` のようにタプルで記述し、条件分岐は辞書で定義します。これにより、従来のプロンプトベースの制御では難しかった複雑な分岐構造を、明示的かつ信頼性高く実装できます。 🛠 使い方 Workflowクラスのインスタンスを作成し、nodesにノードのリスト、edgesに実行パスを指定します。 ```python from adk import Workflow, Agent agent_a = Agent(name="researcher", ...) agent_b = Agent(name="writer", ...) workflow = Workflow( name="content_pipeline", nodes=[agent_a, agent_b], edges=[("START", agent_a, agent_b)] ) ``` 逐次実行の場合はタプルでノードを列挙し、条件分岐が必要な場合は辞書ベースのルーティングを使います。ネストされたWorkflowをノードとして組み込むことで、大規模なパイプラインも階層的に管理できます。 🏗 本番システムへの組み込み方 ・各ノードの責務を明確に分離し、テスト可能な単位で設計する ・エラーハンドリングを各ノードレベルで実装し、障害の影響範囲を限定する ・ネストされたWorkflowを活用して、再利用可能なサブパイプラインを構築する ・条件分岐のルーティングロジックをドキュメント化し、チームでの保守性を確保する 💡 ユースケース 📄 論文の収集→要約→レビュー→投稿を一連のグラフとして管理 🔀 入力データの種類に応じて異なる処理パイプラインに分岐 🏢 複数部門のエージェントを組み合わせた承認フローの構築 🔁 サブワークフローを再利用した複数プロジェクト横断の自動化 ⚠️ 注意点 WorkflowクラスはLive Streamingとの互換性がなく、一部のサードパーティ統合にも対応していません。リアルタイムのストリーミング応答が必要なケースでは、別のアプローチを検討する必要があります。また、グラフの複雑さが増すとデバッグが難しくなるため、適切な粒度でノードを分割することが重要です。 ✨ 明示的なグラフ定義により、エージェントワークフローの信頼性と保守性が大きく向上します。複雑な処理フローを構築する際にはぜひ活用してみてください。 #ADK# #AIAgent#

🗒️ ElevenLabs MORNING RESEARCH✍️ 番組専属のAIエージェント #ラボちゃん# とともに、 気になるニュースや身近なギモンをリサーチ！👀 今朝は、 皆さんに「レコード」についてお届けしました！ ElevenLabsはこちら👇 📻 #ohayomorning# #jwave#

🗒️ ElevenLabs MORNING RESEARCH✍️ 番組専属のAIエージェント #ラボちゃん# とともに、 気になるニュースや身近なギモンをリサーチ！ 今朝は、 皆さんの口腔ケア🦷についてお届けしました！ ElevenLabsはこちら👇 📻 #ohayomorning# #jwave# #radiko#