Agentic Workflow 解説 — AI エージェントの協調パターンを概念から C# 実装まで

// Sequential: 翻訳チェーンの構築
// AgentWorkflowBuilder.BuildSequential() は内部で
// AddEdge(agent[0], agent[1]), AddEdge(agent[1], agent[2]), ...
// を自動生成する
Workflow workflow = AgentWorkflowBuilder.BuildSequential(
    from lang in new[] { "French", "Spanish", "English" }
    select new ChatClientAgent(
        chatClient,
        instructions: $"You are a translation assistant. Translate the input to {lang}.",
        name: $"{lang}Translator"
    )
);
 
// ワークフロー実行
await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(workflow, "Hello, world!");
 
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
        Console.Write(e.Update.Text);
    else if (evt is WorkflowOutputEvent output)
        Console.WriteLine($"\nFinal: {output.Data}");
}

// Concurrent: 並列翻訳
// BuildConcurrent() は内部で Fan-out Edge と Fan-in Barrier Edge を生成する
// Fan-out: 入力ノードから全エージェントに同時にメッセージを配信
// Fan-in Barrier: 全エージェントの完了を待ってから結果を集約
Workflow workflow = AgentWorkflowBuilder.BuildConcurrent(
    from lang in new[] { "French", "Spanish", "English" }
    select new ChatClientAgent(
        chatClient,
        instructions: $"Translate the input to {lang}. Output only the translation.",
        name: $"{lang}Translator"
    )
);

// Conditional Routing: スパム分類の例
// AddSwitch() で ClassificationResult の Category プロパティに基づき分岐
// ラムダ式による型安全な条件判定
WorkflowBuilder builder = new WorkflowBuilder(classifier)
    .AddSwitch(classifier, sw => sw
        .AddCase<ClassificationResult>(
            result => result?.Category == "spam",
            spamHandler)
        .AddCase<ClassificationResult>(
            result => result?.Category == "normal",
            replyGenerator)
        .WithDefault(manualReviewHandler)
    )
    .WithOutputFrom(spamHandler, replyGenerator, manualReviewHandler);

// Handoff: トリアージ → 専門エージェントへの委任
// CreateHandoffBuilderWith() でトリアージエージェントを起点に構築
// WithHandoffs() で双方向のハンドオフルートを定義
ChatClientAgent triageAgent = new(chatClient,
    instructions: "Determine which specialist to use. ALWAYS hand off.",
    name: "triage_agent",
    description: "Routes to the right specialist.");
 
ChatClientAgent mathTutor = new(chatClient,
    instructions: "You help with math. Explain step by step.",
    name: "math_tutor",
    description: "Math specialist.");
 
ChatClientAgent historyTutor = new(chatClient,
    instructions: "You help with history. Explain events and context.",
    name: "history_tutor",
    description: "History specialist.");
 
// Handoff ワークフローの構築
// WithHandoffs(source, targets) は source が targets の description を
// ツール定義として受け取り、LLM が handoff ツールを呼び出すことで
// 自動的にルーティングが行われる
Workflow workflow = AgentWorkflowBuilder
    .CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor])
    .WithHandoffs([mathTutor, historyTutor], triageAgent)
    .Build();

// Writer-Critic ループ: 反復改善ワークフロー
// WorkflowBuilder でグラフ構造を宣言的に定義
WorkflowBuilder builder = new WorkflowBuilder(writer)
    .AddEdge(writer, critic)                    // Writer → Critic
    .AddSwitch(critic, sw => sw
        .AddCase<CriticDecision>(                // 承認 → Summary
            cd => cd?.Approved == true, summary)
        .AddCase<CriticDecision>(                // 拒否 → Writer に戻る
            cd => cd?.Approved == false, writer))
    .WithOutputFrom(summary);                    // Summary の出力が最終出力

