耐久性の高い拡張機能

Microsoft Agent Framework の Durable Extension は、エージェント、マルチエージェントオーケストレーション、および Microsoft Agent Framework ワークフローに永続的な実行をもたらします。これを使用すると、コアエージェントロジックを変更することなく、エージェントセッションの永続化、チェックポイントオーケストレーションとワークフローの進行状況の保持、障害からの復旧、分散ホスト間での作業のスケーリングを行うことができます。

この拡張機能では、C# と Python の 2 つのホスティングモデルがサポートされています。

Azure Functions Azure Functions プログラミングモデルを使用したマネージドサーバーレスホスティング用。
独自のワーカープロセス、サービス、コンテナー、Kubernetes 環境、または既存のアプリインフラストラクチャで永続的なエージェントとワークフローを実行するための独自のコンピューティング/セルフホステッド。

Note

このガイドでは、C# と Python Durable Extension ホスティングについて説明します。 Go ホスティングについては、 A2A ホスティングと Agent Framework Go リポジトリを参照してください。

概要

非消耗品エージェントは、Agent Framework プログラミングモデルと Durable Task インフラストラクチャ ( Durable Task Scheduler など) を組み合わせて、次のようなエージェントを作成します。

要求と worker の実行全体で状態を自動的に保持する
会話コンテキストを失ったり、完了した作業を繰り返したりせずに、失敗後に再開する
需要に基づいて分散されたステートレスワーカー間でスケーリングする
信頼性の高い実行保証を使用してマルチエージェントワークフローを調整する
グラフベースのワークフローモデルを使用して構築された Checkpoint Agent Framework ワークフロー
待機中にコンピューティングトークンまたはモデルトークンを使用せずに、人間の入力イベントまたは外部イベントを一時停止する
Redis などのストリームブローカーで構成されている場合は、ストリーム配信を再開します。永続的な完了を個別に確認する
セッション の有効期間 (TTL) のクリーンアップとダッシュボードベースの監視を使用してセッションライフサイクルを管理する

耐久性のあるエージェントを使用する時

必要に応じ、永続的なエージェントを選択します。

永続的な会話状態: エージェントセッションは、プロセスのクラッシュ、再起動、およびスケールアウトイベントに存続します
複雑なオーケストレーション: 数日または数週間実行できる確定的で信頼性の高いワークフローで複数のエージェントを調整する
イベントドリブンオーケストレーション: トリガー、キュー、Webhook、タイマー、または既存のアプリケーションイベントとの統合
自動会話状態: エージェントの会話履歴は自動的に管理され、コードで明示的な状態処理を必要とせずに保持されます
Durable Agent Framework ワークフロー: グラフベースの Microsoft Agent Framework ワークフローを永続的にして、各ステップをチェックポイント処理して再開できるようにします
有効期間の長いセッション: セッションの Time-to-Live (TTL) クリーンアップを使用してアイドル状態のセッションを自動的に削除しながら、有用な会話を利用できるようにします
リアルタイム応答: 永続的な結果を個別にチェックしながら、ブローカーベースの再接続を使用して増分トークン出力を配信する

このホスティング方法は、マネージドサービスベースのエージェントホスティング (Foundry Agent Service など) とは異なります。これは、ワーカーホストのデプロイや管理を必要とせずにフルマネージドインフラストラクチャを提供します。永続的なエージェントは、永続的な状態管理と組み合わせたコード優先デプロイの柔軟性が必要な場合に最適です。

ホスティングモデルを選択する

ホスティングモデル	必要なときに選択する
Azure Functions	管理されたサーバーレスホスティングモデル。組み込みのスケールアウトとゼロへのスケールアウト。トリガーとバインドのAzure Functions;Functions プログラミングモデルによって生成される HTTP エンドポイント。MCP サーバートリガー。ホストインフラストラクチャ管理を最小限に抑えることができます。
Bring-your-own-compute / self-hosted	ホストプロセス、デプロイ環境、ランタイムライフサイクル、インフラストラクチャ、ネットワーク、認証、または既存のアプリまたはサービスとの統合をより詳細に制御します。このモデルは、コンテナー、Kubernetes、実行時間の長いワーカー、コンソールアプリ、カスタムサービス、または Functions 以外のホスティング環境に使用します。

ホスティングモデル

必要なときに選択する

Azure Functions

管理されたサーバーレスホスティングモデル。組み込みのスケールアウトとゼロへのスケールアウト。トリガーとバインドのAzure Functions;Functions プログラミングモデルによって生成される HTTP エンドポイント。MCP サーバートリガー。ホストインフラストラクチャ管理を最小限に抑えることができます。

Bring-your-own-compute / self-hosted

ホストプロセス、デプロイ環境、ランタイムライフサイクル、インフラストラクチャ、ネットワーク、認証、または既存のアプリまたはサービスとの統合をより詳細に制御します。このモデルは、コンテナー、Kubernetes、実行時間の長いワーカー、コンソールアプリ、カスタムサービス、または Functions 以外のホスティング環境に使用します。

Azure Functions Flex Consumption ホスティングプランでホストされている場合、エージェントは何千ものインスタンスまたは使用されていない場合はゼロインスタンスにスケーリングでき、必要なコンピューティングに対してのみ料金を支払うことができます。セルフホステッドシナリオでは、独自のホストがプロセスの有効期間、スケーリング、ネットワーク、デプロイを制御します。

作業の開始

以下のホスティングモデルのインストールコマンドを使用します。プレビューパッケージのプレリリースオプションは表示したままにします。完全なアプリケーションについては、ソースとサンプルを参照してください。

ソースブランチのドキュメントでは、発行済みパッケージではまだ使用できない機能について説明できます。インストールされているリリースに含まれる機能にのみ、構成と移行の手順を適用します。

Important

既存のデプロイをアップグレードする前に、デプロイとスキーマの移行を確認します。再生の互換性を確認せずに、アップグレードされたワーカーを古いワークフロー履歴でポイントしないでください。

.NET プロジェクトで、ホスティングモデルのパッケージセットを選択します。

Azure Functions ホスティングの場合は、Azure Functions統合パッケージと Functions ワーカーパッケージを追加します。

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Hosting.AzureFunctions --prerelease

Note

エージェント拡張機能の依存関係によって宣言された Functions Worker、DurableTask worker 拡張機能、および AzureManaged worker 拡張機能のバージョンを使用します。これらの依存関係をダウングレードする直接パッケージ参照を削除します。 .NET依存関係の参照を参照してください。 side-by-side schema-v2 アップグレードでは、作業項目のフィルター処理も必要です。デプロイとスキーマの移行に関する説明を参照してください。

独自のコンピューティングの持ち込みホスティングの場合は、基本の Durable Task 統合パッケージと、ホストで使用される Durable Task Scheduler worker/クライアントパッケージを追加します。

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.DurableTask --prerelease
dotnet add package Microsoft.DurableTask.Client.AzureManaged
dotnet add package Microsoft.DurableTask.Worker.AzureManaged
dotnet add package Microsoft.Extensions.Hosting

Python プロジェクトで、ホスティングモデルのパッケージを選択します。 Pythonホストには、明示的なデプロイ戦略が必要です。

Azure Functions ホスティングの場合は、Python 3.13 以降を使用し、Azure Functions統合パッケージをインストールします。次のコマンドでは、このガイドで使用Azure OpenAI クライアントもインストールします。

pip install agent-framework-azurefunctions agent-framework-openai azure-identity --pre

独自のコンピューティング持ち込みホスティングの場合は、Python 3.10 以降を使用し、Durable Task 統合パッケージをインストールします。

pip install agent-framework-durabletask agent-framework-openai azure-identity --pre

ワーカーとクライアントがアップグレードされた新しい分離タスクハブの場合は、いずれかのホストを開始する前に DURABLE_AGENTS_DEPLOYMENT_MODE=isolated_v2 設定します。既存のハブの場合は、この設定を盲目的に適用するのではなく、移行戦略を選択します。この設定は分離を確認します。他のワーカーを検出したり、古い履歴を互換性のあるものにしたりすることはありません。

Azure Functions ホスティング

Durable Extension を使用すると、組み込みの HTTP エンドポイントとオーケストレーションベースの呼び出しを使用して、Azure Functions にMicrosoft Agent Framework エージェントをデプロイしてホストできます。 Azure Functions は、自動スケーリングと最小限のインフラストラクチャ管理を使用して、イベントドリブンの呼び出しごとの料金を提供します。

Azure Functionsで永続的なエージェントを構成すると、拡張機能によってエージェントの HTTP エンドポイントが自動的に作成され、会話状態の格納、同時要求の処理、マルチエージェントワークフローの調整のための基になるインフラストラクチャが管理されます。また、Azure Functions ホスティング統合では、メッセージの送信、状態の確認、セッションの管理のための生成された REST API や、トリガーの接着剤を記述せずに MCP サーバーとしてエージェントをホストするための MCP サーバートリガーなどのトリガーなど、関数固有の便利さも提供されます。

using System;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") ?? "gpt-4o-mini";

// Create an AI agent following the standard Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are good at telling jokes.",
        name: "Joker");

// Configure the function app to host the agent with durable thread management
// This automatically creates HTTP endpoints and manages state persistence
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options =>
        options.AddAIAgent(agent)
    )
    .Build();
app.Run();

Warning

DefaultAzureCredential は開発には便利ですが、運用環境では慎重に考慮する必要があります。運用環境では、待機時間の問題、意図しない資格情報のプローブ、フォールバックメカニズムによる潜在的なセキュリティリスクを回避するために、特定の資格情報 ( ManagedIdentityCredential など) を使用することを検討してください。

