エージェントエグゼキューター

しくみ

C# では、ワークフロー エンジンは、ワークフローに追加される各AIAgentHostExecutorごとに内部的にAIAgentを作成します。 この特殊な Executor は、 ChatProtocolExecutor を拡張し、 ターン トークン パターンを使用します。

1. メッセージ キャッシュ — メッセージが他の Executor から到着すると、エージェント Executor によって収集されます。 ForwardIncomingMessagesが有効になっている場合 (既定値)、受信メッセージもダウンストリーム Executor に転送されます。
2. ターン トークン トリガー — エージェントは、 TurnTokenを受信した後にのみ、キャッシュされたメッセージを処理します。
3. エージェント呼び出し — Executor は、基になるエージェントの RunAsync (非ストリーミング) または RunStreamingAsync (ストリーミング) を呼び出します。
4. 出力の生成 — ストリーミング イベントが有効になっている場合、各増分 AgentResponseUpdate はワークフロー出力として生成されます。 EmitAgentResponseEventsが有効になっている場合、集計されたAgentResponseもワークフロー出力として生成されます。
5. ダウンストリーム メッセージング — エージェントの応答メッセージは、接続されたダウンストリーム Executor に送信されます。
6. ターン トークンパススルー — ターン が完了すると、Executor は新しい TurnToken ダウンストリームを送信して、チェーン内の次のエージェントが処理を開始できるようにします。

暗黙的な作成と明示的な作成

AIAgentにWorkflowBuilderを渡すと、フレームワークによって自動的にAIAgentBindingにラップされ、基になるAIAgentHostExecutorが作成されます。 エージェント Executor を直接インスタンス化する必要はありません。

AIAgent writerAgent = /* create your agent */;
AIAgent reviewerAgent = /* create your agent */;

// Agents are automatically wrapped — no manual executor creation required
var workflow = new WorkflowBuilder(writerAgent)
    .AddEdge(writerAgent, reviewerAgent)
    .Build();

また、一般的なパターンの AgentWorkflowBuilder でヘルパー メソッドを使用することもできます。

// Build a sequential pipeline of agents
var workflow = AgentWorkflowBuilder.BuildSequential(writerAgent, reviewerAgent);

カスタム構成

エージェント Executor の動作をカスタマイズするには、BindAsExecutorでAIAgentHostOptionsを使用します。

var options = new AIAgentHostOptions
{
    EmitAgentUpdateEvents = true,
    EmitAgentResponseEvents = true,
    ReassignOtherAgentsAsUsers = true,
    ForwardIncomingMessages = true,
};

ExecutorBinding writerBinding = writerAgent.BindAsExecutor(options);
var workflow = new WorkflowBuilder(writerBinding)
    .AddEdge(writerBinding, reviewerAgent)
    .Build();

入力の種類

C# のエージェント 実行プログラムは、 string、 ChatMessage、 IEnumerable<ChatMessage>の複数の入力の種類を受け入れます。 文字列入力は、ChatMessage ロールを持つUserインスタンスに自動的に変換されます。 すべての受信メッセージは、 TurnToken が受信されるまで蓄積され、その時点で Executor がバッチを処理します。 ReassignOtherAgentsAsUsersが有効になっている場合 (既定)、他のエージェントからのメッセージはUser ロールに再割り当てされるため、基になるモデルではユーザー入力として扱われますが、現在のエージェントからのメッセージはAssistant ロールを保持します。

出力とチェーン

エージェントのターンが完了すると、Executor は次の手順を実行します。

1. 接続されているすべてのダウンストリーム 実行プログラムにエージェントの応答メッセージを送信します。
2. チェーン内の次のエージェントが処理を開始できるように、新しい TurnToken を転送します。

これにより、エージェントをチェーンできるので、エッジで接続するだけです。

var workflow = new WorkflowBuilder(frenchTranslator)
    .AddEdge(frenchTranslator, spanishTranslator)
    .AddEdge(spanishTranslator, englishTranslator)
    .Build();

ストリーミング動作