// Group Chat: マネージャーが発言順を制御する複数エージェントの会話
// CreateGroupChatBuilderWith() で GroupChatManager のファクトリを指定
// RoundRobinGroupChatManager: エージェントを順番に指名する最もシンプルなマネージャー
Workflow workflow = AgentWorkflowBuilder
    .CreateGroupChatBuilderWith(
        agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 3 })
    .AddParticipants(researchAgent, criticAgent, synthesizerAgent)
    .Build();

// Sub-Workflow: ワークフローを Executor として埋め込む
// 1. まず子ワークフローを通常通り構築
Workflow textProcessingWorkflow = new WorkflowBuilder(uppercase)
    .AddEdge(uppercase, reverse)
    .AddEdge(reverse, append)
    .WithOutputFrom(append)
    .Build();
 
// 2. BindAsExecutor() でワークフローを Executor に変換
ExecutorBinding subWorkflowExecutor =
    textProcessingWorkflow.BindAsExecutor("TextProcessingSubWorkflow");
 
// 3. 親ワークフローの中で通常の Executor と同様に接続
Workflow mainWorkflow = new WorkflowBuilder(prefix)
    .AddEdge(prefix, subWorkflowExecutor)
    .AddEdge(subWorkflowExecutor, postProcess)
    .WithOutputFrom(postProcess)
    .Build();

// Workflow-as-Agent: ワークフローを AIAgent として公開
// AsAIAgent() でワークフローを AIAgent インターフェースにラップ
Workflow workflow = WorkflowFactory.BuildWorkflow(chatClient);
AIAgent agent = workflow.AsAIAgent("workflow-agent", "Workflow Agent");
 
// 通常のエージェントと同じように対話できる
AgentSession session = await agent.CreateSessionAsync();
await foreach (AgentResponseUpdate update in
    agent.RunStreamingAsync("Hello!", session))
{
    Console.Write(update.Text);
}

<!-- csproj ファイルに追加 -->
<PackageReference Include="Microsoft.Agents.AI" Version="1.1.0" />
<PackageReference Include="Microsoft.Agents.AI.Workflows" Version="1.1.0" />
<PackageReference Include="Microsoft.Agents.AI.Workflows.Generators" Version="1.1.0"
                OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
<PackageReference Include="Microsoft.Agents.AI.OpenAI" Version="1.1.0" />
<!-- or Microsoft.Agents.AI.Foundry for Azure AI Foundry -->

// Executor の基本形
// partial キーワードとソースジェネレータにより、
// メッセージのディスパッチコードが自動生成される
internal sealed partial class UpperCaseExecutor() : Executor("UpperCase")
{
    // [MessageHandler] 属性でメッセージハンドラを定義
    // 引数の型 (string) でメッセージのルーティングが決まる
    [MessageHandler]
    public ValueTask<string> HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        return ValueTask.FromResult(message.ToUpperInvariant());
    }
}

// LLM を呼び出す Executor の例
// ChatClientAgent は IChatClient を受け取り、AIAgent インターフェースを提供する
internal sealed partial class WriterExecutor : Executor
{
    private readonly AIAgent _agent;
 
    public WriterExecutor(IChatClient chatClient) : base("Writer")
    {
        // ChatClientAgent: IChatClient → AIAgent のアダプタ
        // instructions がシステムプロンプトとして LLM に送信される
        this._agent = new ChatClientAgent(
            chatClient,
            name: "Writer",
            instructions: """
                You are a skilled writer. Create clear, engaging content.
                If you receive feedback, revise to address all concerns.
                """
        );
    }
 
    // 初回リクエスト用のハンドラ（string 型）
    [MessageHandler]
    public async ValueTask<ChatMessage> HandleInitialRequestAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // RunStreamingAsync で LLM にストリーミングリクエスト
        StringBuilder sb = new();
        await foreach (AgentResponseUpdate update in
            this._agent.RunStreamingAsync(
                new ChatMessage(ChatRole.User, message),
                cancellationToken: cancellationToken))
        {
            if (!string.IsNullOrEmpty(update.Text))
            {
                sb.Append(update.Text);
                Console.Write(update.Text); // リアルタイム出力
            }
        }
 
