Python 用 Azure Communication JobRouter パッケージ クライアント ライブラリ - バージョン 1.0.0
このパッケージには、JobRouter 用の python SDK for Azure Communication Servicesが含まれています。 Azure Communication Servicesの詳細については、こちらを参照してください
ソースコード | パッケージ (Pypi) | 製品ドキュメント
免責事項
Python 2.7 の Azure SDK Python パッケージのサポートは、2022 年 1 月 1 日に終了しました。 詳細と質問については、https://github.com/Azure/azure-sdk-for-python/issues/20691 を参照してください
作業の開始
前提条件
このパッケージを使用するには、 Azure サブスクリプション と Communication Service リソース が必要です。
- このパッケージを使用するには、Python 3.7 以降が必要です。 詳細については、 Azure SDK for Python バージョンのサポート ポリシーに関するページを参照してください。
- 新しい Communication Service を作成するには、Azure Portal の Azure PowerShell
パッケージをインストールする
pip を使用して Python 用 Azure Communication JobRouter クライアント ライブラリをインストールします。
pip install azure-communication-jobrouter
主要な概念
ジョブ
ジョブは、使用できる worker にルーティングする必要のある作業単位を表します。 この実際の例としては、コール センターのコンテキストでの着信通話やチャットがあります。
ワーカー
Worker は、ジョブを処理するために使用できる供給を表します。 各ワーカーは、ジョブを受信するために複数のキューに登録します。 この実際の例としては、コール センターで作業しているエージェントが挙げられる場合があります。
キュー
Queue は、ワーカーによる処理を待機しているジョブの順序付きリストを表します。 ワーカーはキューに登録して、そこから作業を受け取ります。 実際の例として、コール センターのコール キューがあります。
チャネル
チャネルは、何らかの種類のジョブのグループ化を表します。 ワーカーが作業を受信するように登録するときは、作業を処理できるチャネルと、それぞれが同時に処理できるチャネルの数も指定する必要があります。
実際voice callschatsの例として、またはコール センターがあります。
プラン
オファーは JobRouter によってワーカーに拡張され、一致が判断されたときに特定のジョブを処理します。この通知は通常、 EventGrid を介して配信されます。 ワーカーは JobRouter API を使用してオファーを受け入れるか拒否するか、配布ポリシーで構成された有効期間に従って期限切れになります。 実際の例として、コール センターでのエージェントの呼び出しが挙げられる場合があります。
配布ポリシー
配布ポリシーは、キュー内のジョブをそのキューに登録されているワーカーに配布する方法を制御する構成セットを表します。 この構成には、オファーの有効期限が切れるまでの期間と、複数の利用可能な場合にワーカーを選択する順序を定義する配布モードが含まれます。
配布モード
3 種類のモードは次のとおりです。
- ラウンド ロビン: ワーカーは
Idの順に並べられ、オファーを受けた前のワーカーの次のワーカーが選ばれます。 - 最も長いアイドル: 仕事に関して最も長く働いていないワーカー。
- Best Worker: 2 人のワーカーを比較して、選択するワーカーを決定する式を指定できます。
ラベル
ワーカー、ジョブ、キューにラベルを添付できます。 これらは、、または boolean データ型のキー値のstringnumberペアです。
実際の例としては、特定のワーカーのスキル レベル、チーム、地理的な場所があります。
ラベル セレクター
ラベル セレクターは、キューにサービスを提供するワーカーのサブセットをターゲットにするために、ジョブにアタッチできます。 この実際の例は、エージェントが特定の製品に関する最低限の知識レベルを持っている必要がある着信呼び出しの条件である可能性があります。
分類ポリシー
分類ポリシーを使用すると、ルール エンジンを利用して、キューを動的に選択し、ジョブの優先順位を決定し、ワーカー ラベル セレクターをジョブにアタッチできます。
例外ポリシー
例外ポリシーを使用すると、トリガーに基づいてジョブの動作を制御し、目的のアクションを実行することができます。 例外ポリシーはキューにアタッチされているので、キュー内のジョブの動作を制御することができます。
例
クライアントの初期化
SMS クライアントを初期化するために、接続文字列を使用してインスタンス化できます。 または、DefaultAzureCredential を使用して Active Directory 認証を使用することもできます。
from azure.communication.jobrouter import (
JobRouterClient,
JobRouterAdministrationClient
)
connection_string = "endpoint=ENDPOINT;accessKey=KEY"
router_client = JobRouterClient.from_connection_string(conn_str = connection_string)
router_admin_client = JobRouterAdministrationClient.from_connection_string(conn_str = connection_string)
配布ポリシー
キューを作成する前に、配布ポリシーが必要です。
from azure.communication.jobrouter.models import (
LongestIdleMode,
DistributionPolicy
)
distribution_policy: DistributionPolicy = DistributionPolicy(
offer_expires_after_seconds = 24 * 60 * 60,
mode = LongestIdleMode(
min_concurrent_offers = 1,
max_concurrent_offers = 1
)
)
distribution_policy: DistributionPolicy = router_admin_client.upsert_distribution_policy(
"distribution-policy-1",
distribution_policy
)
キュー
次に、キューを作成できます。
from azure.communication.jobrouter.models import (
RouterQueue
)
queue: RouterQueue = RouterQueue(
distribution_policy_id = "distribution-policy-1"
)
queue: RouterQueue = router_admin_client.upsert_queue(
"queue-1",
queue
)
ジョブ
これで、ジョブをそのキューに直接送信できます。ワーカー セレクターを使用する場合、ワーカーのラベル Some-Skill が 10 より大きい必要があります。
from azure.communication.jobrouter.models import (
RouterJob,
RouterWorkerSelector,
LabelOperator
)
router_job: RouterJob = RouterJob(
channel_id = "my-channel",
queue_id = "queue-1",
channel_reference = "12345",
priority = 1,
requested_worker_selectors = [
RouterWorkerSelector(key = "Some-Skill", label_operator = LabelOperator.EQUAL, value = 10)
]
)
job: RouterJob = router_client.upsert_job(
"jobId-1",
router_job
)
ワーカー
次に、そのキューから作業を受け取るワーカーを登録します。ラベルは Some-Skill 11 です。
from azure.communication.jobrouter.models import (
RouterWorker,
RouterChannel
)
router_worker: RouterWorker = RouterWorker(
capacity = 1,
queues = [
"queue-1"
],
labels = {
"Some-Skill": 11
},
channels = [
RouterChannel(channel_id = "my-channel", capacity_cost_per_job = 1)
],
available_for_offers = True
)
worker = router_client.upsert_worker(
"worker-1",
router_worker
)
プラン
EventGrid subscription から RouterWorkerOfferIssued を受け取る必要があります。
イベント ハンドラーとして機能するいくつかの異なる Azure サービスがあります。 このシナリオでは、イベント配信用の Webhook を想定します。 Webhook イベント配信の詳細を確認する
イベントがイベント ハンドラーに配信されたら、JSON ペイロードをイベントの一覧に逆シリアル化できます。
# Parse the JSON payload into a list of events
from azure.eventgrid import EventGridEvent
import json
## deserialize payload into a list of typed Events
events = [EventGridEvent.from_json(json.loads(msg)) for msg in payload]
offer_id = ""
for event in events:
if event.event_type == "Microsoft.Communication.RouterWorkerOfferIssued":
offer_id = event.data.offer_id
else:
continue
ただし、数秒待ってから、JobRouter API に対してワーカーに直接クエリを実行して、オファーが発行されたかどうかを確認することもできます。
from azure.communication.jobrouter.models import (
RouterWorker,
)
router_worker: RouterWorker = router_client.get_worker(worker_id = "worker-1")
for offer in router_worker.offers:
print(f"Worker {router_worker.id} has an active offer for job {offer.job_id}")
オファーを受け入れる
ワーカーがオファーを受け取ると、受け入れまたは拒否という 2 つの可能なアクションを実行できます。 私たちはオファーを受け入れるつもりです。
from azure.communication.jobrouter.models import (
RouterJobOffer,
AcceptJobOfferResult,
RouterJobStatus
)
# fetching the offer id
job_offer: RouterJobOffer = [offer for offer in router_worker.offers if offer.job_id == "jobId-1"][0]
offer_id = job_offer.offer_id
# accepting the offer sent to `worker-1`
accept_job_offer_result: AcceptJobOfferResult = router_client.accept_job_offer(
worker_id = "worker-1",
offer_id = offer_id
)
print(f"Offer: {job_offer.offer_id} sent to worker: {router_worker.id} has been accepted")
print(f"Job has been assigned to worker: {router_worker.id} with assignment: {accept_job_offer_result.assignment_id}")
# verify job assignment is populated when querying job
updated_job = router_client.get_job(job_id = "jobId-1")
print(f"Job assignment has been successful: {updated_job.job_status == RouterJobStatus.Assigned and accept_job_offer_result.assignment_id in updated_job.assignments}")
ジョブの完了
ワーカーがジョブで完了したら、ワーカーはジョブを として completedマークする必要があります。
import datetime
from azure.communication.jobrouter.models import (
CompleteJobOptions
)
complete_job_result = router_client.complete_job(
"jobId-1",
accept_job_offer_result.assignment_id,
CompleteJobOptions(
note = f"Job has been completed by {router_worker.id} at {datetime.datetime.utcnow()}"
)
)
print(f"Job has been successfully completed.")
ジョブを閉じる
ジョブが完了した後、ワーカーはジョブを閉じる前にジョブに対してラップ アップ アクションを実行し、最終的に、より多くの受信ジョブを受け入れるための容量を解放できます
from azure.communication.jobrouter.models import (
RouterJob,
RouterJobStatus,
CloseJobOptions,
)
close_job_result = router_client.close_job(
"jobId-1",
accept_job_offer_result.assignment_id,
CloseJobOptions(
note = f"Job has been closed by {router_worker.id} at {datetime.datetime.utcnow()}"
)
)
print(f"Job has been successfully closed.")
update_job: RouterJob = router_client.get_job(job_id = "jobId-1")
print(f"Updated job status: {update_job.job_status == RouterJobStatus.CLOSED}")
import time
from datetime import datetime, timedelta
from azure.communication.jobrouter.models import (
RouterJob,
RouterJobStatus,
CloseJobOptions,
)
close_job_in_future_result = router_client.close_job(
"jobId-1",
accept_job_offer_result.assignment_id,
CloseJobOptions(
note = f"Job has been closed by {router_worker.id} at {datetime.utcnow()}",
close_at = datetime.utcnow() + timedelta(seconds = 2)
)
)
print(f"Job has been marked to close")
time.sleep(secs = 2)
update_job: RouterJob = router_client.get_job(job_id = "jobId-1")
print(f"Updated job status: {update_job.job_status == RouterJobStatus.CLOSED}")
トラブルシューティング
問題が発生していますか? このセクションには、そこで何を行うかの詳細が含まれている必要があります。
次のステップ
その他のサンプル コード
このライブラリの使用方法の詳細な例については、 samples ディレクトリを参照してください。
フィードバックの提供
バグが発生した場合、または提案がある場合は、プロジェクトの [問題 ] セクションに問題を提出してください
共同作成
このプロジェクトでは、共同作成と提案を歓迎しています。 ほとんどの共同作成では、共同作成者使用許諾契約書 (CLA) にご同意いただき、ご自身の共同作成内容を使用する権利を Microsoft に供与する権利をお持ちであり、かつ実際に供与することを宣言していただく必要があります。 詳細については、「 cla.microsoft.com」を参照してください。
このプロジェクトでは、Microsoft オープン ソースの倫理規定を採用しています。 詳しくは、「Code of Conduct FAQ (倫理規定についてよくある質問)」を参照するか、opencode@microsoft.com 宛てに質問またはコメントをお送りください。
Azure SDK for Python