ストリーミング動作は、EmitAgentUpdateEventsのAIAgentHostOptions オプションによって制御されるか、TurnTokenを介して動的に制御されます。

• 有効にすると 、Executor はエージェントで RunStreamingAsync を呼び出し、各 AgentResponseUpdate をワークフロー出力イベントとして生成します。 これにより、トークンごとのリアルタイム更新が提供されます。
• 無効にすると 、Executor は RunAsync を呼び出し、1 つの完全な応答を生成します。

// Enable streaming events at the configuration level
var options = new AIAgentHostOptions
{
    EmitAgentUpdateEvents = true,
};

// Or enable streaming dynamically via TurnToken
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

共有セッション

各エージェント実行プログラムは、既定で独自のセッションを維持します。 エージェント間でセッションを共有するには、エージェントをワークフローに追加する前に、共通のセッション プロバイダーでエージェントを構成します。

構成オプション

AIAgentHostOptions は、エージェント実行プログラムの動作を制御します。

チェックポイント処理

エージェント Executor は、実行時間の長いワークフローのチェックポイント処理をサポートします。 チェックポイントが取得されると、Executor は次のコマンドをシリアル化します。

• エージェントのセッション状態 ( SerializeSessionAsync経由)。
• 現在のターンのイベント放出構成 (要求が保留中であり、Executor がまだ受信 TurnTokenを生成していない場合にのみ存在します)。
• 保留中のユーザー入力要求と関数呼び出し要求。

復元時に、Executor はセッションと保留中の要求状態を逆シリアル化し、中断したところからワークフローを再開できるようにします。

しくみ

AgentExecutor クラスは、SupportsAgentRun プロトコルを実装するエージェントをラップします。 Executor がメッセージを受信した場合:

1. メッセージの正規化 — 入力は Message オブジェクトのリストに正規化され、Executor の内部キャッシュに追加されます。 Executor は、キャッシュ前に入力を正規化する専用ハンドラーにルーティングされる複数の入力の種類 ( str、 Message、 list[str | Message]、 AgentExecutorRequest、および AgentExecutorResponse ) を受け入れます。
2. エージェント呼び出し — Executor はキャッシュされたメッセージで agent.run() を呼び出し、ワークフロー実行モードに基づいてストリーミング モードまたは非ストリーミング モードを自動的に選択します。
3. 出力出力 — ストリーミング モードでは、各 AgentResponseUpdate がワークフロー出力イベントとして生成されます。 非ストリーミング モードでは、1 つの AgentResponse が生成されます。
4. ダウンストリーム ディスパッチ — エージェントが完了すると、Executor は接続されているすべてのダウンストリーム Executor に AgentExecutorResponse を送信します。 この応答には、全会話履歴が含まれるので、シームレスな連結ができるようになります。
5. キャッシュのリセット — エージェントの呼び出し後に Executor の内部メッセージ キャッシュがクリアされ、各エージェント呼び出しで最後の呼び出し以降に受信した新しいメッセージのみが処理されるようにします。

暗黙的な作成と明示的な作成

エージェントを直接渡すと、WorkflowBuilderはエージェントをAgentExecutorインスタンスで自動的にラップします。 ほとんどのワークフローでは、暗黙的な作成で十分です。

from agent_framework import WorkflowBuilder

writer_agent = client.as_agent(name="Writer", instructions="...")
reviewer_agent = client.as_agent(name="Reviewer", instructions="...")

# Agents are automatically wrapped — no manual AgentExecutor creation required
workflow = (
    WorkflowBuilder(start_executor=writer_agent)
    .add_edge(writer_agent, reviewer_agent)
    .build()
)

明示的な作成

必要に応じて、 AgentExecutor を明示的に作成します。

• 複数のエージェント間でセッションを共有します。
• ルーティング用のカスタム Executor ID とターゲット ランタイム kwargs を指定します。
• 複数のエッジで同じ Executor インスタンスを参照します。

from agent_framework import AgentExecutor

writer_executor = AgentExecutor(writer_agent, id="my-writer")
reviewer_executor = AgentExecutor(reviewer_agent, id="my-reviewer")