        return new ChatMessage(ChatRole.User, sb.ToString());
    }
 
    // 修正リクエスト用のハンドラ（CriticDecision 型）
    // 同じ Executor が異なる型のメッセージを処理できる
    [MessageHandler]
    public async ValueTask<ChatMessage> HandleRevisionAsync(
        CriticDecision decision,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        string prompt =
            $"Revise based on this feedback:\n{decision.Feedback}\n\n" +
            $"Original:\n{decision.Content}";
 
        StringBuilder sb = new();
        await foreach (AgentResponseUpdate update in
            this._agent.RunStreamingAsync(
                new ChatMessage(ChatRole.User, prompt),
                cancellationToken: cancellationToken))
        {
            if (!string.IsNullOrEmpty(update.Text))
                sb.Append(update.Text);
        }
 
        return new ChatMessage(ChatRole.User, sb.ToString());
    }
}

// Direct Edge: 無条件で A → B にメッセージを流す
builder.AddEdge(executorA, executorB);
 
// Conditional Edge: 条件を満たすときだけメッセージを流す
builder.AddEdge<ClassificationResult>(executorA, executorB,
    condition: r => r?.Category == "normal");
 
// Switch-Case Edge: 複数の条件分岐
builder.AddSwitch(executorA, sw => sw
    .AddCase<ResultType>(r => r?.IsSuccess == true, successHandler)
    .AddCase<ResultType>(r => r?.IsSuccess == false, failureHandler)
    .WithDefault(fallbackHandler));
 
// Fan-out Edge: 1 → N の並列配信
builder.AddFanOutEdge<RoutingResult>(
    sourceExecutor,
    targets: [handlerA, handlerB, handlerC],
    targetSelector: (result, count) =>
    {
        // result の内容に応じて実行先を選択
        // 戻り値は実行先のインデックスのリスト
        if (result?.NeedsAll == true) return Enumerable.Range(0, count);
        return [0]; // handlerA のみ
    });
 
// Fan-in Barrier Edge: N → 1 の集約（全ソースの完了待ち）
builder.AddFanInBarrierEdge(
    sources: [workerA, workerB, workerC],
    target: aggregator);

// WorkflowBuilder のフル活用例
Workflow workflow = new WorkflowBuilder(entryExecutor)
    // Direct Edge: entry → processor
    .AddEdge(entryExecutor, processorExecutor)
    // Switch-Case: processor の出力で分岐
    .AddSwitch(processorExecutor, sw => sw
        .AddCase<ProcessResult>(r => r?.Status == "approved", approvedHandler)
        .AddCase<ProcessResult>(r => r?.Status == "rejected", rejectedHandler)
        .WithDefault(manualReviewHandler))
    // 出力元の指定: これらの Executor の出力がワークフロー全体の出力になる
    .WithOutputFrom(approvedHandler, rejectedHandler, manualReviewHandler)
    // ビルド: Workflow インスタンスの生成
    .Build();

// ストリーミング実行
// InProcessExecution.RunStreamingAsync() は StreamingRun を返す
// StreamingRun は IAsyncDisposable で、using で確実にリソースが解放される
await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(workflow, "入力メッセージ");
 
// WatchStreamAsync() でリアルタイムにイベントを購読
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case AgentResponseUpdateEvent agentUpdate:
            // エージェントのストリーミング出力（トークン単位）
            Console.Write(agentUpdate.Update.Text);
            break;
 
        case WorkflowOutputEvent output:
            // ワークフロー全体の最終出力
            Console.WriteLine($"\n=== Final Output ===");
            Console.WriteLine(output.Data);
            break;
    }
}