import os
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create an AI agent following the standard Microsoft Agent Framework pattern
agent = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
).as_agent(
    instructions="You are good at telling jokes.",
    name="Joker"
)

# Configure the function app to host the agent with durable thread management
# This automatically creates HTTP endpoints and manages state persistence
app = AgentFunctionApp(agents=[agent])

独自のコンピューティングの持ち込み/セルフホステッドホスティング

Azure Functions プログラミングモデルを使用せずに Durable Extension 機能が必要な場合は、独自のコンピューティング持ち込みホスティングを使用します。このモデルでは、プロセスによって Durable Task worker が開始され、永続的なエージェントまたはワークフローが登録され、Durable Task Scheduler バックエンドに接続されます。クライアントコードは、同じプロセスまたは別のサービスで実行できます。

セルフホステッドワーカーは、Azure Functionsホスティングと同じコア Durable Extension 機能を使用します。チェックポイント処理と再開、決定論的エージェントオーケストレーション、永続的なエージェントフレームワークワークフロー、人間のループ内待機、信頼性の高いストリーミング、アイドルセッションのクリーンアップ、ダッシュボードの可視性、ステートレス worker インスタンス間での分散実行。ホストは、独自の API、ライフサイクル管理、ネットワーク、認証、デプロイモデルを公開する役割を担います。

基本の Durable Task 統合パッケージを使用してホストを構成します。永続的なエージェントには ConfigureDurableAgents を使用し、グラフベースの Microsoft Agent Framework ワークフローには ConfigureDurableWorkflows を使用します。

string connectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableAgents(
            options => options.AddAIAgent(agent),
            workerBuilder: builder => builder.UseDurableTaskScheduler(connectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(connectionString));
    })
    .Build();

await host.StartAsync();

実行可能なセルフホステッドの例については、.NET Durable Agents コンソールのサンプルおよび .NET Durable Workflows コンソールのサンプルを参照してください。

Durable Task 統合パッケージを使用して、エージェントを登録して要求をリッスンするワーカープロセスを実行します。クライアントコードは、別のプロセスから同じ Durable Task Scheduler タスクハブに接続できます。

from agent_framework_durabletask import DurableAIAgentWorker
from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker

worker = DurableTaskSchedulerWorker(
    host_address="http://localhost:8080",
    secure_channel=False,
    taskhub="default",
)

agent_worker = DurableAIAgentWorker(worker)
agent_worker.add_agent(agent)

worker.start()

Python Durable Task のサンプルを参照して、単一エージェントホスティング、マルチエージェントルーティング、信頼性の高いストリーミング、オーケストレーションチェーン、コンカレンシー、条件、および人間のループ内パターンなど、worker クライアントの例を確認してください。

Durable Agent Framework ワークフロー

持続性は、永続的なオーケストレーションに限定されません。グラフベースのワークフローモデルを使用して構築された Microsoft Agent Framework ワークフローを永続的にすることもできます。 Durable Extension チェックポイントワークフローの実行により、完了した Executor とエージェントのステップは、プロセスの再起動または失敗後に繰り返されません。

コードベースの分岐、タイマー、アクティビティ、および外部イベントとの命令型の調整が必要な場合は、永続的オーケストレーションを使用します。型指定されたルーティング、ファンアウト/ファンイン、条件付きエッジ、ワークフローイベント、共有状態、サブワークフロー、または人間のループ内要求ポートを含む Executor とエージェントの宣言型グラフが必要な場合は、Durable Agent Framework ワークフローを使用します。

Note

Durable Agent Framework ワークフローは、標準ワークフローのチェックポイントストレージとは異なります。チェックポイントストレージは、Agent Framework ランタイムでワークフローの実行を再開するのに役立ちます。 Durable Extension は Durable Task インフラストラクチャでワークフローを実行するため、ワークフローの進行状況はチェックポイント処理され、分散された非消耗品ワーカー間で復旧されます。標準ワークフローのチェックポイント処理については、「チェックポイントと再開」を参照してください。

セルフホステッドアプリの場合は ConfigureDurableWorkflows、Azure Functions ホスティング用の Functions アプリビルダーでは ConfigureDurableWorkflows にグラフベースのワークフローを登録します。

.NET Durable Workflows Azure Functions サンプルおよび .NET Durable Workflows コンソールのサンプルを参照してください。

永続的なワークフローサンプルは、共有状態、共有状態なし、並列ワークフロー実行、人間のループ内ワークフローなど、Azure Functions ホスティングで使用できます。 Pythonでは、スタンドアロンの Durable Task worker でのグラフベースのワークフローもサポートされています。

非消耗品エージェント、オーケストレーション、MCP サーバー、ワークフローの例については、Python Azure Functions サンプルを参照してください。

セルフホステッドグラフの実行については、スタンドアロンのPythonワークフローサンプルを参照してください。

Samples

Language	ホスティングモデル	永続的なエージェントとオーケストレーション	グラフベースのワークフロー
C#	Azure Functions	.NET エージェントのサンプル	.NET ワークフローのサンプル
C#	Self-hosted	.NET コンソールエージェントのサンプル	コンソールワークフローサンプル.NETする
Python	Azure Functions	Python Functions のサンプル	Python Functions ワークフローのサンプル
Python	Self-hosted	Python worker とクライアントのサンプル	スタンドアロンワークフロー Python

これらのサンプルは、複数の言語で同一の API コントラクトではなく、サポートを示しています。関数は HTTP エンドポイントを生成します。セルフホステッドアプリケーションは、独自の API を実装してセキュリティで保護します。

会話履歴を含むステートフルエージェントスレッド

エージェントは、複数の相互作用にわたって存続する永続的なセッションを維持します。各セッションには一意のセッション ID があり、 Durable Task Scheduler などの Durable Task インフラストラクチャによって管理される永続的ストレージに会話状態が格納されます。

このパターンにより、プロセスのクラッシュと再起動によってエージェントの状態が保持される会話の継続性が可能になり、ユーザースレッド間で完全な会話履歴を維持できます。永続ストレージを使用すると、ホストプロセスが再起動したり、別のワーカーインスタンスで作業が再開されたりした場合でも、中断した場所から会話がシームレスに続行されます。

セッションのクリーンアップ、履歴の保持、結果の保持を意図的に構成します。これらのポリシーの効果は異なります。状態のライフサイクルと容量を参照してください。

次のAzure Functions例は、同じセッションに対する複数の HTTP 要求を示しています。どちらの言語統合も、 x-ms-session-id を返し、正規の sessionId クエリパラメーターを受け入れます。返された ID を不透明として扱います。この例では、.NET schema-v2 エンティティアドレスが示されていますが、セッションの書式設定Pythonはデプロイ戦略によって異なります。セルフホステッドアプリでは、独自のプロセスまたはサービスから Durable Task クライアント API を使用します。

# First interaction - start a new session
curl -i -X POST https://your-function-app.azurewebsites.net/api/agents/Joker/run \
  -H "Content-Type: text/plain" \
  -d "Tell me a joke about pirates"

# Response includes session ID in x-ms-session-id header and joke as plain text
# HTTP/1.1 200 OK
# Content-Type: text/plain
# x-ms-session-id: @dafx-v2-joker@263fa373-fa01-4705-abf2-5a114c2bb87d
#
# Why don't pirates shower before they walk the plank? Because they'll just wash up on shore later!

# Second interaction - reuse the exact session ID returned by the first request
curl -X POST "https://your-function-app.azurewebsites.net/api/agents/Joker/run?sessionId=@dafx-v2-joker@263fa373-fa01-4705-abf2-5a114c2bb87d" \
  -H "Content-Type: text/plain" \
  -d "Tell me another one about the same topic"

# Agent remembers the pirate context from the first message and responds with plain text
# What's a pirate's favorite letter? You'd think it's R, but it's actually the C!

エージェントの状態は永続的ストレージで維持され、複数のインスタンス間で分散実行が可能になります。中断または障害の後、任意のインスタンスがエージェントの実行を再開でき、継続的な操作が保証されます。

新しいエージェント HTTP クライアントで sessionId と waitForResponse を使用します。 thread_id は非推奨の要求エイリアスであり、キャプチャする応答ヘッダーではありません。 PYTHON JSON 応答については、sessionIdとcorrelationIdを読み取ります。エイリアスと競合処理については、 HTTP API 移行ガイドを参照してください。

永続的な完了を確認する

生成された .NET Functions エージェントエンドポイントの場合、完了を次のように解釈します。

HTTP 状態	Meaning
`202 Accepted`	ディスパッチは受け入れられます。実行が成功したことが確認されません。
`200 OK`	コミットされた成功した結果を使用できます。
`500 Internal Server Error`	サポートされているコミット済みターミナルエラーが返されます。これは、通常のホストまたはトランスポートエラーと区別します。
`410 Gone`	完了はわかっていますが、結果は使用できません。保持される結果は、 `succeeded` または `failed`できます。これは必ずしも失敗した実行または結果の期限切れであるとは限りません。

完了に使用可能な結果がない場合に x-ms-durable-outcome を読み取る。 410を保留中として扱ったり、会話履歴の結果を推測したり、新しい ID で自動的に再送信したりしないでください。 JSON 呼び出し元は、文字列 status と数値 statusCodeを使用します。以前の status_code フィールドは一時的な互換性エイリアスです。

Python Functions では、使用可能な成功、失敗した結果の保持、および使用できない完了した結果も区別されます。エラーエンベロープと SDK オプションは、.NETと同じではありません。 .NET完了コントラクトと HTTP コントラクトPython参照してください。