workflow = (
    WorkflowBuilder(start_executor=writer_executor)
    .add_edge(writer_executor, reviewer_executor)
    .build()
)

コンストラクター パラメーター:

入力の種類

AgentExecutorは複数のハンドラー メソッドを定義し、それぞれが異なる入力型を受け入れます。 ワークフロー エンジンは、メッセージの種類に基づいて正しいハンドラーを自動的にディスパッチします。 AgentExecutorRequest フラグがエージェントを実行するか、単にメッセージをキャッシュするかを制御するshould_respondを除き、すべての入力の種類によってエージェントがすぐに実行されます。

AgentExecutorRequest の使用

AgentExecutorRequest は正規の入力型であり、最も多くの制御を提供します。

from agent_framework import AgentExecutorRequest, Message

# Create a request with messages
request = AgentExecutorRequest(
    messages=[Message(role="user", contents=["Hello, world!"])],
    should_respond=True,
)

# Run the workflow
result = await workflow.run(request)

should_respond フラグは、エージェントがメッセージをすぐに処理するか、後でメッセージをキャッシュするかを制御します。

• True (既定値) - エージェントが実行され、応答が生成されます。
• False — メッセージはキャッシュに追加されますが、エージェントは実行されません。 これは、応答をトリガーする前に会話コンテキストを事前に読み込む場合に便利です。

出力とチェーン

エージェントが完了すると、Executor はダウンストリームに AgentExecutorResponse を送信します。 このデータクラスには次のものが含まれます。

エージェント実行プログラムをチェーンする場合、ダウンストリーム Executor は、AgentExecutorResponse ハンドラーを介してfrom_responseを受け取ります。 full_conversation フィールドを使用して完全な会話履歴を保持し、ダウンストリーム エージェントが以前のコンテキストを失うことを防ぎます。

spam_detector = AgentExecutor(create_spam_detector_agent())
email_assistant = AgentExecutor(create_email_assistant_agent())

# The email_assistant receives the spam_detector's full conversation context
workflow = (
    WorkflowBuilder(start_executor=spam_detector)
    .add_edge(spam_detector, email_assistant)
    .build()
)

ストリーミング動作

AgentExecutorは、ワークフロー実行モードに自動的に適応します。

• stream=True — agent.run(stream=True) を呼び出し、各 AgentResponseUpdate をワークフロー出力イベントとして生成します。 ストリーミングが完了すると、更新は集計されて完全な AgentResponse になり、ダウンストリームにディスパッチされます。
• stream=False (既定値) - agent.run(stream=False) を呼び出し、ワークフロー出力イベントとして 1 つの AgentResponse を生成します。

# Streaming mode — receive incremental updates
events = workflow.run("Write a story about a cat.", stream=True)
async for event in events:
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        print(event.data.text, end="", flush=True)

# Non-streaming mode — receive complete response
result = await workflow.run("Write a story about a cat.")

# Retrieve AgentResponse objects from the result
outputs = result.get_outputs()
for output in outputs:
    if isinstance(output, AgentResponse):
        print(output.text)

共有セッション

既定では、各 AgentExecutor は独自のセッションを作成します。 複数のエージェント間でセッションを共有するには (たとえば、共通の会話スレッドを維持するために)、セッションを明示的に作成し、各 Executor に渡します。

from agent_framework import AgentExecutor

# Create a shared session from one agent
shared_session = writer_agent.create_session()

# Both executors share the same session
writer_executor = AgentExecutor(writer_agent, session=shared_session)
reviewer_executor = AgentExecutor(reviewer_agent, session=shared_session)

チェックポイント処理

AgentExecutorでは、実行時間の長いワークフローで状態を保存および復元するためのチェックポイント処理がサポートされています。 チェックポイントが取得されると、Executor は次のコマンドをシリアル化します。

• 内部メッセージ キャッシュ。
• 完全な会話履歴。
• エージェント セッションの状態。
• 保留中のユーザー入力リクエストとその応答。

復元時に、Executor はこの状態を逆シリアル化し、中断したところからワークフローを再開できるようにします。