// InProcessRunner.RunSuperStepAsync() の内部動作（簡略化）
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs
async ValueTask<bool> RunSuperStepAsync(CancellationToken cancellationToken)
{
    // 1. AdvanceAsync(): 前のスーパーステップでキューされた
    //    メッセージを、現在のスーパーステップの StepContext に移動
    StepContext currentStep = await RunContext.AdvanceAsync(cancellationToken);
 
    // メッセージがなければスーパーステップは不要
    if (!currentStep.HasMessages && !RunContext.HasQueuedExternalDeliveries)
        return false;
 
    // 2. SuperStepStartedEvent を発火
    await RaiseWorkflowEventAsync(StepTracer.Advance(currentStep));
 
    // 3. 全ての宛先 Executor に対してメッセージを並行配信
    //    Task.WhenAll で同一スーパーステップ内の Executor を並行実行
    List<Task> receiverTasks = currentStep.QueuedMessages.Keys
        .Select(receiverId =>
            DeliverMessagesAsync(receiverId,
                currentStep.MessagesFor(receiverId),
                cancellationToken).AsTask())
        .ToList();
    await Task.WhenAll(receiverTasks);
 
    // 4. サブワークフローがあればそちらも並行駆動
    List<Task> subworkflowTasks = [];
    foreach (var subRunner in RunContext.JoinedSubworkflowRunners)
        subworkflowTasks.Add(subRunner.RunSuperStepAsync(cancellationToken).AsTask());
    await Task.WhenAll(subworkflowTasks);
 
    // 5. バリア同期: ステート更新をバッチ適用
    //    StateManager.PublishUpdatesAsync() が
    //    QueueStateUpdateAsync() でキューされた更新を一括適用
    await StateManager.PublishUpdatesAsync(StepTracer);
 
    // 6. チェックポイント作成（有効時）
    await CheckpointAsync(cancellationToken);
 
    // 7. SuperStepCompletedEvent を発火
    await RaiseWorkflowEventAsync(
        StepTracer.Complete(
            RunContext.NextStepHasActions,
            RunContext.HasUnservicedRequests));
}

// StepContext: スーパーステップのメッセージキュー管理
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StepContext.cs
internal sealed class StepContext
{
    // 宛先 Executor ID → メッセージキュー
    // ConcurrentDictionary + ConcurrentQueue でスレッドセーフ
    public ConcurrentDictionary<string, ConcurrentQueue<MessageEnvelope>>
        QueuedMessages { get; } = [];
 
    public bool HasMessages =>
        !QueuedMessages.IsEmpty &&
        QueuedMessages.Values.Any(q => !q.IsEmpty);
}

// 実行モードの選択
// デフォルト（OffThread）: リアルタイムストリーミング
await using var run = await InProcessExecution.RunStreamingAsync(workflow, input);
 
// Lockstep: テスト・デバッグ用（イベントがバッチで配信される）
await using var run = await InProcessExecution.Lockstep
    .RunStreamingAsync(workflow, input);
 
// Concurrent: 同一ワークフローの並行実行
// ワークフロー内の全 Executor が IResettableExecutor を実装している必要がある
await using var run = await InProcessExecution.Concurrent
    .RunStreamingAsync(workflow, input);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case SuperStepStartedEvent stepStarted:
            // SendingExecutors: 前ステップでメッセージを送った Executor の ID
            Console.WriteLine(
                $"[Step Start] Senders: {string.Join(", ", stepStarted.StartInfo.SendingExecutors)}");
            break;
 
        case SuperStepCompletedEvent stepCompleted:
            // ActivatedExecutors: このステップで実行された Executor の ID
            // HasPendingMessages: 次のステップに配送予定のメッセージがあるか
            // StateUpdated: ステートが更新されたか
            Console.WriteLine(
                $"[Step Done] Activated: {string.Join(", ", stepCompleted.CompletionInfo.ActivatedExecutors)}, " +
                $"Pending: {stepCompleted.CompletionInfo.HasPendingMessages}, " +
                $"StateUpdated: {stepCompleted.CompletionInfo.StateUpdated}");
            break;
 
        case AgentResponseUpdateEvent agentUpdate:
            Console.Write(agentUpdate.Update.Text);
            break;
 
        case WorkflowOutputEvent output:
            Console.WriteLine($"\nFinal: {output.Data}");
            break;
    }
}