これらの状態は、エージェントエンドポイントを表します。 Durable Functions オーケストレーション管理 API には、独自の状態応答とポーリングコントラクトがあります。

Reliable Streaming

Durable Extension は、永続的なエージェントの実行と共にリアルタイムのトークン配信を必要とするアプリケーションの信頼性の高いストリーミングをサポートします。ストリーミングは両方のホスティングモデルでコア拡張機能と共に使用できますが、分散ホストには Redis などの信頼性の高いストリームブローカーが必要です。そのため、クライアントは、ブローカーの永続化と保持の設定に従って、再接続または worker の変更後に保持されたトークンストリームの読み取りを再開できます。

ユーザーエクスペリエンスが増分応答に依存しているが、ワークロードに永続的な実行セマンティクスが必要な場合は、信頼性の高いストリーミングを使用します。ストリームの完了だけでは、エージェントの結果と状態が永続的に保持されたことを確認するわけではありません。権限のある永続的な結果を個別に確認します。実行可能な例については、信頼性の高いストリーミングパターンを含む Python Durable Task のサンプルを参照してください。

確定的なマルチエージェントオーケストレーション

Durable Extension では、Durable Task オーケストレーションを使用して複数のエージェントを調整する決定論的ワークフローの構築がサポートされています。 Azure Functionsでは、Durable Functions オーケストレーションが使用されます。独自のコンピューティングホストでは、構成した Durable Task worker とクライアントを介して実行されます。

オーケストレーション は、複数の操作 (エージェント呼び出し、外部 API 呼び出し、タイマーなど) を信頼できる方法で調整するコードベースのワークフローです。 決定論的 とは、障害が発生した後に再生されたときにオーケストレーションコードが同じように実行され、ワークフローの信頼性とデバッグ性が向上することを意味します。オーケストレーションの履歴を再生すると、各ステップで何が起こったかを正確に確認できます。

オーケストレーションは、エージェントの呼び出しの間に障害が発生する場合でも、安定して実行され、解除して続行できるように設計されており、予測可能で繰り返し可能なプロセスを提供します。これにより、実行順序とフォールトトレランスを保証する必要がある複雑なマルチエージェントシナリオに最適です。

永続的なチェックポイント処理では、ツールや外部書き込みがトランザクションまたは厳密に 1 回行われません。アクティビティとツールの副作用をべき等にし、再試行の境界を定義します。外部書き込み後、永続的コミットの前にエラーが発生すると、操作が再実行される可能性があります。

シーケンシャルオーケストレーション

順次マルチエージェントパターンでは、特殊化されたエージェントが特定の順序で実行され、各エージェントの出力が次のエージェントの実行に影響を与える可能性があります。このパターンでは、エージェントの応答に基づく条件付きロジックと分岐がサポートされます。

オーケストレーションでエージェントを使用する場合は、 context.GetAgent() API を使用して DurableAIAgent インスタンスを取得する必要があります。これは、登録済みエージェントのいずれかをラップする標準 AIAgent 型の特殊なサブクラスです。 DurableAIAgent ラッパーを使用すると、永続的オーケストレーションフレームワークによってエージェント呼び出しが適切に追跡され、チェックポイント処理されます。

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(SpamDetectionOrchestration))]
public static async Task<string> SpamDetectionOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    Email email = context.GetInput<Email>();

    // Check if the email is spam
    DurableAIAgent spamDetectionAgent = context.GetAgent("SpamDetectionAgent");
    AgentSession spamSession = await spamDetectionAgent.CreateSessionAsync();

    AgentResponse<DetectionResult> spamDetectionResponse = await spamDetectionAgent.RunAsync<DetectionResult>(
        message: $"Analyze this email for spam: {email.EmailContent}",
        session: spamSession);
    DetectionResult result = spamDetectionResponse.Result;

    if (result.IsSpam)
    {
        return await context.CallActivityAsync<string>(nameof(HandleSpamEmail), result.Reason);
    }

    // Generate response for legitimate email
    DurableAIAgent emailAssistantAgent = context.GetAgent("EmailAssistantAgent");
    AgentSession emailSession = await emailAssistantAgent.CreateSessionAsync();

    AgentResponse<EmailResponse> emailAssistantResponse = await emailAssistantAgent.RunAsync<EmailResponse>(
        message: $"Draft a professional response to: {email.EmailContent}",
        session: emailSession);

    return await context.CallActivityAsync<string>(nameof(SendEmail), emailAssistantResponse.Result.Response);
}

オーケストレーションでエージェントを使用する場合は、app.get_agent() メソッドを使用して持続可能なエージェントインスタンスを取得する必要があります。これは、登録済みエージェントの 1 つを対象とした特別なラッパーです。永続的なエージェントラッパーを使用すると、永続的オーケストレーションフレームワークによってエージェント呼び出しが適切に追跡され、チェックポイントが設定されます。

from durabletask.task import OrchestrationContext
from agent_framework_azurefunctions import AgentFunctionApp
from pydantic import BaseModel

class SpamDetectionResult(BaseModel):
    is_spam: bool
    reason: str

class EmailResponse(BaseModel):
    response: str

app = AgentFunctionApp(agents=[spam_detection_agent, email_assistant_agent])

@app.orchestration_trigger(context_name="context")
def spam_detection_orchestration(context: OrchestrationContext, email: dict):
    # Check if the email is spam
    spam_agent = app.get_agent(context, "SpamDetectionAgent")
    spam_thread = spam_agent.create_session()

    spam_result_raw = yield spam_agent.run(
        messages=f"Analyze this email for spam: {email['content']}",
        session=spam_thread,
        options={"response_format": SpamDetectionResult},
    )
    spam_result = SpamDetectionResult.model_validate(spam_result_raw.value)

    if spam_result.is_spam:
        result = yield context.call_activity("handle_spam_email", input=spam_result.reason)
        return result

    # Generate response for legitimate email
    email_agent = app.get_agent(context, "EmailAssistantAgent")
    email_thread = email_agent.create_session()

    email_response_raw = yield email_agent.run(
        messages=f"Draft a professional response to: {email['content']}",
        session=email_thread,
        options={"response_format": EmailResponse},
    )
    email_response = EmailResponse.model_validate(email_response_raw.value)

    result = yield context.call_activity("send_email", input=email_response.response)
    return result

オーケストレーションは、複数のエージェント間の作業を調整し、エージェント呼び出し間で発生する障害にも対応できるようにします。オーケストレーションコンテキストは、オーケストレーション内でホストされているエージェントを取得して操作するメソッドを提供します。

並列オーケストレーション

並列マルチエージェントパターンでは、複数のエージェントを同時に実行し、その結果を集計します。このパターンは、多様なパースペクティブを収集したり、独立したサブタスクを同時に処理したりするのに役立ちます。

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(ResearchOrchestration))]
public static async Task<string> ResearchOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string topic = context.GetInput<string>();

    // Execute multiple research agents in parallel
    DurableAIAgent technicalAgent = context.GetAgent("TechnicalResearchAgent");
    DurableAIAgent marketAgent = context.GetAgent("MarketResearchAgent");
    DurableAIAgent competitorAgent = context.GetAgent("CompetitorResearchAgent");

    // Start all agent runs concurrently
    Task<AgentResponse<TextResponse>> technicalTask = 
        technicalAgent.RunAsync<TextResponse>($"Research technical aspects of {topic}");
    Task<AgentResponse<TextResponse>> marketTask = 
        marketAgent.RunAsync<TextResponse>($"Research market trends for {topic}");
    Task<AgentResponse<TextResponse>> competitorTask = 
        competitorAgent.RunAsync<TextResponse>($"Research competitors in {topic}");

    // Wait for all tasks to complete
    await Task.WhenAll(technicalTask, marketTask, competitorTask);

    // Aggregate results
    string allResearch = string.Join("\n\n", 
        technicalTask.Result.Result.Text,
        marketTask.Result.Result.Text,
        competitorTask.Result.Result.Text);

    DurableAIAgent summaryAgent = context.GetAgent("SummaryAgent");
    AgentResponse<TextResponse> summaryResponse = 
        await summaryAgent.RunAsync<TextResponse>($"Summarize this research:\n{allResearch}");

    return summaryResponse.Result.Text;
}

from durabletask.task import OrchestrationContext, when_all
from agent_framework_azurefunctions import AgentFunctionApp

app = AgentFunctionApp(agents=[technical_agent, market_agent, competitor_agent, summary_agent])

@app.orchestration_trigger(context_name="context")
def research_orchestration(context: OrchestrationContext, topic: str):
    # Execute multiple research agents in parallel
    technical_agent = app.get_agent(context, "TechnicalResearchAgent")
    market_agent = app.get_agent(context, "MarketResearchAgent")
    competitor_agent = app.get_agent(context, "CompetitorResearchAgent")

    technical_task = technical_agent.run(messages=f"Research technical aspects of {topic}")
    market_task = market_agent.run(messages=f"Research market trends for {topic}")
    competitor_task = competitor_agent.run(messages=f"Research competitors in {topic}")

    # Wait for all tasks to complete
    results = yield when_all([technical_task, market_task, competitor_task])

    # Aggregate results
    all_research = "\n\n".join(r.text for r in results)

    summary_agent = app.get_agent(context, "SummaryAgent")
    summary = yield summary_agent.run(messages=f"Summarize this research:\n{all_research}")

    return summary.text

並列実行は、タスクの一覧を使用して追跡されます。自動チェックポイント処理により、集計中にエラーが発生した場合に、完了したエージェントの実行が繰り返されたり失われたりしないようにします。

