エージェントへのカスタムストリーム出力の追加

Overview

カスタムストリーム出力を構成して、プレーンテキストに加えて、ストリーミング呼び出し用に構造化されたマシンが判読可能な出力を発行するようにエージェントに指示できます。 Atlas Agent Engine は、この出力をカスタムイベントとしてクライアントに送信します。

このガイドでは、次のタスクを実行する方法について説明しています。

カスタムストリーム出力を有効にする: features.use_custom_parser: trueagent.yamlファイルのフラグを設定します。
出力パーサーを登録する:アプリにOutputParser サブクラスを登録します。
カスタムイベントを発行:エージェントコードからカスタムイベントを発行します。

前提条件

開始する前に、agent.yaml ファイルを使用して登録されたエージェントがあることを確認してください。詳細については、「 Atlas Agent Engine の使用」を参照してください。

カスタムストリーム出力の有効化

エージェントのカスタムストリーム出力を有効にするには、次のコードを agent.yamlファイルに追加します。

features:
  use_custom_parser: true

features.use_custom_parserフラグをtrue に設定しているが出力パーサーを登録していない場合、呼び出しが開始されたときにエージェントは失敗し、RuntimeError が返されます。

出力パーサーの登録

@app.output_parser 修飾子を使用してアプリに OutputParser サブクラスを登録します。 Atlas Agent Engine はコレクション時にクラスを検証するため、モジュールのインポート時に OutputParser サブクラスではないクラスを渡すと失敗します。

LgGraph エージェントの場合は、 LangGraphOutputParserクラスをサブクラス化します。カスタムイベントに変換する LingGraph ストリームモードに stream_modesクラス属性を設定します。次に、parse() メソッドと on_stream_error() メソッドを実装します。

次の例では、LingGraphエージェントの出力パーサーを登録します。

from agent_engine_sdk_langgraph.output_parser import LangGraphOutputParser
@app.output_parser
class BriefParser(LangGraphOutputParser):
    stream_modes = ("messages", "values")
    async def parse(self, item, ctx):
        ...
        yield {"event": "brief", "data": new_brief}
    async def on_stream_error(self, ctx, error):
        ...
        return {"event": "error"}

次の表では、LangGraphOutputParser メンバーを説明しています。

ノード	説明
`stream_modes`	LgGraph ストリームモード名のシーケンス（`("messages", "values")` など）。 Atlas Agent Engine は、リスト化された各モードの正規化されたストリームアイテムを `parse()` メソッドに渡します。
`parse(item, ctx)`	1 つのストリームアイテムに対して発行するカスタムイベントを決定します。 JSONシリアル化可能なオブジェクトを生成する非同期メソッドとして `parse()` を実装します。項目をスキップするには、値を生成せずにを返します。
`on_stream_error(ctx, error)`	プラットフォームがストリームエラーを表示する前に発行する任意のJSONシリアル化可能なカスタムイベントを返します。

parse() または on_stream_error() メソッドでエラーが発生した場合、出力が自動的に削除されるのではなく、ストリームは失敗します。

カスタムイベントの発行

ストリーミング呼び出し中に、emit_custom_event() 関数を使用して、LingGraph ノード、ツール、または実行内で実行されるその他のコードからカスタムイベントを直接出力できます。次の例に示すように、string キーを含むJSONオブジェクトを emit_custom_event() に渡します。

from runner_shared import emit_custom_event
await emit_custom_event({"event": "todo", "items": ["a", "b"]})

Atlas Agent Engine はJSONオブジェクトを変更せずに転送し、スキーマを適用しないため、クライアントが期待するものに合わせてペイロードの構造を定義できます。

emit_custom_event() 関数を呼び出すと、次のルールが適用されます。

ストリーミング呼び出しからのみメソッドを呼び出します。同期 invoke パスから呼び出すとエラーが発生します。
agent.yamlファイルで features.use_custom_parser フラグを有効にします。フラグが無効になっている場合、プラットフォームはエラーを発生させます。
非同期コードからのメソッドを待機します。同期コードから、代わりに emit_custom_event_sync() を使用してください。

次のステップ

ストリーム出力をクライアントがどのように受信するかについては、エージェントの呼び出しガイドの「エージェントの出力のストリーム」を参照してください。

戻る

リモート MCP サーバー

エージェントへのカスタム ストリーム出力の追加