Executor A が context.SendMessageAsync(output) を呼ぶ
  → InProcessRunnerContext.SendMessageAsync(sourceId, output) が呼ばれる
    → sourceId に紐づく全 Edge を走査
      → EdgeMap.PrepareDeliveryForEdgeAsync() で条件判定
        → DeliveryMapping.MapInto(_nextStep) で
          次のスーパーステップ用 StepContext にエンキュー
            → _nextStep.MessagesFor(targetId).Enqueue(envelope)

// InProcessRunnerContext.AdvanceAsync() — スーパーステップの切り替え
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
public async ValueTask<StepContext> AdvanceAsync(CancellationToken cancellationToken = default)
{
    // まず外部メッセージを Edge 経由で _nextStep にルーティング
    while (_queuedExternalDeliveries.TryDequeue(out var deliveryPrep))
        await deliveryPrep();
 
    // _nextStep をアトミックに取り出して新しい空の StepContext に差し替え
    return Interlocked.Exchange(ref _nextStep, new StepContext());
}

// InProcessRunner.DeliverMessagesAsync() — 1 つの Executor 宛てのキューを順に処理する（抜粋）
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs
private async ValueTask DeliverMessagesAsync(string receiverId, ConcurrentQueue<MessageEnvelope> envelopes, CancellationToken cancellationToken)
{
    Executor executor = await this.RunContext.EnsureExecutorAsync(receiverId, this.StepTracer, cancellationToken).ConfigureAwait(false);
 
    this.StepTracer.TraceActivated(receiverId);
 
    // ...
    IWorkflowContext tracelessContext = this.RunContext.BindWorkflowContext(receiverId);
 
    try
    {
        await executor.OnMessageDeliveryStartingAsync(tracelessContext, cancellationToken)
                      .ConfigureAwait(false);
 
        while (envelopes.TryDequeue(out var envelope))
        {
            (object message, TypeId messageType) = await TranslateMessageAsync(envelope).ConfigureAwait(false);
 
            await executor.ExecuteCoreAsync(
                message,
                messageType,
                this.RunContext.BindWorkflowContext(receiverId, envelope.TraceContext),
                this.TelemetryContext,
                cancellationToken
            ).ConfigureAwait(false);
        }
    }
    finally
    {
        await executor.OnMessageDeliveryFinishedAsync(tracelessContext, cancellationToken)
                      .ConfigureAwait(false);
    }
 
    // （各メッセージの型を復元するローカル関数 TranslateMessageAsync は省略）
}

// ConcurrentEventSink — イベント通知の仕組み
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ConcurrentEventSink.cs
internal class ConcurrentEventSink : IEventSink
{
    public ValueTask EnqueueAsync(WorkflowEvent workflowEvent)
        => this.EventRaised?.Invoke(this, workflowEvent) ?? default;
 
    public event Func<object?, WorkflowEvent, ValueTask>? EventRaised;
}

// CriticDecision: Structured Output 用の DTO
// Description 属性は JSON Schema の description フィールドにマップされ、
// LLM がフィールドの意味を理解するためのヒントになる
[Description("Critic's review decision")]
internal sealed class CriticDecision
{
    [JsonPropertyName("approved")]
    [Description("Whether the content is approved (true) or needs revision (false)")]
    public bool Approved { get; set; }
 
    [JsonPropertyName("feedback")]
    [Description("Specific feedback for improvements if not approved")]
    public string Feedback { get; set; } = "";
 
    // ワークフロー内部で使用するフィールド（JSON には含まれない）
    [JsonIgnore]
    public string Content { get; set; } = "";
 
    [JsonIgnore]
    public int Iteration { get; set; }
}
 
// Critic Executor: Structured Output を使ったレビュー
internal sealed class CriticExecutor : Executor<ChatMessage, CriticDecision>
{
    private readonly AIAgent _agent;
 