Human-in-the-loop オーケストレーション

確定的なエージェントオーケストレーションは、コンピューティングリソースを消費することなく、人間の入力、承認、またはレビューのために一時停止できます。永続的な実行により、オーケストレーションは、人間の応答を待機している間、数日または数週間待機できます。サーバーレスホスティングと組み合わせると、待機期間中にすべてのコンピューティングリソースがスピンダウンされ、人間が入力を提供するまでコンピューティングコストが削減されます。

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(ContentApprovalWorkflow))]
public static async Task<string> ContentApprovalWorkflow(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string topic = context.GetInput<string>();

    // Generate content using an agent
    DurableAIAgent contentAgent = context.GetAgent("ContentGenerationAgent");
    AgentResponse<GeneratedContent> contentResponse = 
        await contentAgent.RunAsync<GeneratedContent>($"Write an article about {topic}");
    GeneratedContent draftContent = contentResponse.Result;

    // Send for human review
    await context.CallActivityAsync(nameof(NotifyReviewer), draftContent);

    // Wait for approval with timeout
    HumanApprovalResponse approvalResponse;
    try
    {
        approvalResponse = await context.WaitForExternalEvent<HumanApprovalResponse>(
            eventName: "ApprovalDecision",
            timeout: TimeSpan.FromHours(24));
    }
    catch (OperationCanceledException)
    {
        // Timeout occurred - escalate for review
        return await context.CallActivityAsync<string>(nameof(EscalateForReview), draftContent);
    }

    if (approvalResponse.Approved)
    {
        return await context.CallActivityAsync<string>(nameof(PublishContent), draftContent);
    }

    return "Content rejected";
}

from durabletask.task import OrchestrationContext, when_any
from datetime import timedelta
from agent_framework_azurefunctions import AgentFunctionApp

app = AgentFunctionApp(agents=[content_agent])

@app.orchestration_trigger(context_name="context")
def content_approval_workflow(context: OrchestrationContext, topic: str):
    # Generate content using an agent
    content_agent = app.get_agent(context, "ContentGenerationAgent")
    draft_content = yield content_agent.run(
        messages=f"Write an article about {topic}"
    )

    # Send for human review
    yield context.call_activity("notify_reviewer", input=draft_content.text)

    # Wait for approval with timeout
    approval_task = context.wait_for_external_event("ApprovalDecision")
    timeout_task = context.create_timer(
        context.current_utc_datetime + timedelta(hours=24)
    )

    winner = yield when_any([approval_task, timeout_task])

    if winner == approval_task:
        timeout_task.cancel()
        approval_data = approval_task.result
        if approval_data.get("approved"):
            result = yield context.call_activity("publish_content", input=draft_content.text)
            return result
        return "Content rejected"

    # Timeout occurred - escalate for review
    result = yield context.call_activity("escalate_for_review", input=draft_content.text)
    return result

確定的なエージェントオーケストレーションは、外部イベントを待ちながらその状態を耐久的に保持し、人間のフィードバックを待つ間も、障害や再起動、そして長期間の待機を乗り越えることができます。人間の応答が到着すると、オーケストレーションは完全な会話コンテキストと実行状態をそのまま使用して自動的に再開します。

人間の入力を提供する

待機中のオーケストレーションに承認または入力を送信するには、Durable Task クライアント SDK または Azure Functions Durable 拡張機能エンドポイントを使用して、オーケストレーションインスタンスに外部イベントを発生させます。たとえば、校閲者は、次を呼び出す Web フォームを使用してコンテンツを承認できます。

await client.RaiseEventAsync(instanceId, "ApprovalDecision", new HumanApprovalResponse 
{ 
    Approved = true,
    Feedback = "Looks great!"
});

approval_data = {
    "approved": True,
    "feedback": "Looks great!"
}
await client.raise_event(instance_id, "ApprovalDecision", approval_data)

コスト効率

永続的なエージェントを含む人間のループ内ワークフローは、 Azure Functions Flex 従量課金プランでホストされている場合、非常にコスト効率が高くなります。承認を 24 時間待機しているワークフローでは、数秒間の実行時間 (コンテンツの生成、通知の送信、応答の処理の時間) についてのみ支払います。24 時間の待機時間は支払われません。待機期間中は、コンピューティングリソースは消費されません。

Durable Task Scheduler を使用した可観測性

Durable Task Scheduler (DTS) は、優れたパフォーマンス、フルマネージドのインフラストラクチャ、および組み込みの可観測性を UI ダッシュボードで提供する、永続的なエージェントに推奨される永続的なバックエンドです。 Azure Functionsアプリは他のストレージバックエンド (Azure Storage など) を使用できますが、DTS は永続的なワークロード専用に最適化されており、優れたパフォーマンスと監視機能を提供します。セルフホステッドワーカーは、永続的なスケジュール設定、状態、ダッシュボードの可視性にも DTS を使用します。

エージェントセッションの分析情報

会話履歴: すべてのメッセージ、ツール呼び出し、会話コンテキストなど、各エージェントセッションの完全なチャット履歴を任意の時点で表示する
タスクのタイミング: 特定のタスクとエージェントの対話が完了するまでの時間を監視する

会話スレッドとメッセージを含むエージェントチャット履歴を示す Durable Task Scheduler ダッシュボードのスクリーンショット。

オーケストレーションの分析情報

マルチエージェントの視覚化: 並列実行と条件付き分岐の視覚的表現を使用して複数の特殊なエージェントを呼び出すときの実行フローを確認する
実行履歴: 詳細な実行ログにアクセスする
リアルタイム監視: デプロイ全体でアクティブなオーケストレーション、キューに置かれた作業項目、エージェントの状態を追跡する
パフォーマンスメトリック: エージェントの応答時間、トークンの使用状況、オーケストレーションの期間を監視する

複数のエージェント操作とワークフローの実行によるオーケストレーションの視覚化を示す Durable Task Scheduler ダッシュボードのスクリーンショット。

デバッグ機能

構造化されたエージェントの出力とツール呼び出しの結果を表示する
トレースツールの呼び出しとその結果
人間が介在するシナリオにおける外部イベント処理の監視

ダッシュボードを使用すると、エージェントの実行内容を正確に把握し、問題をすばやく診断し、実際の実行データに基づいてパフォーマンスを最適化できます。

デプロイとスキーマの移行

Schema-v2 エージェントの状態は、永続化された状態とディスパッチ ID の両方を変更します。状態変換とは別にワーカーの共存を計画します。

新しいセッションでは、 dafx-v2- エンティティ名前空間が使用されます。 2 フェーズロールアウト中は、v1 ワーカーがアクティブなまま ProcessLegacyEntities 無効のままにします。対象となるレガシエンティティを処理およびアップグレード ProcessLegacyEntities 有効にする前に、すべての v1 worker を停止します。

スタンドアロン worker は、作業項目フィルターを登録します。 Side-by-side Functions のデプロイには、参加しているすべてのホストでのホスト側フィルター処理に加えて、スケジューラストレージプロバイダー構成の workItemFilteringEnabled をサポートする、互換性のある Functions Worker、DurableTask worker 拡張機能、および AzureManaged worker 拡張機能パッケージも必要です。互換性のあるフィルター処理を使用しない場合は、1 つのハブ上の v1 ホストと v2 ホストを同時に使用するのではなく、オフラインカットオーバーを使用します。完全な構成については、.NET アップグレードガイドに従ってください。

ホストを構築する前に、明示的な戦略を選択します。

two_phase: 互換性のあるバックエンド作業項目のフィルター処理で Durable Task Scheduler を使用します。新しいバージョン管理された名前はディスパッチを分離しますが、対象となるレガシエンティティは古いワーカーに残ります。履歴名の処理とインプレースエンティティのアップグレードを有効にする前に、古いワーカーを停止します。共存中に互換性のあるフィルター処理されたワーカーを維持する。そのフリートが切断された場合、名前フィルターは永続的な予約ではありません。
isolated_v2: アップグレードされたワーカーとクライアントが一致する分離タスクハブを使用します。この設定は分離を確認しますが、他のワーカーのプローブは行いません。新しいワークフローインスタンスを使用します。

新しいハブは、2 フェーズエンティティ移行パスの無条件要件ではありません。ホスト固有のセットアップと前提条件については、Pythonエンティティの移行に関するページを参照してください。

Warning

エンティティの移行では、ワークフロー再生の互換性は修復されません。影響を受ける古いワークフロー履歴 (古い v2 履歴を含む) は、アップグレードされたランタイムではサポートされていません。完了する必要がある場合は、元のワーカーとハブに保持します。アップグレード後に新しいインスタンスを開始します。互換性を証明するためにワークフローバージョンマーカーだけに依存しないでください。ランタイムの互換性Python参照してください。

状態のライフサイクルと容量

各レイヤーのポリシーを個別に選択します。

レイヤー	制御する内容
セッション TTL	会話状態や完了確認メッセージを含むエンティティ全体を削除します。
永続的な履歴の保持	対象となる保存済みトランスクリプトコンテンツを削除します。保護された状態は引き続き容量を超える可能性があります。
モデル入力圧縮	モデルに送信されたコンテキストを変更します。それは必ずしも永続的な歴史を排除するとは限りません。
結果のリテンション期間	保持された完了結果とは別に、応答ペイロードの可用性を制御します。
BLOB ペイロードのライフサイクル	SDK オフロードペイロード BLOB を制御します。セッションの削除またはオーケストレーションの消去ではクリーンアップされません。

schema-v2 レシートを保持.NET状態では、未構成のレガシ 14 日間 TTL によってエンティティが自動的に削除されることはありません。アイドルセッションのクリーンアップが必要な場合は、グローバルまたはエージェントごとの TTL を明示的に構成します。 AddAIAgentにtimeToLive: nullを渡すと、構成済みのグローバル既定値が継承されます。7 日間のグローバル既定値は "無期限" でオーバーライドされません。DefaultTimeToLive = null設定すると、エージェントごとに正のオーバーライドを行わずに、エージェントのグローバル TTL が無効になります。

.NETの既定値は DurableAgentHistoryRetentionMode.KeepAll です。Autoにより、圧力ベースのトランスクリプトの保持が可能になります。 ResultRetentionPeriod は既定で 60 秒に設定され、セッション TTL とは別です。 Pythonには、既定でretention="keep_all"など、個別の保有期間と予算のオプションがあります。ホスト間で同じオプションまたは既定値を想定しないでください。

Pythonでは、両方のホスティングモデルでオプションの SDK マネージド BLOB ペイロードオフロードがサポートされます。プロデューサーストアとコンシューマーストア、資格情報、BLOB ライフサイクルの両方を構成します。オフロードでは、エンティティ全体の容量制限が削除されたり、リテンション期間が置き換えられたりすることはありません。参照された状態または再生履歴に必要な BLOB を保持します。 .NET状態とリテンション期間、Python保持と予算、Python Functions BLOB オフロードに関する説明を参照してください。

スケジューラのインフラストラクチャ、容量、運用に関するガイダンスについては、 Agent Framework の Durable Task 統合に関するトピックを参照してください。

チュートリアル: Azure Functionsを使用して非消耗品エージェントを作成して実行する

このチュートリアルでは、Durable Extension のAzure Functions ホスティングモデルを使用して、Durable AI エージェントを作成して実行する方法について説明します。組み込みの HTTP エンドポイントを使用してステートフルエージェントをホストする Azure Functions アプリを構築し、Durable Task Scheduler ダッシュボードを使用して監視する方法について説明します。セルフホステッドエージェントについては、サンプルを参照してください。

Prerequisites

開始する前に、次の前提条件を満たしていることを確認してください。

.NET 9.0 SDK 以降
Azure Functions Core Tools v4.x
Azure Developer CLI (azd)
Azure CLI のインストールと認証
Docker Desktop がインストールされ、実行されている (Azurite と Durable Task Scheduler エミュレーターを使用したローカル開発用)
リソースを作成するアクセス許可を持つ Azure サブスクリプション

Note

Microsoft Agent Framework は、アクティブにサポートされているすべてのバージョンの .NET でサポートされています。このサンプルでは、.NET 9 SDK 以降のバージョンをお勧めします。

Functions ホスティングPython 3.13 以降。スタンドアロンパッケージの最小Python 3.10 は、このチュートリアルには適用されません。
仮想環境を作成し、依存関係をインストールするための uv
Azure Functions Core Tools v4.x
Azure Developer CLI (azd)
Azure CLI のインストールと認証
Docker Desktop がインストールされ、実行されている (Azurite と Durable Task Scheduler エミュレーターを使用したローカル開発用)
リソースを作成するアクセス許可を持つ Azure サブスクリプション

チュートリアルハブが分離され、すべてのワーカーとクライアントがアップグレードされたランタイムを使用していることを確認した後にのみ、 DURABLE_AGENTS_DEPLOYMENT_MODE=isolated_v2 を設定します。

既存の永続的なアプリケーションとは別に、このチュートリアルでは新しいタスクハブを使用します。既存のアプリケーションの場合は、代わりにデプロイとスキーマの移行に従ってください。

クイックスタートプロジェクトをダウンロードする

Azure Developer CLI を使用して、Durable Agents クイックスタートテンプレートから新しいプロジェクトを初期化します。

プロジェクトの新しいディレクトリを作成し、そこに移動します。

Bash
PowerShell

mkdir MyDurableAgent
cd MyDurableAgent

New-Item -ItemType Directory -Path MyDurableAgent
Set-Location MyDurableAgent

テンプレートからプロジェクトを初期化します。
```
azd init --template durable-agents-quickstart-dotnet
```
環境名の入力を求められたら、 my-durable-agentなどの名前を入力します。

これにより、Azure Functions の構成、エージェントコード、コードテンプレートとしてのインフラストラクチャなど、必要なすべてのファイルを含むクイックスタートプロジェクトがダウンロードされます。

プロジェクトの新しいディレクトリを作成し、そこに移動します。

Bash
PowerShell

mkdir MyDurableAgent
cd MyDurableAgent

New-Item -ItemType Directory -Path MyDurableAgent
Set-Location MyDurableAgent

テンプレートからプロジェクトを初期化します。
```
azd init --template durable-agents-quickstart-python
```
環境名の入力を求められたら、 my-durable-agentなどの名前を入力します。

仮想環境を作成してアクティブ化する:

Bash
PowerShell

uv venv --python 3.13 .venv
source .venv/bin/activate

uv venv --python 3.13 .venv
.venv\Scripts\Activate.ps1

Note

インストールされている Python 3.13 インタープリターでpython -m venv .venvを使用することもできます。そのコマンドがWindowsでハングする場合は、代わりに上記の uv venv コマンドを使用します。パッケージをインストールする前に、 python --version でアクティブなインタープリターを確認します。

必要なパッケージをインストールします。

インストールする前に、agent-framework-openai クライアントパッケージを含め、「はじめに」の Functions パッケージにテンプレートのrequirements.txtを更新します。競合する依存関係ピンを強制的にインストールするのではなく、マニフェスト内の依存関係ピンを解決します。
```
uv pip install -r requirements.txt
```

これにより、Azure Functions の構成、エージェントコード、コードテンプレートとしてのインフラストラクチャなど、必要なすべてのファイルを含むクイックスタートプロジェクトがダウンロードされます。また、必要な依存関係を持つ仮想環境も準備します。

Azure リソースをプロビジョニングする

Azure Developer CLI を使用して、永続エージェントに必要な Azure リソースを作成します。

インフラストラクチャをプロビジョニングします。
```
azd provision
```
このコマンドを実行すると、次のものが作成されます。
- gpt-4o-mini デプロイを使用した Azure OpenAI サービス
- Flex Consumption ホスティングプランを使用した Azure Functions アプリ
- Azure Functions のランタイムと永続ストレージ用の Azure Storage アカウント
- エージェントの状態を管理するための Durable Task Scheduler インスタンス (従量課金プラン)
- 必要なネットワークと ID の構成
メッセージが表示されたら、Azure サブスクリプションを選択し、リソースの場所を選択します。

プロビジョニングプロセスには数分かかります。完了すると、azd によって作成されたリソース情報が環境内に格納されます。

エージェントコードを確認する

次に、Durable Agent を定義するコードを調べてみましょう。

Program.csを開いて、エージェントの構成を確認します。

using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") 
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT environment variable is not set");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") ?? "gpt-4o-mini";

// Create an AI agent following the standard Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are a helpful assistant that can answer questions and provide information.",
        name: "MyDurableAgent");

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options => options.AddAIAgent(agent))
    .Build();
app.Run();

このコード：

環境変数から Azure OpenAI 構成を取得します。
Azure 資格情報を使用して Azure OpenAI クライアントを作成します。
命令と名前を使用して AI エージェントを作成します。
永続的なスレッド管理を使用してエージェントをホストするように Azure Functions アプリを構成します。

function_app.pyを開いて、エージェントの構成を確認します。

import os
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT is not set.")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create an AI agent following the standard Microsoft Agent Framework pattern
agent = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
).as_agent(
    instructions="You are a helpful assistant that can answer questions and provide information.",
    name="MyDurableAgent"
)

# Configure the function app to host the agent with durable thread management
app = AgentFunctionApp(agents=[agent])

このコード：

環境変数から Azure OpenAI 構成を取得します。
Azure 資格情報を使用して Azure OpenAI クライアントを作成します。
命令と名前を使用して AI エージェントを作成します。
永続的なスレッド管理を使用してエージェントをホストするように Azure Functions アプリを構成します。

これで、エージェントを Azure Functions でホストする準備ができました。永続的タスク拡張機能は、エージェントと対話するための HTTP エンドポイントを自動的に作成し、複数の要求にわたって会話の状態を管理します。

ローカル設定を構成する

プロジェクトに含まれるサンプルファイルに基づいて、ローカル開発用の local.settings.json ファイルを作成します。

サンプル設定ファイルをコピーします。

Bash
PowerShell

cp local.settings.sample.json local.settings.json

Copy-Item local.settings.sample.json local.settings.json

プロビジョニングされたリソースから Azure OpenAI エンドポイントを取得します。
```
azd env get-value AZURE_OPENAI_ENDPOINT
```
local.settings.jsonを開き、<your-resource-name>値のAZURE_OPENAI_ENDPOINTを前のコマンドのエンドポイントに置き換えます。

local.settings.jsonは次のようになります。

{
  "IsEncrypted": false,
  "Values": {
    // ... other settings ...
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "AZURE_OPENAI_ENDPOINT": "https://your-openai-resource.openai.azure.com",
    "AZURE_OPENAI_DEPLOYMENT": "gpt-4o-mini",
    "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
    "TASKHUB_NAME": "default"
  }
}

Note

local.settings.json ファイルはローカル開発にのみ使用され、Azure にはデプロイされません。運用環境のデプロイの場合、これらの設定はインフラストラクチャテンプレートによって Azure Functions アプリで自動的に構成されます。