    public CriticExecutor(IChatClient chatClient) : base("Critic")
    {
        this._agent = new ChatClientAgent(chatClient, new ChatClientAgentOptions
        {
            Name = "Critic",
            ChatOptions = new()
            {
                Instructions = """
                    You are a constructive critic. Review content and provide feedback.
                    Provide your decision as structured output.
                    """,
                // ChatResponseFormat.ForJsonSchema<T>() で
                // CriticDecision の JSON Schema を LLM に送信
                // LLM はこのスキーマに準拠した JSON を必ず返す
                ResponseFormat = ChatResponseFormat.ForJsonSchema<CriticDecision>()
            }
        });
    }
 
    public override async ValueTask<CriticDecision> HandleAsync(
        ChatMessage message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // ストリーミングでレスポンスを取得
        IAsyncEnumerable<AgentResponseUpdate> updates =
            this._agent.RunStreamingAsync(message, cancellationToken: cancellationToken);
 
        // ストリーミング出力をリアルタイムで表示
        await foreach (AgentResponseUpdate update in updates)
        {
            if (!string.IsNullOrEmpty(update.Text))
                Console.Write(update.Text);
        }
 
        // ストリームを AgentResponse に変換し、JSON をデシリアライズ
        AgentResponse response = await updates.ToAgentResponseAsync(cancellationToken);
        CriticDecision decision = JsonSerializer.Deserialize<CriticDecision>(
            response.Text, JsonSerializerOptions.Web)
            ?? throw new JsonException("Failed to deserialize CriticDecision");
 
        return decision;
    }
}

// 共有ステートの読み書き
// IWorkflowContext は各 Executor に注入される
// scopeName でスコープを分離し、key で個別のステートを識別する
 
// ステートの読み取り
FlowState? state = await context.ReadStateAsync<FlowState>(
    key: "singleton",
    scopeName: "FlowStateScope");
 
// ステートの書き込み（キュー経由でスーパーステップ終了時に適用）
await context.QueueStateUpdateAsync(
    key: "singleton",
    value: updatedState,
    scopeName: "FlowStateScope");

// IResettableExecutor: ステートフルな共有 Executor の再利用
// フレームワーク提供の Executor（ChatForwardingExecutor 等）は既に実装済み
// カスタム Executor でステートを持つ場合は自分で実装する必要がある
internal sealed class AggregationExecutor()
    : Executor<List<ChatMessage>>("Aggregator"), IResettableExecutor
{
    private readonly List<ChatMessage> _messages = [];
 
    public override async ValueTask HandleAsync(
        List<ChatMessage> message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        this._messages.AddRange(message);
        if (this._messages.Count >= 2)
            await context.YieldOutputAsync(
                string.Join("\n", this._messages.Select(m => m.Text)),
                cancellationToken);
    }
 
    // フレームワークがワークフロー実行間で自動的に呼び出す
    public ValueTask ResetAsync()
    {
        this._messages.Clear();
        return default;
    }
}

// ⚠️ 両方の条件が true なら、processorA と processorB の両方が実行される！
builder
    .AddEdge<MyResult>(source, processorA, r => r?.Score > 50)
    .AddEdge<MyResult>(source, processorB, r => r?.Score > 30);
// Score = 70 の場合 → processorA と processorB の両方にメッセージが配信される

// InProcessRunnerContext.SendMessageAsync() — 全 Edge を独立評価
foreach (Edge edge in edges)  // break も return もない
{
    DeliveryMapping? maybeMapping =
        await _edgeMap.PrepareDeliveryForEdgeAsync(edge, envelope, cancellationToken);
    maybeMapping?.MapInto(this._nextStep);  // マッチした分だけ全部配信
}

// SwitchBuilder 内部の EdgeSelector（簡略化）
IEnumerable<int> EdgeSelector(object? input, int targetCount)
{
    for (int i = 0; i < caseMap.Count; i++)
    {
        if (caseMap[i].Predicate(input))
            return caseMap[i].OutgoingIndicies;  // 最初のマッチで即 return
    }
    return defaultIndicies;  // どれにもマッチしなければデフォルト
}