Pythonの場合は、チュートリアルハブが分離されていることを確認した後にのみ、Valuesに"DURABLE_AGENTS_DEPLOYMENT_MODE": "isolated_v2"を追加します。配置時に同等のアプリケーション設定を構成します。ローカル設定は自動的にアップロードされません。

スケジューラベースのホスティングの場合は、 host.json ( %TASKHUB_NAME%)、 TASKHUB_NAME、 DURABLE_TASK_SCHEDULER_CONNECTION_STRING、エミュレーター、ダッシュボードで同じタスクハブを使用します。このチュートリアルでは、エミュレーターによって自動的に作成される defaultを使用します。そのハブに既に古い履歴が含まれている場合、または他のワーカーによって使用されている場合は、一意の未使用のハブ名を選択し、エミュレーターの DTS_TASK_HUB_NAMES をその名前に設定し、アプリケーション設定とダッシュボードの選択を一貫して更新します。スケジューラ接続文字列を定義するだけでは、スケジューラストレージプロバイダーは選択されません。

host.jsonで、テンプレートの他のホスト設定を保持して、extensions.durableTaskでスケジューラプロバイダーを構成します。

{
  "extensions": {
    "durableTask": {
      "hubName": "%TASKHUB_NAME%",
      "storageProvider": {
        "type": "AzureManaged",
        "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
      }
    }
  }
}

ホスティング言語にスケジューラプロバイダーの依存関係を使用します。この構成では、新しい分離されたチュートリアルデプロイのバックエンドが選択されます。サイドバイサイドアップグレードに必要なフィルター処理は有効になりません。

ローカル開発の依存関係を開始する

永続エージェントをローカルで実行するには、次の 2 つのサービスを開始する必要があります。

Azurite: Azure Storage サービスをエミュレートします (トリガーと内部状態を管理するために Azure Functions によって使用されます)。
Durable Task Scheduler (DTS) エミュレーター: エージェントの永続的な状態 (会話履歴、オーケストレーション状態) とスケジュールを管理します

これらの手順では、Functions ホストで Durable Task Scheduler ストレージプロバイダーが使用されていることを前提としています。リリースの Python Functions サンプルは、既定でAzure Storageされます。これらのサンプルから開始する場合は、DTS ダッシュボードを使用する前にスケジューラプロバイダーを選択して構成します。それ以外の場合、Azurite は永続的な状態を保持し、スケジューラダッシュボードは空のままです。 Durable Task Scheduler のセットアップを参照してください。

Azurite を起動する

Azurite は、Azure Storage サービスをローカルでエミュレートします。 Azure Functions では、内部状態の管理に使用されます。これは、新しいターミナルウィンドウで実行し、Durable Agent の開発とテスト中に実行し続ける必要があります。

新しいターミナルウィンドウを開き、Azurite Docker イメージをプルします。
```
docker pull mcr.microsoft.com/azure-storage/azurite
```
ターミナルウィンドウで Azurite を起動します。
```
docker run -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite
```
Azurite は、BLOB (10000)、キュー (10001)、テーブル (10002) サービスの既定ポートで起動し、リッスンを開始します。

Durable Agent の開発とテスト中は、このターミナルウィンドウを開いたままにしておきます。

Tip

別のインストール方法を含む Azurite の詳細については、「 Azure Storage のローカル開発に Azurite エミュレーターを使用する」を参照してください。

Durable Task Scheduler エミュレーターを起動する

DTS エミュレーターは、エージェントの状態とオーケストレーションを管理するための永続的なバックエンドを提供します。会話履歴が格納され、再起動後もエージェントの状態が維持されます。また、永続性のあるオーケストレーションとエージェントもトリガーします。これは、別の新しいターミナルウィンドウで実行し、Durable Agent の開発とテスト中に実行し続ける必要があります。

別の新しいターミナルウィンドウを開き、DTS エミュレーターの Docker イメージをプルします。
```
docker pull mcr.microsoft.com/dts/dts-emulator:latest
```
DTS エミュレーターを実行します。
```
docker run -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
```
このコマンドは、 default タスクハブを作成し、このチュートリアルの設定に一致するエミュレーターを起動します。このコマンドは次を公開します。
- ポート 8080: Durable Task Scheduler の gRPC エンドポイント (Functions アプリで使用)
- ポート 8082: 管理ダッシュボード
ダッシュボードは、 http://localhost:8082で使用できるようになります。

Durable Agent の開発とテスト中は、このターミナルウィンドウを開いたままにしておきます。

Tip

複数のタスクハブを構成してダッシュボードにアクセスする方法など、DTS エミュレーターの詳細については、「 Durable Task Scheduler を使用した開発」を参照してください。

関数アプリを実行する

これで、Durable Agent を使用して Azure Functions アプリを実行する準備ができました。

新しいターミナルウィンドウ (Azurite と DTS エミュレーターの両方を別々のウィンドウで実行したまま) で、プロジェクトディレクトリに移動します。
Azure Functions ランタイムを開始します。
```
func start
```
関数アプリが実行されていることを示す出力 (エージェントの HTTP エンドポイントを含む) が表示されます。
```
Functions:
     http-MyDurableAgent: [POST] http://localhost:7071/api/agents/MyDurableAgent/run
     dafx-v2-MyDurableAgent: entityTrigger
```

これは、schema-v2 の登録.NET示しています。トリガー名Pythonは、デプロイ戦略によって異なります。これらのエンドポイントは会話状態を自動的に管理するため、セッションオブジェクトを自分で作成する必要はありません。

エージェントをローカルでテストする

HTTP 要求を使用して永続エージェントと対話できるようになりました。エージェントは、複数の要求にわたって会話状態を維持し、複数ターンの会話を有効にします。

会話を開始する

新しいセッションを作成し、最初のメッセージを送信します。

Bash
PowerShell

curl -i -X POST http://localhost:7071/api/agents/MyDurableAgent/run \
  -H "Content-Type: text/plain" \
  -d "What are three popular programming languages?"

$response = Invoke-WebRequest -Uri "http://localhost:7071/api/agents/MyDurableAgent/run" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "What are three popular programming languages?"
$response.Headers
$response.Content

応答のサンプル (x-ms-session-id ヘッダーにはセッション ID が含まれています。この例では、.NET schema-v2 ID を示します)。

HTTP/1.1 200 OK
Content-Type: text/plain
x-ms-session-id: @dafx-v2-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d
Content-Length: 189

Three popular programming languages are Python, JavaScript, and Java. Python is known for its simplicity and readability, JavaScript powers web interactivity, and Java is widely used in enterprise applications.

次の要求の x-ms-session-id ヘッダーの正確なセッション ID を保存します。構築したり書き直したりしないでください。 Pythonもこのヘッダーを返しますが、ID 形式が異なる場合があります。

会話を続ける

返された ID を正規の sessionId クエリパラメーターに含めることで、同じセッションにフォローアップメッセージを送信します。

Bash
PowerShell

例の ID を、前の応答の x-ms-session-id ヘッダーの実際の値に置き換えます。必要に応じて値を URL エンコードします。

curl -X POST "http://localhost:7071/api/agents/MyDurableAgent/run?sessionId=@dafx-v2-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d" \
  -H "Content-Type: text/plain" \
  -d "Which one is best for beginners?"

前の応答の x-ms-session-id ヘッダーで返されたセッション ID を読み取ってエンコードします。

$sessionId = $response.Headers["x-ms-session-id"]
$encodedSessionId = [Uri]::EscapeDataString($sessionId)
Invoke-RestMethod -Uri "http://localhost:7071/api/agents/MyDurableAgent/run?sessionId=$encodedSessionId" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "Which one is best for beginners?"

応答の例:

Python is often considered the best choice for beginners among those three. Its clean syntax reads almost like English, making it easier to learn programming concepts without getting overwhelmed by complex syntax. It's also versatile and widely used in education.

エージェントは、前のメッセージ (3 つのプログラミング言語) のコンテキストを記憶していることに注意してください。もう一度指定する必要はありません。会話の状態は Durable Task Scheduler によって永続的に保存されるため、関数アプリを再起動した場合や、別のインスタンスによって会話が再開された場合でも、この履歴は保持されます。

Durable Task Scheduler ダッシュボードを使用して監視する

Durable Task Scheduler には、永続的なエージェントを監視およびデバッグするための組み込みのダッシュボードが用意されています。ダッシュボードでは、エージェントの操作、会話履歴、実行フローを詳細に把握できます。

ダッシュボードにアクセスする

Web ブラウザーの http://localhost:8082 で、ローカル DTS エミュレーターのダッシュボードを開きます。
既定またはアプリケーションとエミュレーターの両方で構成した同じ一意のハブ名を選択して、その詳細を表示します。
右上隅にある歯車アイコンを選択して設定を開き、[プレビュー機能] の [エージェントページを有効にする] オプションが選択されていることを確認します。

エージェントの会話を探索する

ダッシュボードで、[ エージェント ] タブに移動します。
一覧から永続的なエージェントスレッド ( mydurableagent - 263fa373-fa01-4705-abf2-5a114c2bb87d など) を選択します。

すべてのメッセージと応答を含む完全な会話履歴など、エージェントスレッドの詳細なビューが表示されます。

ダッシュボードには、会話のフローを理解するのに役立つタイムラインビューが用意されています。主な情報は次のとおりです。

各操作のタイムスタンプと期間
プロンプトと応答のコンテンツ
使用されたトークンの数

Tip

DTS ダッシュボードにはリアルタイムの更新プログラムが用意されているため、HTTP エンドポイントを介してエージェントを操作するときにエージェントの動作を確認できます。

Azure にデプロイ

Durable Agent をローカルでテストしたので、Azure にデプロイします。