// DirectEdgeRunner.ChaseEdgeAsync() — 型不一致はサイレントにドロップ
if (CanHandle(target, messageType))
{
    return new DeliveryMapping(envelope, target);  // 配信
}
// ↓ ここに到達すると DroppedTypeMismatch — 例外は発生しない
return null;

// ⚠️ targetSelector 省略 → 全ターゲットに配信
builder.AddFanOutEdge(source, [handlerA, handlerB, handlerC]);
 
// ✅ targetSelector で配信先を制御
builder.AddFanOutEdge<MyResult>(source, [handlerA, handlerB, handlerC],
    targetSelector: (result, count) =>
    {
        if (result?.Category == "urgent") return [0];  // handlerA のみ
        return Enumerable.Range(0, count);              // 全ターゲット
    });

// ⚠️ workerC がメッセージを送らないと、aggregator は永遠に実行されない
builder.AddFanInBarrierEdge(
    sources: [workerA, workerB, workerC],
    target: aggregator);

// ✅ 同一 Executor 内では自分の書き込みは即座に読める
await context.QueueStateUpdateAsync("key", newValue, scopeName: "MyScope");
var value = await context.ReadStateAsync<MyState>("key", scopeName: "MyScope");
// value は newValue が返る（StateManager が _queuedUpdates を参照するため）
 
// ⚠️ ただし、別の Executor がキューした更新は同一スーパーステップ内では見えない
// Executor A の書き込みは、同一スーパーステップ内の Executor B からは読めない
// 反映されるのは次のスーパーステップから（PublishUpdatesAsync() 実行後）

// ⚠️ 例外がスローされる
builder.AddEdge(source, target);
builder.AddEdge(source, target);  // InvalidOperationException!
 
// ✅ idempotent で安全に
builder.AddEdge(source, target, idempotent: true);
builder.AddEdge(source, target, idempotent: true);  // OK（2回目は NoOp）

// Program.cs — Writer-Critic ワークフローのエントリーポイント
// https://github.com/microsoft/agent-framework/blob/main/dotnet/samples/03-workflows/_StartHere/07_WriterCriticWorkflow/Program.cs
using System.ComponentModel;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
 
public static class Program
{
    public const int MaxIterations = 3; // 安全弁: 最大3回のイテレーション
 
    private static async Task Main()
    {
        // ── 1. IChatClient の初期化 ──
        string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
            ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
        string deploymentName =
            Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME")
            ?? "gpt-4o-mini";
 
        // IChatClient: Microsoft.Extensions.AI の統一インターフェース。
        // 現在の公式サンプルは AIProjectClient.AsAIAgent() で初期化するが、
        // ここではプロバイダ差し替えを示すため ChatClientAgent + IChatClient を使う。
        IChatClient chatClient = new AzureOpenAIClient(
            new Uri(endpoint),
            new AzureCliCredential()
        ).GetChatClient(deploymentName).AsIChatClient();
 
        // ── 2. Executor の生成 ──
        WriterExecutor writer = new(chatClient);
        CriticExecutor critic = new(chatClient);
        SummaryExecutor summary = new(chatClient);
 
        // ── 3. ワークフローグラフの構築 ──
        // WorkflowBuilder はコンストラクタの引数がエントリーポイント
        WorkflowBuilder workflowBuilder = new WorkflowBuilder(writer)
            .AddEdge(writer, critic)           // Writer → Critic
            .AddSwitch(critic, sw => sw        // Critic で条件分岐
                .AddCase<CriticDecision>(
                    cd => cd?.Approved == true,
                    summary)                   //   承認 → Summary
                .AddCase<CriticDecision>(
                    cd => cd?.Approved == false,
                    writer))                   //   拒否 → Writer（ループ）
            .WithOutputFrom(summary);          // Summary の出力が最終出力
 
        // ── 4. ワークフロー実行 ──
        Workflow workflow = workflowBuilder.Build();
        await using StreamingRun run =
            await InProcessExecution.RunStreamingAsync(
                workflow,
                "Write a 200-word blog post about AI ethics.");
 
        // ── 5. イベント購読 ──
        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
        {
            switch (evt)
            {
                case AgentResponseUpdateEvent agentUpdate:
                    if (!string.IsNullOrEmpty(agentUpdate.Update.Text))
                        Console.Write(agentUpdate.Update.Text);
                    break;
 
                case WorkflowOutputEvent output:
                    Console.WriteLine($"\n\n=== FINAL APPROVED CONTENT ===");
                    Console.WriteLine(output.Data);
                    break;
            }
        }
    }
}