アプリケーションをデプロイします。
```
azd deploy
```
このコマンドは、アプリケーションをパッケージ化し、プロビジョニング中に作成された Azure Functions アプリにデプロイします。
デプロイが完了するまで待ちます。出力は、エージェントが Azure で実行されていることを確認します。

デプロイされたエージェントをテストする

デプロイ後、Azure で実行されているエージェントをテストします。

関数キーを取得する

生成されたエンドポイントの既定の関数レベル承認の場合は、Azureへの要求に Functions キーを含めます。 Functions キーはアプリケーションユーザー承認ではありません。運用環境のセキュリティとトラブルシューティングを参照してください。

Bash
PowerShell

API_KEY=`az functionapp function keys list --name $(azd env get-value AZURE_FUNCTION_NAME) --resource-group $(azd env get-value AZURE_RESOURCE_GROUP) --function-name http-MyDurableAgent --query default -o tsv`

$functionName = azd env get-value AZURE_FUNCTION_NAME
$resourceGroup = azd env get-value AZURE_RESOURCE_GROUP
$API_KEY = az functionapp function keys list --name $functionName --resource-group $resourceGroup --function-name http-MyDurableAgent --query default -o tsv

Azure で新しい会話を開始する

新しいセッションを作成し、デプロイされたエージェントに最初のメッセージを送信します。

Bash
PowerShell

curl -i -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" \
  -H "Content-Type: text/plain" \
  -d "What are three popular programming languages?"

$functionName = azd env get-value AZURE_FUNCTION_NAME
$response = Invoke-WebRequest -Uri "https://$functionName.azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "What are three popular programming languages?"
$response.Headers
$response.Content

x-ms-session-id応答ヘッダーに返されたセッション ID を保存します。

Azure で会話を続ける

同じセッションでフォローアップメッセージを送信します。前の応答によって返された正確な ID を使用します。

Bash
PowerShell

SESSION_ID="<session-id>"
curl -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" \
  --url-query "sessionId=$SESSION_ID" \
  -H "Content-Type: text/plain" \
  -d "Which is easiest to learn?"

$sessionId = $response.Headers["x-ms-session-id"]
$encodedSessionId = [Uri]::EscapeDataString($sessionId)
$functionName = azd env get-value AZURE_FUNCTION_NAME
Invoke-RestMethod -Uri "https://$functionName.azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY&sessionId=$encodedSessionId" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "Which is easiest to learn?"

エージェントは、ローカルで行ったのと同様に、Azure で会話コンテキストを維持し、エージェントの状態の持続性を示します。

デプロイされたエージェントを監視する

Azure の Durable Task Scheduler ダッシュボードを使用して、デプロイされたエージェントを監視できます。

Durable Task Scheduler インスタンスの名前を取得します。
```
azd env get-value DTS_NAME
```
Azure portal を開き、前の手順の Durable Task Scheduler 名を検索します。
Durable Task Scheduler リソースの [概要] ブレードで、デプロイされたアプリケーション用に構成されたタスクハブを選択します。
タスクハブページの上部にある [ ダッシュボードを開く ] を選択して、監視ダッシュボードを開きます。
ローカルエミュレーターで行ったのと同じように、エージェントの会話を表示します。

Azure でホストされるダッシュボードには、ローカルエミュレーターと同じデバッグおよび監視機能が用意されており、会話履歴の検査、ツール呼び出しのトレース、運用環境でのパフォーマンスの分析を行うことができます。

チュートリアル: Azure Functionsを使用して非消耗品エージェントを調整する

このチュートリアルでは、Azure Functions ホスティングモデルとファンアウト/ファンインパターンを使用して、複数の永続的な AI エージェントを調整する方法について説明します。前のチュートリアルの非消耗品エージェントを拡張して、ユーザーの質問を処理するマルチエージェントシステムを作成し、応答を複数の言語に同時に変換します。セルフホステッドオーケストレーションの例については、サンプルを参照してください。

オーケストレーションパターンについて

ビルドするオーケストレーションは、次のフローに従います。

ユーザー入力 - ユーザー からの質問またはメッセージ
メインエージェント - 最初のチュートリアルの MyDurableAgent が質問を処理する
ファンアウト - メインエージェントの応答が両方の翻訳エージェントに同時に送信される
翻訳エージェント - 2 つの特殊なエージェントが応答を翻訳します (フランス語とスペイン語)
ファンイン - 結果は、元の応答と翻訳を使用して 1 つの JSON 応答に集計されます

このパターンにより、同時処理が可能になり、連続変換と比較して合計応答時間が短縮されます。

起動時にエージェントを登録する

永続的オーケストレーションでエージェントを適切に使用するには、アプリケーションの起動時に登録します。これらは、オーケストレーションの実行全体で使用できます。

既存のProgram.csと共に翻訳エージェントを登録するようにMyDurableAgentを更新します。

using System;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

// Get the Azure OpenAI configuration
string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT")
    ?? "gpt-4o-mini";

// Create the Microsoft Foundry client
AIProjectClient client = new(new Uri(endpoint), new DefaultAzureCredential());

// Create the main agent from the first tutorial
AIAgent mainAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a helpful assistant that can answer questions and provide information.",
    name: "MyDurableAgent");

// Create translation agents
AIAgent frenchAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a translator. Translate the following text to French. Return only the translation, no explanations.",
    name: "FrenchTranslator");

AIAgent spanishAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a translator. Translate the following text to Spanish. Return only the translation, no explanations.",
    name: "SpanishTranslator");

// Build and configure the Functions host
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options =>
    {
        // Register all agents for use in orchestrations and HTTP endpoints
        options.AddAIAgent(mainAgent);
        options.AddAIAgent(frenchAgent);
        options.AddAIAgent(spanishAgent);
    })
    .Build();

app.Run();

既存のfunction_app.pyと共に翻訳エージェントを登録するようにMyDurableAgentを更新します。

import os
from azure.identity import DefaultAzureCredential
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient

# Get the Azure OpenAI configuration
endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT is not set.")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create the Azure OpenAI client
chat_client = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
)

# Create the main agent from the first tutorial
main_agent = chat_client.as_agent(
    instructions="You are a helpful assistant that can answer questions and provide information.",
    name="MyDurableAgent"
)

# Create translation agents
french_agent = chat_client.as_agent(
    instructions="You are a translator. Translate the following text to French. Return only the translation, no explanations.",
    name="FrenchTranslator"
)

spanish_agent = chat_client.as_agent(
    instructions="You are a translator. Translate the following text to Spanish. Return only the translation, no explanations.",
    name="SpanishTranslator"
)

# Create the function app and register all agents
app = AgentFunctionApp(agents=[main_agent, french_agent, spanish_agent])

オーケストレーション関数を作成する

オーケストレーション関数は、複数のエージェント間でワークフローを調整します。永続コンテキストから登録済みエージェントを取得し、その実行を調整し、最初にメインエージェントを呼び出し、次に翻訳エージェントに同時にファンアウトします。

プロジェクトディレクトリに AgentOrchestration.cs という名前の新しいファイルを作成します。

using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

namespace MyDurableAgent;

public static class AgentOrchestration
{
    // Define a strongly-typed response structure for agent outputs
    public sealed record TextResponse(string Text);

    [Function("agent_orchestration_workflow")]
    public static async Task<Dictionary<string, string>> AgentOrchestrationWorkflow(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        var input = context.GetInput<string>() ?? throw new ArgumentNullException(nameof(context), "Input cannot be null");

        // Step 1: Get the main agent's response
        DurableAIAgent mainAgent = context.GetAgent("MyDurableAgent");
        AgentResponse<TextResponse> mainResponse = await mainAgent.RunAsync<TextResponse>(input);
        string agentResponse = mainResponse.Result.Text;

        // Step 2: Fan out - get the translation agents and run them concurrently
        DurableAIAgent frenchAgent = context.GetAgent("FrenchTranslator");
        DurableAIAgent spanishAgent = context.GetAgent("SpanishTranslator");

        Task<AgentResponse<TextResponse>> frenchTask = frenchAgent.RunAsync<TextResponse>(agentResponse);
        Task<AgentResponse<TextResponse>> spanishTask = spanishAgent.RunAsync<TextResponse>(agentResponse);

        // Step 3: Wait for both translation tasks to complete (fan-in)
        await Task.WhenAll(frenchTask, spanishTask);

        // Get the translation results
        TextResponse frenchResponse = (await frenchTask).Result;
        TextResponse spanishResponse = (await spanishTask).Result;

        // Step 4: Combine results into a dictionary
        var result = new Dictionary<string, string>
        {
            ["original"] = agentResponse,
            ["french"] = frenchResponse.Text,
            ["spanish"] = spanishResponse.Text
        };

        return result;
    }
}

オーケストレーション関数を function_app.py ファイルに追加します。

from durabletask.task import OrchestrationContext, when_all

@app.orchestration_trigger(context_name="context")
def agent_orchestration_workflow(context: OrchestrationContext, input_text: str):
    """
    Orchestration function that coordinates multiple agents.
    Returns a dictionary with the original response and translations.
    """
    # Step 1: Get the main agent's response
    main_agent = app.get_agent(context, "MyDurableAgent")
    main_response = yield main_agent.run(input_text)
    agent_response = main_response.text

    # Step 2: Fan out - get the translation agents and run them concurrently
    french_agent = app.get_agent(context, "FrenchTranslator")
    spanish_agent = app.get_agent(context, "SpanishTranslator")

    parallel_tasks = [
        french_agent.run(agent_response),
        spanish_agent.run(agent_response)
    ]

    # Step 3: Wait for both translation tasks to complete (fan-in)
    translations = yield when_all(parallel_tasks)

    # Step 4: Combine results into a dictionary
    result = {
        "original": agent_response,
        "french": translations[0].text,
        "spanish": translations[1].text
    }

    return result