// FlowState: Executor 間で共有されるイテレーション追跡用ステート
// IWorkflowContext の ReadStateAsync / QueueStateUpdateAsync で読み書きする
internal sealed class FlowState
{
    public int Iteration { get; set; } = 1;
    public List<ChatMessage> History { get; } = [];
}
 
// ステートのスコープとキーの定数定義
internal static class FlowStateShared
{
    public const string Scope = "FlowStateScope";
    public const string Key = "singleton";
}
 
// ヘルパーメソッド
internal static class FlowStateHelpers
{
    public static async Task<FlowState> ReadFlowStateAsync(IWorkflowContext context)
    {
        FlowState? state = await context.ReadStateAsync<FlowState>(
            FlowStateShared.Key,
            scopeName: FlowStateShared.Scope);
        return state ?? new FlowState();
    }
 
    public static ValueTask SaveFlowStateAsync(
        IWorkflowContext context, FlowState state)
        => context.QueueStateUpdateAsync(
            FlowStateShared.Key,
            state,
            scopeName: FlowStateShared.Scope);
}

// CriticExecutor 内の安全弁ロジック
if (!decision.Approved && state.Iteration >= Program.MaxIterations)
{
    Console.WriteLine(
        $"⚠️ Max iterations ({Program.MaxIterations}) reached - auto-approving");
    decision.Approved = true;
    decision.Feedback = "";
}

// Executor 内のリトライとエラーハンドリング
[MessageHandler]
public async ValueTask<ChatMessage> HandleAsync(
    string message,
    IWorkflowContext context,
    CancellationToken cancellationToken = default)
{
    const int maxRetries = 3;
    for (int attempt = 1; attempt <= maxRetries; attempt++)
    {
        try
        {
            StringBuilder sb = new();
            await foreach (AgentResponseUpdate update in
                this._agent.RunStreamingAsync(
                    new ChatMessage(ChatRole.User, message),
                    cancellationToken: cancellationToken))
            {
                if (!string.IsNullOrEmpty(update.Text))
                    sb.Append(update.Text);
            }
            return new ChatMessage(ChatRole.User, sb.ToString());
        }
        catch (HttpRequestException ex) when (attempt < maxRetries)
        {
            // レート制限やタイムアウトの場合はエクスポネンシャルバックオフで再試行
            int delay = (int)Math.Pow(2, attempt) * 1000;
            Console.WriteLine($"⚠️ Attempt {attempt} failed: {ex.Message}. Retrying in {delay}ms...");
            await Task.Delay(delay, cancellationToken);
        }
    }
    throw new InvalidOperationException("All retry attempts exhausted");
}

// チェックポイントの設定
// FileSystemJsonCheckpointStore はファイルシステムにステートを保存する
// 本番環境では Cosmos DB や Azure Table Storage ベースの ICheckpointStore を使うことが多い
var store = new FileSystemJsonCheckpointStore("./checkpoints");
var checkpointManager = new CheckpointManager(store);
 
// チェックポイント付きワークフロー実行
await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(
        workflow,
        input,
        checkpointManager);

// OpenTelemetry セットアップ例
using var tracerProvider = Sdk.CreateTracerProviderBuilder()
    .SetResourceBuilder(ResourceBuilder.CreateDefault()
        .AddService("WriterCriticWorkflow"))
    .AddSource("Microsoft.Agents.AI")     // Agent Framework のトレース
    .AddOtlpExporter()                     // Jaeger / Azure Monitor 等に送信
    .Build();