Note

オーケストレーションをテストする

最初のチュートリアルのローカル開発の依存関係が引き続き実行されていることを確認します。

1 つのターミナルウィンドウで Azurite
別のターミナルウィンドウの Durable Task Scheduler エミュレーター

ローカル開発の依存関係が実行されている場合:

新しいターミナルウィンドウで Azure Functions アプリを起動します。
```
func start
```

Durable Functions 拡張機能は、オーケストレーションを管理するための組み込みの HTTP エンドポイントを自動的に作成します。組み込みの API を使用してオーケストレーションを開始します。

Bash
PowerShell

curl -X POST http://localhost:7071/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow \
  -H "Content-Type: application/json" \
  -d '"\"What are three popular programming languages?\""'

$body = '"What are three popular programming languages?"'
Invoke-RestMethod -Method Post -Uri "http://localhost:7071/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow" `
  -ContentType "application/json" `
  -Body $body

応答には、オーケストレーションインスタンスを管理するための URL が含まれています。

{
  "id": "abc123def456",
  "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456",
  "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456/raiseEvent/{eventName}",
  "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456/terminate",
  "purgeHistoryDeleteUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456"
}

statusQueryGetUriを使用してオーケストレーションの状態を照会します (abc123def456を実際のインスタンス ID に置き換えます)。

Bash
PowerShell

curl http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456

Invoke-RestMethod -Uri "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456"

runtimeStatusがCompletedされるまで、状態エンドポイントをポーリングします。完了すると、メインエージェントの応答とその翻訳を含むオーケストレーション出力が表示されます。

{
  "name": "agent_orchestration_workflow",
  "instanceId": "abc123def456",
  "runtimeStatus": "Completed",
  "output": {
    "original": "Three popular programming languages are Python, JavaScript, and Java. Python is known for its simplicity...",
    "french": "Trois langages de programmation populaires sont Python, JavaScript et Java. Python est connu pour sa simplicité...",
    "spanish": "Tres lenguajes de programación populares son Python, JavaScript y Java. Python es conocido por su simplicidad..."
  }
}

ダッシュボードでオーケストレーションを監視する

Durable Task Scheduler ダッシュボードでは、オーケストレーションに対する可視性が提供されます。

ブラウザーで http://localhost:8082 を開きます。
アプリケーション用に構成されたタスクハブを選択します。
[オーケストレーション] タブを選択します。
一覧でオーケストレーションインスタンスを見つけます。
表示するインスタンスを選択します。
- オーケストレーションのタイムライン
- メインエージェントの実行後に同時翻訳エージェントが続く
- 各エージェントの実行順序（MyDurableAgent、続いてフランス語とスペイン語の翻訳プログラム）
- ファンアウトパターンとファンインパターンを視覚化
- 各ステップのタイミングと期間

オーケストレーションを Azure にデプロイする

Azure Developer CLI を使用して更新されたアプリケーションをデプロイします。

azd deploy

これにより、新しいオーケストレーション関数と追加のエージェントを含む更新されたコードが、最初のチュートリアルで作成された Azure Functions アプリにデプロイされます。

デプロイされたオーケストレーションをテストする

デプロイ後、Azure で実行されているオーケストレーションをテストします。

永続的拡張機能のシステムキーを取得します。

Bash
PowerShell

SYSTEM_KEY=$(az functionapp keys list --name $(azd env get-value AZURE_FUNCTION_NAME) --resource-group $(azd env get-value AZURE_RESOURCE_GROUP) --query "systemKeys.durabletask_extension" -o tsv)

$functionName = azd env get-value AZURE_FUNCTION_NAME
$resourceGroup = azd env get-value AZURE_RESOURCE_GROUP
$SYSTEM_KEY = (az functionapp keys list --name $functionName --resource-group $resourceGroup --query "systemKeys.durabletask_extension" -o tsv)

組み込みの API を使用してオーケストレーションを開始します。

Bash
PowerShell

curl -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow?code=$SYSTEM_KEY" \
  -H "Content-Type: application/json" \
  -d '"\"What are three popular programming languages?\""'

$functionName = azd env get-value AZURE_FUNCTION_NAME
$body = '"What are three popular programming languages?"'
Invoke-RestMethod -Method Post -Uri "https://$functionName.azurewebsites.net/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow?code=$SYSTEM_KEY" `
  -ContentType "application/json" `
  -Body $body

応答の statusQueryGetUri を用いて結果の完了を定期的に確認し、翻訳結果を表示します。

運用環境のセキュリティとトラブルシューティング

これらのアクセス境界を個別に扱います。

アプリケーションユーザー: 呼び出し元を認証し、各セッション、ワークフロー、および人間承認応答へのアクセスを承認します。セッション ID、関連付け ID、および要求 ID は、資格情報ではなく識別子です。
Functions エンドポイント: 関数レベルのキーは、生成された HTTP エンドポイントを保護しますが、アプリケーションユーザー ID またはセッションの所有権を確立しません。キーの保護とローテーション。ブラウザーで提供されるアプリケーションコードで公開しないでください。
モデルアクセス: worker ID に必要なモデルのアクセス許可を付与します。モデルの資格情報では、スケジューラまたは別のユーザーのセッションへのアクセスは許可されません。
スケジューラアクセス: 目的のロールに対して worker、クライアント、ダッシュボードのアクセス許可を構成します。インフラストラクチャのガイダンスについては、 Durable Task の統合に関するトピックを参照してください。

症状:	チェック
Python パッケージのインストールまたは Functions の起動が失敗する	Functions には Python 3.13 以降を使用し、「はじめに」で Functions パッケージをインストールし、デプロイ戦略を明示的に選択します。スタンドアロンPython最小値は異なります。
要求は成功しますが、DTS ダッシュボードは空です	Functions ストレージプロバイダーが`AzureStorage`ではなく`AzureManaged`されていること、およびスケジューラエンドポイントとタスクハブがダッシュボードと一致することを確認します。
セッションがコンテキストを失っているように見える	返された正確な `x-ms-session-id` 値を `sessionId`と共に再利用します。別のハブ、デプロイ名前空間、または構成されたセッションのクリーンアップを確認します。
アップグレード後に古いワークフローが失敗する	履歴実行を再開する前に、再生の互換性を確認します。エンティティの移行では、互換性のないワークフロー履歴は修復されません。
ストリームは終了しますが、永続的な完了がありません	ストリーム出力を暫定的として扱い、権限のある結果または保持された結果を確認します。ストリーム完了マーカーは永続的なコミット確認ではありません。
完了した操作が返される `410`	保持された結果と結果の可用性を調べます。失敗を想定したり、保留中のようにポーリングを続けたり、新しい操作を自動的に送信したりしないでください。
エンティティ状態の書き込みが容量を超えている	エンティティ全体の状態、保護されたレコード、保持ポリシー、ペイロードの参照を検査します。モデル入力圧縮だけでは、格納された状態が収まることは保証されません。

次のステップ

OpenAI互換性のあるエンドポイント

その他のリソース:

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-11-06

耐久性の高い拡張機能

概要

耐久性のあるエージェントを使用する時

ホスティング モデルを選択する

作業の開始

Azure Functions ホスティング

独自のコンピューティングの持ち込み/セルフホステッド ホスティング

Durable Agent Framework ワークフロー

Samples

会話履歴を含むステートフル エージェント スレッド

永続的な完了を確認する

Reliable Streaming

確定的なマルチエージェント オーケストレーション

シーケンシャル オーケストレーション

並列オーケストレーション

Human-in-the-loop オーケストレーション

人間の入力を提供する

コスト効率

Durable Task Scheduler を使用した可観測性

エージェント セッションの分析情報

オーケストレーションの分析情報

デバッグ機能

デプロイとスキーマの移行

状態のライフサイクルと容量

チュートリアル: Azure Functionsを使用して非消耗品エージェントを作成して実行する

Prerequisites

クイック スタート プロジェクトをダウンロードする

Azure リソースをプロビジョニングする

エージェント コードを確認する

ローカル設定を構成する

ローカル開発の依存関係を開始する

Azurite を起動する

Durable Task Scheduler エミュレーターを起動する

関数アプリを実行する

エージェントをローカルでテストする

会話を開始する

会話を続ける

Durable Task Scheduler ダッシュボードを使用して監視する

ダッシュボードにアクセスする

エージェントの会話を探索する

Azure にデプロイ

デプロイされたエージェントをテストする

関数キーを取得する

Azure で新しい会話を開始する

Azure で会話を続ける

デプロイされたエージェントを監視する

チュートリアル: Azure Functionsを使用して非消耗品エージェントを調整する

オーケストレーション パターンについて

起動時にエージェントを登録する

オーケストレーション関数を作成する

オーケストレーションをテストする

ダッシュボードでオーケストレーションを監視する

オーケストレーションを Azure にデプロイする

デプロイされたオーケストレーションをテストする

運用環境のセキュリティとトラブルシューティング

次のステップ

フィードバック

その他のリソース

ホスティングモデルを選択する

独自のコンピューティングの持ち込み/セルフホステッドホスティング

会話履歴を含むステートフルエージェントスレッド

確定的なマルチエージェントオーケストレーション

シーケンシャルオーケストレーション

エージェントセッションの分析情報

クイックスタートプロジェクトをダウンロードする

エージェントコードを確認する

オーケストレーションパターンについて